233 lines
7.9 KiB
Python
233 lines
7.9 KiB
Python
from dataclasses import dataclass
|
|
from dataclasses import field
|
|
from typing import Any
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from hooked_containers import events as e
|
|
from hooked_containers.mapping import HookedMapping
|
|
from hooked_containers.mixins import ContainerMixin
|
|
from hooked_containers.mixins import ListMixin
|
|
from hooked_containers.mixins import MutableMappingMixin
|
|
from hooked_containers.mixins import MutableSetMixin
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mixin", "storage", "method", "arguments", "expected"),
|
|
[
|
|
(ListMixin, [1], "append", (2,), e.Change("add", 1, 2)),
|
|
(MutableMappingMixin, {"a": 1}, "update", ({"b": 2},), e.Change("add", "b", 2)),
|
|
(MutableSetMixin, {1}, "add", (2,), e.Change("add", None, 2)),
|
|
],
|
|
)
|
|
def test_direct_mixin_composition(mixin, storage, method, arguments, expected):
|
|
custom_type = type("CustomContainer", (mixin,), {})
|
|
container = custom_type()
|
|
changes = []
|
|
container._storage = storage.copy()
|
|
container.hook = changes.append
|
|
|
|
assert isinstance(container, ContainerMixin)
|
|
assert len(container) == 1
|
|
assert list(container) == list(storage)
|
|
assert next(iter(storage)) in container
|
|
assert repr(container) == f"CustomContainer({storage!r})"
|
|
assert changes == []
|
|
|
|
getattr(container, method)(*arguments)
|
|
assert changes == [expected]
|
|
assert len(container) == 2
|
|
assert len(storage) == 1
|
|
|
|
|
|
@dataclass
|
|
class HookTracker:
|
|
added: list[object] = field(default_factory=list)
|
|
set: list[object] = field(default_factory=list)
|
|
removed: list[object] = field(default_factory=list)
|
|
|
|
def hook(self, event: e.Change):
|
|
match event.operation:
|
|
case "add":
|
|
self.added.append(event.value)
|
|
case "set":
|
|
self.set.append(event.value)
|
|
case "remove":
|
|
self.removed.append(event.value)
|
|
|
|
|
|
class TestHookedMapping:
|
|
class TestConstruction:
|
|
def test_empty(self):
|
|
m = HookedMapping({})
|
|
assert m._storage == {}
|
|
assert dict(m) == {}
|
|
|
|
def test_with_existing(self):
|
|
existing = {"a": 1, "b": 2}
|
|
m = HookedMapping(existing)
|
|
assert m._storage is not existing
|
|
assert dict(m) == existing
|
|
|
|
def test_nesting(self):
|
|
existing = {"a": {"x": 1}, "b": {"y": 2}}
|
|
m = HookedMapping(existing)
|
|
assert dict(m) == existing
|
|
assert dict(m["a"]) == {"x": 1}
|
|
assert dict(m["b"]) == {"y": 2}
|
|
|
|
def test_recreation_copies_storage_without_duplicate_hooks(self):
|
|
original_tracker = HookTracker()
|
|
tracker = HookTracker()
|
|
original = HookedMapping({"a": 1}, original_tracker.hook)
|
|
m = HookedMapping(original, tracker.hook)
|
|
|
|
assert m._storage is not original._storage
|
|
m["a"] = 2
|
|
assert original._storage == {"a": 1}
|
|
assert tracker.set == [2]
|
|
assert original_tracker.set == []
|
|
|
|
class TestMappingOps:
|
|
def test_setitem(self):
|
|
tracker = HookTracker()
|
|
m = HookedMapping({}, tracker.hook)
|
|
m["a"] = 1
|
|
assert m._storage == {"a": 1}
|
|
assert tracker.added == [1]
|
|
|
|
def test_nested_values_are_not_observed(self):
|
|
tracker = HookTracker()
|
|
m = HookedMapping({"a": {}}, tracker.hook)
|
|
m["a"]["x"] = 1
|
|
assert m._storage == {"a": {"x": 1}}
|
|
assert tracker.added == []
|
|
|
|
def test_getitem(self):
|
|
m = HookedMapping({"a": 1})
|
|
assert m["a"] == 1
|
|
|
|
def test_delitem(self):
|
|
tracker = HookTracker()
|
|
m = HookedMapping({"a": 1}, tracker.hook)
|
|
del m["a"]
|
|
with pytest.raises(KeyError):
|
|
m["a"]
|
|
assert tracker.removed == [1]
|
|
|
|
def test_inherited_mutation_helpers(self):
|
|
tracker = HookTracker()
|
|
m = HookedMapping({"a": 1}, tracker.hook)
|
|
|
|
m.update({"a": 2, "b": 3})
|
|
assert tracker.set == [2]
|
|
assert tracker.added == [3]
|
|
assert m.pop("b") == 3
|
|
assert tracker.removed == [3]
|
|
assert m._storage == {"a": 2}
|
|
|
|
def test_assignment_returns_none_and_reports_key(self):
|
|
changes = []
|
|
m = HookedMapping({"a": 1}, changes.append)
|
|
|
|
assert m.__setitem__("a", 2) is None
|
|
assert m.__setitem__("b", 3) is None
|
|
assert changes == [e.Change("set", "a", 2), e.Change("add", "b", 3)]
|
|
|
|
def test_assign_hooked_mapping(self):
|
|
changes = []
|
|
m = HookedMapping({}, changes.append)
|
|
value = HookedMapping({"x": 1})
|
|
|
|
assert m.__setitem__("a", value) is None
|
|
assert m["a"] is value
|
|
assert len(changes) == 1
|
|
assert changes[0].operation == "add"
|
|
assert changes[0].value is value
|
|
|
|
def test_shallow_value_identity(self):
|
|
changes = []
|
|
m = HookedMapping({"a": {"x": 1}}, changes.append)
|
|
|
|
m["a"]["x"] = 2
|
|
del m["a"]["x"]
|
|
assert changes == []
|
|
value = m["a"]
|
|
del m["a"]
|
|
assert changes == [e.Change("remove", "a", {})]
|
|
assert changes[0].value is value
|
|
|
|
def test_reads_views_and_missing_keys_do_not_mutate(self):
|
|
changes = []
|
|
mapping = HookedMapping({"a": 1}, changes.append)
|
|
keys = mapping.keys()
|
|
assert mapping.get("missing", 9) == 9
|
|
assert mapping.pop("missing", 9) == 9
|
|
with pytest.raises(KeyError):
|
|
mapping["missing"]
|
|
with pytest.raises(KeyError):
|
|
del mapping["missing"]
|
|
assert changes == []
|
|
assert list(mapping.items()) == [("a", 1)]
|
|
mapping["b"] = 2
|
|
assert list(keys) == ["a", "b"]
|
|
assert list(mapping.values()) == [1, 2]
|
|
assert list(reversed(mapping)) == ["b", "a"]
|
|
|
|
def test_native_order_copy_and_merge(self):
|
|
changes = []
|
|
mapping = HookedMapping([("a", 1), ("b", 2)], changes.append, c=3)
|
|
assert mapping == {"a": 1, "b": 2, "c": 3}
|
|
assert mapping.popitem() == ("c", 3)
|
|
assert changes == [e.Change("remove", "c", 3)]
|
|
changes.clear()
|
|
assert type(mapping.copy()) is dict
|
|
assert mapping | {"a": 9} == {"a": 9, "b": 2}
|
|
assert {"a": 9} | mapping == {"a": 1, "b": 2}
|
|
assert changes == []
|
|
original = mapping
|
|
mapping |= [("b", 4), ("c", 5)]
|
|
assert mapping is original
|
|
assert changes == [e.Change("set", "b", 4), e.Change("add", "c", 5)]
|
|
|
|
def test_update_duck_mapping_and_partial_failure(self):
|
|
class KeysOnly:
|
|
def keys(self):
|
|
return ["a"]
|
|
|
|
def __getitem__(self, _key):
|
|
return 1
|
|
|
|
changes = []
|
|
mapping = HookedMapping(hook=changes.append)
|
|
mapping.update(KeysOnly())
|
|
with pytest.raises(ValueError):
|
|
mapping.update(cast(Any, [("b", 2), ("bad",)]))
|
|
assert mapping == {"a": 1, "b": 2}
|
|
assert changes == [e.Change("add", "a", 1), e.Change("add", "b", 2)]
|
|
|
|
def test_clear_and_setdefault(self):
|
|
changes = []
|
|
mapping = HookedMapping[str | None, int]({None: 1}, changes.append)
|
|
assert mapping.setdefault(None, 9) == 1
|
|
assert changes == []
|
|
assert mapping.setdefault("a", 2) == 2
|
|
assert changes == [e.Change("add", "a", 2)]
|
|
changes.clear()
|
|
mapping.clear()
|
|
assert mapping == {}
|
|
assert changes == [e.Change("remove", "a", 2), e.Change("remove", None, 1)]
|
|
changes.clear()
|
|
mapping.clear()
|
|
assert changes == []
|
|
|
|
def test_callback_error_leaves_write_in_place(self):
|
|
def fail(_change):
|
|
raise RuntimeError("callback failed")
|
|
|
|
mapping = HookedMapping(hook=fail)
|
|
with pytest.raises(RuntimeError, match="callback failed"):
|
|
mapping["a"] = 1
|
|
assert mapping == {"a": 1}
|