From e9c851d89e12ee1de449ad9d8a4e4dfa0c37f8e0 Mon Sep 17 00:00:00 2001 From: byteface Date: Wed, 9 Sep 2026 21:16:34 +0100 Subject: [PATCH 1/2] fix: preserve mutation old values across overlapping observations Adapt the interested-observer aggregation from #68 so matching ancestor registrations can request old values without producing duplicate records. Credit 7HR4IZ3 beside the implementation and in the regression tests. Validate contradictory observer options, infer omitted mutation types from option presence, and release inactive observers from the global registry. Document the existing synchronous callback behavior. Validation: 248 passed, 89 subtests passed across the observer, DOM and insertion ownership tests. The 22 new cases produce 11 failures against the unchanged baseline and all pass with this change. Co-authored-by: 7HR4IZ3 <90985774+7HR4IZ3@users.noreply.github.com> --- domonic/dom.py | 82 +++++++++------- tests/test_dom_mutation_observer.py | 141 ++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 33 deletions(-) create mode 100644 tests/test_dom_mutation_observer.py diff --git a/domonic/dom.py b/domonic/dom.py index 8e738fb..728be50 100644 --- a/domonic/dom.py +++ b/domonic/dom.py @@ -704,10 +704,18 @@ def _normalize_mutation_observer_options(options: dict[str, Any]) -> dict[str, A "characterData": bool(options.get("characterData", False)), "characterDataOldValue": bool(options.get("characterDataOldValue", False)), } - if normalized["attributeFilter"] is not None or normalized["attributeOldValue"]: + if "attributes" not in options and ( + "attributeFilter" in options or "attributeOldValue" in options + ): normalized["attributes"] = True - if normalized["characterDataOldValue"]: + if "characterData" not in options and "characterDataOldValue" in options: normalized["characterData"] = True + if not normalized["attributes"] and ( + normalized["attributeOldValue"] or "attributeFilter" in options + ): + raise TypeError("Attribute options require attributes to be enabled") + if normalized["characterDataOldValue"] and not normalized["characterData"]: + raise TypeError("characterDataOldValue requires characterData to be enabled") if normalized["attributeFilter"] is not None: normalized["attributeFilter"] = tuple( attr[1:] if isinstance(attr, str) and attr.startswith("_") else attr @@ -8296,7 +8304,9 @@ class MutationObserver: This implementation follows the familiar platform model: call ``observe()`` with a target and options, allow DOM operations to queue records, then - receive them through the callback or ``takeRecords()``. + receive them through the callback or ``takeRecords()``. Callbacks currently + run synchronously after each mutation; browser microtask batching is not + implemented. Call ``disconnect()`` to release an active registration. """ _all_observers: ClassVar[list["MutationObserver"]] = [] @@ -8307,16 +8317,19 @@ def __init__(self, callback: MutationCallback) -> None: self.callback = callback self._records: list[MutationRecord] = [] self._observations: dict[Node, dict[str, Any]] = {} - MutationObserver._all_observers.append(self) def disconnect(self) -> None: self._observations.clear() self._records.clear() + if self in MutationObserver._all_observers: + MutationObserver._all_observers.remove(self) def observe(self, target: Node, options: dict[str, Any]) -> None: if not isinstance(target, Node): raise TypeError("MutationObserver target must be a Node") self._observations[target] = _normalize_mutation_observer_options(options) + if self not in MutationObserver._all_observers: + MutationObserver._all_observers.append(self) def takeRecords(self) -> list[MutationRecord]: records = list(self._records) @@ -8324,48 +8337,51 @@ def takeRecords(self) -> list[MutationRecord]: return records def _enqueue_if_observing(self, record: MutationRecord) -> bool: + # Adapted from 7HR4IZ3's interestedObservers algorithm in PR #68: + # https://github.com/byteface/domonic/pull/68 + # Combine every matching registration before queuing one record. A direct + # observation must not hide an ancestor's request for the old value. + # See test_overlapping_observations_preserve_old_value in + # tests/test_dom_mutation_observer.py. + interested = False + old_value = None for current in _iter_ancestors_inclusive(record.target): options = self._observations.get(current) if options is None: continue if current is not record.target and not options["subtree"]: continue - if record.type == "childList" and not options["childList"]: + if not options.get(record.type, False): continue if record.type == "attributes": - if not options["attributes"]: - continue - attribute_filter = options.get("attributeFilter") + attribute_filter = options["attributeFilter"] if ( attribute_filter is not None and record.attributeName not in attribute_filter ): continue - old_value = record.oldValue if options["attributeOldValue"] else None - filtered_record = MutationRecord( - "attributes", - record.target, - attributeName=record.attributeName, - attributeNamespace=record.attributeNamespace, - oldValue=old_value, - ) - self._records.append(filtered_record) - return True - if record.type == "characterData": - if not options["characterData"]: - continue - old_value = ( - record.oldValue if options["characterDataOldValue"] else None - ) - filtered_record = MutationRecord( - "characterData", record.target, oldValue=old_value - ) - self._records.append(filtered_record) - return True - if record.type == "childList": - self._records.append(record) - return True - return False + if options["attributeOldValue"]: + old_value = record.oldValue + elif record.type == "characterData": + if options["characterDataOldValue"]: + old_value = record.oldValue + interested = True + if not interested: + return False + self._records.append( + MutationRecord( + record.type, + record.target, + addedNodes=record.addedNodes, + removedNodes=record.removedNodes, + previousSibling=record.previousSibling, + nextSibling=record.nextSibling, + attributeName=record.attributeName, + attributeNamespace=record.attributeNamespace, + oldValue=old_value, + ) + ) + return True def _flush(self) -> None: if not self._records: diff --git a/tests/test_dom_mutation_observer.py b/tests/test_dom_mutation_observer.py new file mode 100644 index 0000000..241c621 --- /dev/null +++ b/tests/test_dom_mutation_observer.py @@ -0,0 +1,141 @@ +"""Observer regressions, including overlapping registrations salvaged from PR #68. + +7HR4IZ3's interestedObservers algorithm combines old-value requests across +registrations: https://github.com/byteface/domonic/pull/68. +""" + +import gc +import weakref + +import pytest + +from domonic.dom import MutationObserver, Text +from domonic.html import div, span + + +@pytest.fixture +def observer(): + records = [] + instance = MutationObserver(lambda batch, obs: records.extend(batch)) + yield instance, records + instance.disconnect() + + +@pytest.mark.parametrize("kind", ["attributes", "characterData"]) +@pytest.mark.parametrize("ancestor_first", [True, False]) +@pytest.mark.parametrize("old_value_on_ancestor", [True, False]) +def test_overlapping_observations_preserve_old_value( + observer, kind, ancestor_first, old_value_on_ancestor +): + """PR #68: one record, with oldValue from either matching registration.""" + instance, records = observer + child = span(_id="before") if kind == "attributes" else Text("before") + parent = div(child) + old_option = ( + "attributeOldValue" if kind == "attributes" else "characterDataOldValue" + ) + registrations = [ + (parent, {kind: True, "subtree": True, old_option: old_value_on_ancestor}), + (child, {kind: True, old_option: not old_value_on_ancestor}), + ] + for target, options in registrations if ancestor_first else reversed(registrations): + instance.observe(target, options) + if kind == "attributes": + child.setAttribute("id", "after") + else: + child.data = "after" + assert len(records) == 1 + assert records[0].target is child + assert records[0].type == kind + assert records[0].oldValue == "before" + + +@pytest.mark.parametrize( + "ancestor_options", + [ + {"attributes": True, "attributeOldValue": True}, + { + "attributes": True, + "subtree": True, + "attributeOldValue": True, + "attributeFilter": ["class"], + }, + ], +) +def test_nonmatching_ancestor_does_not_expose_old_value(observer, ancestor_options): + instance, records = observer + child = span(_id="before") + parent = div(child) + instance.observe(parent, ancestor_options) + instance.observe(child, {"attributes": True}) + child.setAttribute("id", "after") + assert len(records) == 1 + assert records[0].oldValue is None + + +@pytest.mark.parametrize( + "options", + [ + {}, + {"childList": True, "attributes": False, "attributeOldValue": True}, + {"childList": True, "attributes": False, "attributeFilter": []}, + {"childList": True, "characterData": False, "characterDataOldValue": True}, + ], +) +def test_invalid_options_preserve_existing_registration(observer, options): + instance, records = observer + target = div() + instance.observe(target, {"attributes": True}) + with pytest.raises(TypeError): + instance.observe(target, options) + target.setAttribute("id", "still-observed") + assert len(records) == 1 + + +@pytest.mark.parametrize( + "options", + [ + {"attributeOldValue": False}, + {"attributeOldValue": True}, + {"attributeFilter": ["id"]}, + {"characterDataOldValue": False}, + {"characterDataOldValue": True}, + ], +) +def test_present_options_enable_omitted_mutation_type(observer, options): + instance, records = observer + character_data = "characterDataOldValue" in options + target = Text("before") if character_data else div(_id="before") + instance.observe(target, options) + if character_data: + target.data = "after" + else: + target.setAttribute("id", "after") + assert len(records) == 1 + wants_old = options.get("attributeOldValue") or options.get("characterDataOldValue") + assert records[0].oldValue == ("before" if wants_old else None) + + +def test_disconnect_releases_observer_and_allows_reuse(observer): + instance, records = observer + target = div() + instance.observe(target, {"attributes": True}) + instance.disconnect() + target.setAttribute("id", "ignored") + assert records == [] + instance.observe(target, {"attributes": True}) + instance.observe(target, {"attributes": True}) + target.setAttribute("id", "observed") + assert len(records) == 1 + + +@pytest.mark.parametrize("observe_first", [False, True]) +def test_inactive_observers_are_collectable(observe_first): + instance = MutationObserver(lambda records, obs: None) + if observe_first: + instance.observe(div(), {"childList": True}) + instance.disconnect() + reference = weakref.ref(instance) + del instance + gc.collect() + assert reference() is None From b6da31dbde1fb556a2a0560b8f0ea1a258c73d0b Mon Sep 17 00:00:00 2001 From: byteface Date: Wed, 9 Sep 2026 21:27:22 +0100 Subject: [PATCH 2/2] test: support detached HEAD in terminal Git checks --- tests/test_terminal.py | 24 ++++++++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/tests/test_terminal.py b/tests/test_terminal.py index b899f2f..568866f 100644 --- a/tests/test_terminal.py +++ b/tests/test_terminal.py @@ -98,8 +98,28 @@ def test_bash_cp(self): @silence def test_bash_git(self): - # print(git('status')) - self.assertIn("master", git("status")) + # Exercise Git in an isolated repository, including the detached HEAD + # used by PR builds. The project's checkout need not be on master. + with TemporaryDirectory() as tmp: + git("init", cwd=tmp) + git("symbolic-ref HEAD refs/heads/terminal-test", cwd=tmp) + git( + "-c user.name=Test -c user.email=test@example.invalid " + "-c commit.gpgsign=false commit --allow-empty -m initial", + cwd=tmp, + ) + for detached in (False, True): + with self.subTest(detached=detached): + if detached: + git("checkout --detach HEAD", cwd=tmp) + self.assertEqual(str(git("status --porcelain", cwd=tmp)), "") + with open(os.path.join(tmp, "untracked.txt"), "w") as handle: + handle.write("test") + self.assertEqual( + str(git("status --porcelain", cwd=tmp)).strip(), + "?? untracked.txt", + ) + os.remove(os.path.join(tmp, "untracked.txt")) def test_bash_general(self): self.assertIn("LS", man("ls").upper())