Make ScopeProxy a MutableMapping

flowrep/models/parsers/object_scope.py

@@ -4,25 +4,58 @@
 import builtins
 import inspect
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, MutableMapping
 from typing import Any
 
 
-class ScopeProxy:
+class _EmptyValue: ...
+
+
+class ScopeProxy(MutableMapping[str, object]):
     """
-    Make the __dict__-like scope dot-accessible without duplicating the dictionary
-    like types.SimpleNamespace would.
+    A mutable mapping to connect symbols to python objects.
+
+    By default, does not allow re-registration of existing symbols to new values.
     """
 
-    def __init__(self, d: dict):
-        self._d = d
+    def __init__(self, d: MutableMapping[str, object], allow_overwrite: bool = False):
+        self._d = {k: v for k, v in d.items()}
+        self.allow_overwrite = allow_overwrite
+
+    def __getitem__(self, name: str):
+        return self._d[name]
+
+    def __setitem__(self, name: str, value: object):
+        if not self.allow_overwrite:
+            old_value = self._d.get(name, _EmptyValue)
+            if old_value is not _EmptyValue and value is not old_value:
+                raise ValueError(
+                    f"Variable {name} already exists as {old_value!r} in this "
+                    f"scope. It cannot be reassigned to a new value of {value!r} "
+                    f"while allow_overwrite is False."
+                )
+            self._d[name] = value
+        else:
+            self._d[name] = value
+
+    def __delitem__(self, name: str):
+        self._d.__delitem__(name)
+
+    def __iter__(self):
+        return iter(self._d)
+
+    def __len__(self):
+        return len(self._d)
 
     def __getattr__(self, name: str):
         try:
-            return self._d[name]
+            return self.__getitem__(name)
         except KeyError:
             raise AttributeError(name) from None
 
+    def __str__(self):
+        return str(self._d)
+
     def register(self, name: str, obj: object) -> None:
         """
         Add a name → object binding to this scope.
@@ -32,7 +65,7 @@ def register(self, name: str, obj: object) -> None:
         resolutions within this scope (or any scope that shares the same
         backing namespace).
         """
-        self._d[name] = obj
+        self[name] = obj
 
     def fork(self) -> ScopeProxy:
         """
@@ -42,7 +75,7 @@ def fork(self) -> ScopeProxy:
         affect the parent.  Used when walking conditional branches so that
         branch-local imports don't leak into sibling branches.
         """
-        return ScopeProxy(dict(self._d))
+        return ScopeProxy(self, allow_overwrite=self.allow_overwrite)
 
 
 def get_scope(func: Callable[..., Any] | type[Any]) -> ScopeProxy:

tests/unit/models/parsers/test_object_scope.py

@@ -33,6 +33,96 @@ def test_missing_key_raises_attribute_error(self):
         with self.assertRaises(AttributeError):
             _ = proxy.nonexistent
 
+    def test_getitem(self):
+        proxy = object_scope.ScopeProxy({"x": 42})
+        self.assertEqual(proxy["x"], 42)
+
+    def test_getitem_missing_raises_key_error(self):
+        proxy = object_scope.ScopeProxy({})
+        with self.assertRaises(KeyError):
+            _ = proxy["missing"]
+
+    def test_setitem_new_key(self):
+        proxy = object_scope.ScopeProxy({})
+        proxy["a"] = 1
+        self.assertEqual(proxy["a"], 1)
+
+    def test_setitem_same_object_allowed(self):
+        """Re-assigning the same object is fine even without allow_overwrite."""
+        sentinel = object()
+        proxy = object_scope.ScopeProxy({"s": sentinel})
+        proxy["s"] = sentinel  # should not raise
+        self.assertIs(proxy["s"], sentinel)
+
+    def test_setitem_different_value_raises(self):
+        proxy = object_scope.ScopeProxy({"x": 1})
+        with self.assertRaises(ValueError, msg="allow_overwrite is False"):
+            proxy["x"] = 2
+
+    def test_setitem_allow_overwrite(self):
+        proxy = object_scope.ScopeProxy({"x": 1}, allow_overwrite=True)
+        proxy["x"] = 2
+        self.assertEqual(proxy["x"], 2)
+
+    def test_delitem(self):
+        proxy = object_scope.ScopeProxy({"a": 1, "b": 2})
+        del proxy["a"]
+        self.assertNotIn("a", proxy)
+        self.assertIn("b", proxy)
+
+    def test_delitem_missing_raises_key_error(self):
+        proxy = object_scope.ScopeProxy({})
+        with self.assertRaises(KeyError):
+            del proxy["missing"]
+
+    def test_iter(self):
+        d = {"a": 1, "b": 2, "c": 3}
+        proxy = object_scope.ScopeProxy(d)
+        self.assertEqual(set(proxy), set(d))
+
+    def test_len(self):
+        proxy = object_scope.ScopeProxy({"a": 1, "b": 2})
+        self.assertEqual(len(proxy), 2)
+
+    def test_str(self):
+        proxy = object_scope.ScopeProxy({"x": 1})
+        self.assertEqual(str(proxy), "{'x': 1}")
+
+    def test_mutations_do_not_propagate_to_source(self):
+        d = {"a": 1}
+        proxy = object_scope.ScopeProxy(d)
+        proxy["b"] = 2
+        self.assertNotIn("b", d)
+
+    def test_register(self):
+        proxy = object_scope.ScopeProxy({})
+        proxy.register("math", 42)
+        self.assertEqual(proxy["math"], 42)
+
+    def test_register_respects_overwrite_guard(self):
+        proxy = object_scope.ScopeProxy({"x": 1})
+        with self.assertRaises(ValueError):
+            proxy.register("x", 2)
+
+    def test_fork_returns_independent_copy(self):
+        proxy = object_scope.ScopeProxy({"a": 1})
+        forked = proxy.fork()
+        forked["b"] = 2
+        self.assertIn("b", forked)
+        self.assertNotIn("b", proxy)
+
+    def test_fork_preserves_existing_data(self):
+        proxy = object_scope.ScopeProxy({"a": 1})
+        forked = proxy.fork()
+        self.assertEqual(forked["a"], 1)
+
+    def test_fork_preserves_allow_overwrite(self):
+        proxy = object_scope.ScopeProxy({}, allow_overwrite=True)
+        forked = proxy.fork()
+        forked["x"] = 1
+        forked["x"] = 2  # should not raise
+        self.assertEqual(forked["x"], 2)
+
 
 class TestGetScope(unittest.TestCase):
     def test_returns_module_globals(self):