hooked-containers/tests/test_set.py

from dataclasses import FrozenInstanceError
from typing import Any
from typing import cast

import pytest

from hooked_containers import Change
from hooked_containers import HookedSet


class TestHookedSet:
    def test_construction_and_owned_storage(self):
        changes = []
        original = {1, 2}
        members = HookedSet(original, changes.append)
        assert members == original
        assert members._storage is not original
        assert HookedSet([1, 1, 2]) == {1, 2}
        assert HookedSet() == set()
        recreated = HookedSet(members)
        assert recreated._storage is not members._storage
        assert changes == []
        members.add(3)
        assert original == {1, 2}
        assert recreated == {1, 2}

    def test_add_discard_and_no_ops(self):
        changes = []
        members = HookedSet[str](hook=changes.append)
        members.add("a")
        members.add("a")
        members.discard("missing")
        members.discard("a")
        assert changes == [Change("add", None, "a"), Change("remove", None, "a")]
        changes.clear()
        members.clear()
        members.update([])
        members.difference_update([])
        assert changes == []

    def test_remove_pop_and_clear(self):
        changes = []
        members = HookedSet([1, 2], changes.append)
        members.remove(1)
        assert members.pop() == 2
        assert changes == [Change("remove", None, 1), Change("remove", None, 2)]
        changes.clear()
        with pytest.raises(KeyError):
            members.remove(3)
        with pytest.raises(KeyError):
            members.pop()
        assert changes == []
        members.update([3, 4])
        changes.clear()
        members.clear()
        assert members == set()
        assert {change.value for change in changes} == {3, 4}
        assert all(change.operation == "remove" and change.key is None for change in changes)

    def test_removal_reports_stored_value(self):
        changes = []
        members = HookedSet([1], changes.append)
        members.discard(True)
        assert changes == [Change("remove", None, 1)]
        assert type(changes[0].value) is int

    def test_hashability_is_validated_before_no_op_detection(self):
        changes = []
        members = HookedSet([frozenset({1})], changes.append)
        with pytest.raises(TypeError):
            members.add(cast(Any, {1}))
        assert members == {frozenset({1})}
        assert changes == []
        members.discard(cast(Any, {1}))
        assert members == set()
        assert changes == [Change("remove", None, frozenset({1}))]

    def test_algebra_and_named_helpers_do_not_notify(self):
        changes = []
        members = HookedSet([1, 2], changes.append)
        assert members == {1, 2}
        assert members <= {1, 2, 3}
        assert members < {1, 2, 3}
        assert members >= {1}
        assert members > {1}
        assert members != {1}
        assert members.isdisjoint([3, 4])
        assert members.issubset([1, 2, 3])
        assert members.issuperset([1])
        results = [
            (members | {2, 3}, {1, 2, 3}),
            (members & {2, 3}, {2}),
            (members - {2, 3}, {1}),
            (members ^ {2, 3}, {1, 3}),
            ({3} | members, {1, 2, 3}),
            ({2, 3} & members, {2}),
            ({2, 3} - members, {3}),
            ({2, 3} ^ members, {1, 3}),
            (members.union([3], [4]), {1, 2, 3, 4}),
            (members.intersection([2, 3]), {2}),
            (members.difference([2], [3]), {1}),
            (members.symmetric_difference([2, 3]), {1, 3}),
            (members.copy(), {1, 2}),
        ]
        for result, expected in results:
            assert result == expected
            assert type(result) is set
            result.add(99)
        assert members == {1, 2}
        assert changes == []

    def test_abc_operators_accept_iterables(self):
        members = HookedSet([1, 2])
        assert members | cast(Any, [3]) == {1, 2, 3}
        assert members & cast(Any, [2]) == {2}
        members |= cast(Any, [3])
        assert members == {1, 2, 3}

    @pytest.mark.parametrize("operation", ["union", "intersection", "difference", "symmetric_difference"])
    def test_in_place_operations_keep_identity_and_notify(self, operation):
        changes = []
        members = HookedSet([1, 2], changes.append)
        original = members
        if operation == "union":
            members |= {2, 3}
            expected = {1, 2, 3}
            expected_changes = {("add", 3)}
        elif operation == "intersection":
            members &= {2, 3}
            expected = {2}
            expected_changes = {("remove", 1)}
        elif operation == "difference":
            members -= {2, 3}
            expected = {1}
            expected_changes = {("remove", 2)}
        else:
            members ^= {2, 3}
            expected = {1, 3}
            expected_changes = {("remove", 2), ("add", 3)}
        assert members is original
        assert members == expected
        assert {(change.operation, change.value) for change in changes} == expected_changes
        assert all(change.key is None for change in changes)

    def test_named_updates_and_self_aliases(self):
        members = HookedSet([1, 2])
        members.update([2, 3], [4])
        assert members == {1, 2, 3, 4}
        members.intersection_update([2, 3, 4], [3, 4])
        assert members == {3, 4}
        members.symmetric_difference_update([4, 5, 5])
        assert members == {3, 5}
        members.difference_update([3], [9])
        assert members == {5}
        members.update(members)
        members.intersection_update(members)
        assert members == {5}
        members.symmetric_difference_update(members)
        assert members == set()
        members.update([1, 2])
        members.difference_update(members)
        assert members == set()

    def test_partial_update_failure_is_observed(self):
        def values():
            yield 1
            raise RuntimeError("iteration failed")

        changes = []
        members = HookedSet[int](hook=changes.append)
        with pytest.raises(RuntimeError, match="iteration failed"):
            members.update(values())
        assert members == {1}
        assert changes == [Change("add", None, 1)]


def test_change_metadata_is_immutable_and_values_are_shallow():
    value = []
    change = Change("set", "key", value)
    with pytest.raises(FrozenInstanceError):
        cast(Any, change).key = "other"
    value.append(1)
    assert change.value is value
    assert change.value == [1]