From d568472738e7e34cbdec5c6f9ea3190bb6c1bb96 Mon Sep 17 00:00:00 2001 From: liamhuber Date: Thu, 19 Mar 2026 12:56:58 -0700 Subject: [PATCH] Make `ScopeProxy` a `MutableMapping` Signed-off-by: liamhuber --- flowrep/models/parsers/object_scope.py | 51 +++++++++-- .../unit/models/parsers/test_object_scope.py | 90 +++++++++++++++++++ 2 files changed, 132 insertions(+), 9 deletions(-) diff --git a/flowrep/models/parsers/object_scope.py b/flowrep/models/parsers/object_scope.py index ea28cb90..c8699cb5 100644 --- a/flowrep/models/parsers/object_scope.py +++ b/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: diff --git a/tests/unit/models/parsers/test_object_scope.py b/tests/unit/models/parsers/test_object_scope.py index f7a96cc7..87009276 100644 --- a/tests/unit/models/parsers/test_object_scope.py +++ b/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):