testpoint removal support (#91)

tests/testpoint/test_fixture.py

@@ -83,3 +83,38 @@ async def test_not_handled(
     )
     assert response.status_code == 200
     assert response.json() == {'data': None, 'handled': False}
+
+
+def test_deletion_by_name(testpoint):
+    @testpoint('foo')
+    def foo_point(data): ...
+
+    assert 'foo' in testpoint
+    del testpoint['foo']
+
+    assert 'foo' not in testpoint
+
+    with pytest.raises(KeyError):
+        del testpoint['foo']
+
+
+def test_deletion_func(testpoint):
+    @testpoint('foo')
+    def foo_point(data): ...
+
+    del testpoint[foo_point]
+    assert 'foo' not in testpoint
+
+    with pytest.raises(KeyError):
+        del testpoint[foo_point]
+
+    @testpoint('foo')
+    @testpoint('bar')
+    def foo_point(data): ...
+
+    del testpoint[foo_point]
+    assert 'foo' not in testpoint
+    assert 'bar' not in testpoint
+
+    with pytest.raises(KeyError):
+        del testpoint[foo_point]

testsuite/plugins/testpoint.py

@@ -19,7 +19,7 @@
 ]
 
 
-class TestpointFixture(collections.abc.Mapping):
+class TestpointFixture(collections.abc.MutableMapping):
     """Testpoint control object."""
 
     def __init__(self, *, checker_factory) -> None:
@@ -29,6 +29,21 @@ def __init__(self, *, checker_factory) -> None:
     def __getitem__(self, name: str) -> callinfo.AsyncCallQueue:
         return self._handlers[name]
 
+    def __setitem__(self, key: str, value: callinfo.AsyncCallQueue):
+        self._handlers[key] = value
+
+    def __delitem__(self, key):
+        if isinstance(key, callinfo.AsyncCallQueue):
+            names = [
+                name for name, value in self._handlers.items() if value == key
+            ]
+            if not names:
+                raise KeyError(f'{key!r}')
+            for name in names:
+                del self._handlers[name]
+        else:
+            del self._handlers[key]
+
     def __len__(self):
         return len(self._handlers)
 
@@ -45,7 +60,7 @@ def __call__(self, name: str) -> TestpointDecorator:
 
         def decorator(func) -> callinfo.AsyncCallQueue:
             wrapped = callinfo.acallqueue(func, checker=checker)
-            self._handlers[name] = wrapped
+            self[name] = wrapped
             return wrapped
 
         return decorator