Add API command to delete event by given id

changelog.d/270.added.md

@@ -0,0 +1 @@
+Add API command to delete an event

src/zino/api/legacy.py

@@ -17,6 +17,7 @@
 from zino import version
 from zino.api import auth
 from zino.api.notify import Zino1NotificationProtocol
+from zino.events import EventIndex
 from zino.scheduler import get_scheduler
 from zino.state import ZinoState, config
 from zino.statemodels import (
@@ -619,3 +620,21 @@
         that exceptions that go unhandled by a command responder is handled by the protocol engine.
         """
         1 / 0  # noqa
+
+    @requires_authentication
+    @Zino1ServerProtocol._translate_case_id_to_event
+    async def do_deleteevent(self, event: Event):
+        """Implements an DELETEEVENT command that did not exist in the Zino 1 protocol. This is just used for testing
+        the frontend.
+        """
+        events = self._state.events
+
+        index = EventIndex(event.router, event.subindex, type(event))
+
+        if index in events._events_by_index and event.id == events._events_by_index[index].id:
+            del events._events_by_index[index]
+        if index in events._closed_events_by_index and event.id == events._closed_events_by_index[index].id:
+            del events._closed_events_by_index[index]
+        del events.events[event.id]
+
+        return self._respond_ok(f"event {event.id} deleted")

tests/api/legacy_test.py

@@ -25,6 +25,8 @@
 )
 from zino.time import now
 
+DEVICE_NAME = "example-gw.example.org"
+
 
 class TestZino1BaseServerProtocol:
     def test_should_init_without_error(self):
@@ -892,6 +894,54 @@ async def test_when_authenticated_should_list_all_pm_ids(self, authenticated_pro
             assert re.search(pattern_string.format(id), response), f"Expected response to contain id {id}"
 
 
+class TestZino1TestProtocolDeleteEventCommand:
+    @pytest.mark.asyncio
+    async def test_should_delete_open_event(self):
+        protocol = ZinoTestProtocol()
+        fake_transport = Mock()
+        protocol.connection_made(fake_transport)
+        protocol.user = "foo"
+        fake_transport.write = Mock()
+
+        event = ReachabilityEvent(router=DEVICE_NAME, state=EventState.OPEN)
+        protocol._state.events.commit(event=event)
+
+        command = f"DELETEEVENT {event.id}"
+        await protocol.message_received(command)
+
+        assert fake_transport.write.called
+        response = fake_transport.write.call_args[0][0].decode("utf-8")
+        assert response.startswith("200 ")
+        assert event.id not in protocol._state.events.events.keys()
+        assert not protocol._state.events.get(device_name=DEVICE_NAME, subindex=None, event_class=ReachabilityEvent)
+        assert not protocol._state.events.get_closed_event(
+            device_name=DEVICE_NAME, subindex=None, event_class=ReachabilityEvent
+        )
+
+    @pytest.mark.asyncio
+    async def test_should_delete_closed_event(self):
+        protocol = ZinoTestProtocol()
+        fake_transport = Mock()
+        protocol.connection_made(fake_transport)
+        protocol.user = "foo"
+        fake_transport.write = Mock()
+
+        event = ReachabilityEvent(router=DEVICE_NAME, state=EventState.CLOSED)
+        protocol._state.events.commit(event=event)
+
+        command = f"DELETEEVENT {event.id}"
+        await protocol.message_received(command)
+
+        assert fake_transport.write.called
+        response = fake_transport.write.call_args[0][0].decode("utf-8")
+        assert response.startswith("200 ")
+        assert event.id not in protocol._state.events.events.keys()
+        assert not protocol._state.events.get(device_name=DEVICE_NAME, subindex=None, event_class=ReachabilityEvent)
+        assert not protocol._state.events.get_closed_event(
+            device_name=DEVICE_NAME, subindex=None, event_class=ReachabilityEvent
+        )
+
+
 def test_requires_authentication_should_set_function_attribute():
     @requires_authentication
     def throwaway():