Fix overlapping MutationObserver registrations, with credit to #68

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,65 +8317,71 @@ 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)
         self._records.clear()
         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:

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

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