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 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())