From f0fa02040a69398e008b9ead1c86f3e61a9e2e3f Mon Sep 17 00:00:00 2001 From: Hanne Moa Date: Fri, 6 Oct 2023 14:46:47 +0200 Subject: [PATCH] Rename EventEngine to EventManager As per feedback. --- src/zinolib/controllers/base.py | 4 +-- src/zinolib/controllers/zino1.py | 30 +++++++++++----------- src/zinolib/zino1.py | 6 ++++- tests/test_zinolib_controllers_zino1.py | 12 ++++----- tests/test_zinolib_event_types.py | 34 ++++++++++++------------- 5 files changed, 45 insertions(+), 41 deletions(-) diff --git a/src/zinolib/controllers/base.py b/src/zinolib/controllers/base.py index 664461d..31e447d 100644 --- a/src/zinolib/controllers/base.py +++ b/src/zinolib/controllers/base.py @@ -6,7 +6,7 @@ EventOrId = Union[EventType, int] -class EventEngine: +class EventManager: """ Implementation-agnostic controller for events @@ -15,7 +15,7 @@ class EventEngine: """ events: Dict[int, Event] - class EngineException(Exception): + class ManagerException(Exception): pass def __init__(self, session=None): diff --git a/src/zinolib/controllers/zino1.py b/src/zinolib/controllers/zino1.py index 4e139c8..3903e7a 100644 --- a/src/zinolib/controllers/zino1.py +++ b/src/zinolib/controllers/zino1.py @@ -11,38 +11,38 @@ ) > session.connect() -Now you can use the session when initializing Zino1EventEngine:: +Now you can use the session when initializing Zino1EventManager:: - > from zinolib.zino1 import Zino1EventEngine - > event_engine = Zino1EventEngine(session) + > from zinolib.zino1 import Zino1EventManager + > event_manager = Zino1EventManager(session) To get a list of currently available events:: - > event_engine.get_events() + > event_manager.get_events() The events are then available as:: - > event_engine.events + > event_manager.events This is a dictionary of event_id, event object pairs. To get history for a specific event:: - > history_list = event_engine.get_history_for_id(INT) + > history_list = event_manager.get_history_for_id(INT) To get the log for a specific event:: - > log_list = event_engine.get_log_for_id(INT) + > log_list = event_manager.get_log_for_id(INT) History can be stored on the correct event with one of:: - > event_engine.set_history_for_event(event, history_list) - > event_engine.set_history_for_event(INT, history_list) + > event_manager.set_history_for_event(event, history_list) + > event_manager.set_history_for_event(INT, history_list) Log can be stored on the correct event with one of:: - > event_engine.set_log_for_event(event, log_list) - > event_engine.set_log_for_event(INT, log_list) + > event_manager.set_log_for_event(event, log_list) + > event_manager.set_log_for_event(INT, log_list) Both return the changed event. @@ -51,7 +51,7 @@ from datetime import datetime, timezone from typing import Iterable, List, TypedDict, Optional -from .base import EventEngine +from .base import EventManager from ..event_types import EventType, Event, HistoryEntry, LogEntry, AdmState from ..ritz import ProtocolError @@ -228,7 +228,7 @@ def parse_response(log_data: Iterable[str]) -> list[LogDict]: return log_list -class Zino1EventEngine(EventEngine): +class Zino1EventManager(EventManager): # Easily replaced in order to ease testing _event_adapter = EventAdapter _history_adapter = HistoryAdapter @@ -240,7 +240,7 @@ def rename_exception(self, function, *args): try: return function(*args) except ProtocolError as e: - raise self.EngineException(e) + raise self.ManagerException(e) def clear_flapping(self, event: EventType): """Clear flapping state of a PortStateEvent @@ -258,7 +258,7 @@ def get_events(self): for event_id in self._event_adapter.get_event_ids(self.session): try: event = self.create_event_from_id(event_id) - except self.EngineException: + except self.ManagerException: self.removed_ids.add(event_id) continue self.events[event_id] = event diff --git a/src/zinolib/zino1.py b/src/zinolib/zino1.py index ee3895b..763951a 100644 --- a/src/zinolib/zino1.py +++ b/src/zinolib/zino1.py @@ -4,7 +4,10 @@ from datetime import datetime, timezone -from .controllers.zino1 import SessionAdapter, EventAdapter, HistoryAdapter, LogAdapter, Zino1EventEngine, HistoryDict, LogDict +from .controllers.zino1 import SessionAdapter, EventAdapter, HistoryAdapter, LogAdapter, Zino1EventManager, HistoryDict, LogDict + + +Zino1EventEngine = Zino1EventManager __all__ = [ @@ -13,6 +16,7 @@ "HistoryAdapter", "LogAdapter", "Zino1EventEngine", + "Zino1EventManager", "HistoryDict", "LogDict", ] diff --git a/tests/test_zinolib_controllers_zino1.py b/tests/test_zinolib_controllers_zino1.py index 1125312..11d8ef9 100644 --- a/tests/test_zinolib_controllers_zino1.py +++ b/tests/test_zinolib_controllers_zino1.py @@ -2,7 +2,7 @@ from datetime import datetime, timedelta, timezone from zinolib.event_types import AdmState, Event, HistoryEntry, LogEntry -from zinolib.controllers.zino1 import EventAdapter, HistoryAdapter, LogAdapter, Zino1EventEngine +from zinolib.controllers.zino1 import EventAdapter, HistoryAdapter, LogAdapter, Zino1EventManager raw_event_id = 139110 raw_attrlist = [ @@ -69,7 +69,7 @@ def get_log(session, event_id: int): return raw_log -class FakeZino1EventEngine(Zino1EventEngine): +class FakeZino1EventManager(Zino1EventManager): _event_adapter = FakeEventAdapter _history_adapter = FakeHistoryAdapter _log_adapter = FakeLogAdapter @@ -78,10 +78,10 @@ def __init__(self, session=None): super().__init__(session) -class Zino1EventEngineTest(unittest.TestCase): +class Zino1EventManagerTest(unittest.TestCase): def test_get_events(self): - zino1 = FakeZino1EventEngine('foo') + zino1 = FakeZino1EventManager('foo') self.assertEqual(len(zino1.events), 0) zino1.get_events() self.assertEqual(len(zino1.events), 1) @@ -89,7 +89,7 @@ def test_get_events(self): self.assertEqual(zino1.events[raw_event_id].id, raw_event_id) def test_get_history_for_id(self): - zino1 = FakeZino1EventEngine('foo') + zino1 = FakeZino1EventManager('foo') history_list = zino1.get_history_for_id(4567) expected_history_list = [ HistoryEntry( @@ -116,7 +116,7 @@ def test_get_history_for_id(self): self.assertEqual(history_list, expected_history_list) def test_get_log_for_id(self): - zino1 = FakeZino1EventEngine('foo') + zino1 = FakeZino1EventManager('foo') log_list = zino1.get_log_for_id(4567) expected_log_list = [ LogEntry( diff --git a/tests/test_zinolib_event_types.py b/tests/test_zinolib_event_types.py index 0487655..cbc4507 100644 --- a/tests/test_zinolib_event_types.py +++ b/tests/test_zinolib_event_types.py @@ -5,7 +5,7 @@ from pydantic import ValidationError -from zinolib.controllers.base import EventEngine +from zinolib.controllers.base import EventManager from zinolib.event_types import Event, HistoryEntry, LogEntry from zinolib.event_types import AdmState, BFDState, PortState, ReachabilityState @@ -200,7 +200,7 @@ def test_create_list_should_return_list_of_log_entries(self): self.assertEqual(log, [expected_log_entry]) -class EventEngineTest(unittest.TestCase): +class EventManagerTest(unittest.TestCase): def setUp(self): minimal_input = common_minimal_input.copy() @@ -214,41 +214,41 @@ def setUp(self): self.event = event def test_check_session_wjen_no_session_should_fail_noisily(self): - event_engine = EventEngine() + event_manager = EventManager() with self.assertRaises(ValueError): - event_engine.check_session() + event_manager.check_session() def test_get_event_with_id_should_succeed(self): event = self.event - event_engine = EventEngine() - event_engine._set_event(event) + event_manager = EventManager() + event_manager._set_event(event) - resulting_event = event_engine._get_event(event.id) + resulting_event = event_manager._get_event(event.id) self.assertEqual(self.event, resulting_event) def test_set_history_for_event(self): event = self.event - event_engine = EventEngine() - event_engine._set_event(event) - self.assertFalse(event_engine.events[event.id].history) + event_manager = EventManager() + event_manager._set_event(event) + self.assertFalse(event_manager.events[event.id].history) dt = datetime.fromisoformat("2023-06-28T10:41:54+00:00") history_list = [{ "date": dt, "log": "fhgj", "user": "ghj", }] - updated_event = event_engine.set_history_for_event(event, history_list) - self.assertTrue(event_engine.events[event.id].history) + updated_event = event_manager.set_history_for_event(event, history_list) + self.assertTrue(event_manager.events[event.id].history) def test_set_log_for_event(self): event = self.event - event_engine = EventEngine() - event_engine._set_event(event) - self.assertFalse(event_engine.events[event.id].log) + event_manager = EventManager() + event_manager._set_event(event) + self.assertFalse(event_manager.events[event.id].log) dt = datetime.fromisoformat("2023-06-28T10:41:54+00:00") log_list = [{ "date": dt, "log": "fhgj", }] - updated_event = event_engine.set_log_for_event(event, log_list) - self.assertTrue(event_engine.events[event.id].log) + updated_event = event_manager.set_log_for_event(event, log_list) + self.assertTrue(event_manager.events[event.id].log)