diff --git a/src/zinolib/controllers/base.py b/src/zinolib/controllers/base.py index 09e56c5..703efac 100644 --- a/src/zinolib/controllers/base.py +++ b/src/zinolib/controllers/base.py @@ -21,6 +21,7 @@ class ManagerException(Exception): def __init__(self, session=None): self.session = session self.events = {} + self.removed_ids = set() def _get_event(self, event_or_id: EventOrId) -> Event: if isinstance(event_or_id, Event): @@ -29,12 +30,27 @@ def _get_event(self, event_or_id: EventOrId) -> Event: return self.events[event_or_id] raise ValueError("Unknown type") + def _get_event_id(self, event_or_id: EventOrId) -> int: + if isinstance(event_or_id, int): + return event_or_id + if isinstance(event_or_id, Event): + return event_or_id.id + raise ValueError("Unknown type") + def _set_event(self, event: Event): self.events[event.id] = event - def check_session(self): + def remove_event(self, event_or_id: EventOrId): + event_id = self._get_event_id(event_or_id) + self.events.pop(event_id) + self.removed_ids.add(event_id) + + def _verify_session(self, quiet=False): if not self.session: + if quiet: + return False raise ValueError # raise correct error + return True def set_history_for_event(self, event_or_id: EventOrId, history_list: List[HistoryEntry]) -> Event: event = self._get_event(event_or_id) diff --git a/src/zinolib/controllers/zino1.py b/src/zinolib/controllers/zino1.py index bc49aab..975d0f3 100644 --- a/src/zinolib/controllers/zino1.py +++ b/src/zinolib/controllers/zino1.py @@ -1,20 +1,23 @@ """ Get a live Zino 1 session to use:: - > from zinolib.ritz import ritz, parse_tcl_config - > conf = parse_tcl_config("~/.ritz.tcl")['default'] - > session = ritz( - conf['Server'], - username=conf['User'], - password=conf['Secret'], - timeout=30, - ) - > session.connect() - -Now you can use the session when initializing Zino1EventManager:: - + > from zinolib.config.zino1 import ZinoV1Config > from zinolib.zino1 import Zino1EventManager - > event_manager = Zino1EventManager(session) + > config = ZinoV1Config.from_tcl('.ritz.tcl', 'default') + > event_manager = Zino1EventManager.configure(config) + > event_manager.connect() + +The configuration can also be stored in a toml-file:: + + > config = ZinoV1Config.from_toml('.ritz.toml', 'default') + +Authenticate using a username and password from the config-file:: + + > event_manager.authenticate() + +Explicitly autenticate with a username and password:: + + > event_manager.authenticate(username, password) To get a list of currently available events:: @@ -26,6 +29,18 @@ This is a dictionary of event_id, event object pairs. +To get a set of removed event ids:: + + > event_manager.removed_ids + +For updates, either regularly use ``get_events()`` or utilize the UpdateHandler:: + + > updater = UpdateHandler(event_manager) + > updated = updater.poll() + +This updates ``event_manager.events`` and ``event_manager.removed_ids`` and +returns ``True`` on any change, falsey otherwise. + To get history for a specific event:: > history_list = event_manager.get_history_for_id(INT) @@ -51,10 +66,12 @@ from datetime import datetime, timezone from typing import Iterable, List, TypedDict, Optional, Set +import logging from .base import EventManager +from ..compat import StrEnum from ..event_types import EventType, Event, HistoryEntry, LogEntry, AdmState -from ..ritz import ProtocolError +from ..ritz import ProtocolError, ritz, notifier __all__ = [ @@ -74,10 +91,121 @@ ) +DEFAULT_TIMEOUT = 30 +LOG = logging.getLogger(__name__) + + def convert_timestamp(timestamp: int) -> datetime: return datetime.fromtimestamp(timestamp, timezone.utc) +class UpdateHandler: + class UpdateType(StrEnum): + STATE = "state" + ATTR = "attr" + HISTORY = "history" + LOG = "log" + SCAVENGED = "scavenged" + + def __init__(self, manager, autoremove=False): + self.manager = manager + self.events = manager.events + self.autoremove = autoremove + + def poll(self): + update = self.manager.session.push.poll() + if not update: + return False + return self.handle(update) + + def update(self, event_id: int): + event = self.manager.get_updated_event_for_id(event_id) + self.manager._set_event(event) + LOG.debug("Updated event #%i", event_id) + + def remove(self, event_id: int): + self.manager.remove_event(event_id) + LOG.debug("Removed event #%i", event_id) + + def handle(self, update): + if update.id not in self.events and update.type != self.UpdateType.STATE: + # new event that still don't have a state + return None + if update.type in tuple(self.UpdateType): + method = getattr(self, f"cmd_{update.type}") + return method(update) + return self.fallback(update) + + def cmd_state(self, update): + states = update.info.split(" ") + if states[1] == "closed" and self.autoremove: + LOG.debug('Autoremoving "%s"', update.id) + self.remove(update.id) + else: + self.update(update.id) + return True + + def cmd_attr(self, update): + self.update(update.id) + return True + + cmd_history = cmd_attr + cmd_log = cmd_attr + + def cmd_scavenged(self, update): + self.remove(update.id) + return True + + def fallback(self, update): + LOG.warning('Unknown update type: "%s" for id %s' % (update.type, update.id)) + return False + + +class SessionAdapter: + + class _Session: + push = None + request = None + + @classmethod + def create_session(cls, config): + session = cls._setup_request(cls._Session, config) + return session + + @staticmethod + def _setup_request(session, config): + if not session.request: + session.request = ritz( + config.server, + username=config.username, + password=config.password, + timeout=config.timeout, + ) + return session + + @staticmethod + def connect_session(session): + session.request.connect() + session.push = notifier(session.request) + session.push.connect() # ntie + return session + + @staticmethod + def authenticate(session, username=None, password=None): + username = username if username else session.request.username + password = password if password else session.request.password + if not username and password: + raise ValueError("Both username and password must be set and truthy") + session.request.authenticate(username, password) + return session + + @staticmethod + def close_session(session): + session.push._sock.close() + session.request.close() + return None + + class EventAdapter: FIELD_MAP = { "state": "adm_state", @@ -96,8 +224,8 @@ class EventAdapter: } @staticmethod - def get_attrlist(session, event_id: int): - return session.get_raw_attributes(event_id) + def get_attrlist(request, event_id: int): + return request.get_raw_attributes(event_id) @classmethod def attrlist_to_attrdict(cls, attrlist: Iterable[str]): @@ -126,20 +254,20 @@ def convert_values(cls, attrdict): return attrdict @staticmethod - def set_admin_state(session, event: EventType, state: AdmState) -> bool: - return session.set_state(event.id, state.value) + def set_admin_state(request, event: EventType, state: AdmState) -> bool: + return request.set_state(event.id, state.value) @staticmethod - def get_event_ids(session): - return session.get_caseids() + def get_event_ids(request): + return request.get_caseids() class HistoryAdapter: SYSTEM_USER = "monitor" @staticmethod - def get_history(session, event_id: int): - return session.get_raw_history(event_id).data + def get_history(request, event_id: int): + return request.get_raw_history(event_id).data @classmethod def parse_response(cls, history_data: Iterable[str]) -> list[HistoryDict]: @@ -190,11 +318,11 @@ def parse_response(cls, history_data: Iterable[str]) -> list[HistoryDict]: return history_list @classmethod - def add(cls, session, message: str, event: EventType) -> Optional[EventType]: - success = session.add_history(event.id, message) + def add(cls, request, message: str, event: EventType) -> Optional[EventType]: + success = request.add_history(event.id, message) if success: # fetch history - raw_history = cls.get_history(session, event.id) + raw_history = cls.get_history(request, event.id) parsed_history = cls.parse_response(raw_history) new_history = HistoryEntry.create_list(parsed_history) if new_history != event.history: @@ -205,8 +333,8 @@ def add(cls, session, message: str, event: EventType) -> Optional[EventType]: class LogAdapter: @staticmethod - def get_log(session, event_id: int) -> list[str]: - return session.get_raw_log(event_id).data + def get_log(request, event_id: int) -> list[str]: + return request.get_raw_log(event_id).data @staticmethod def parse_response(log_data: Iterable[str]) -> list[LogDict]: @@ -236,11 +364,22 @@ def parse_response(log_data: Iterable[str]) -> list[LogDict]: class Zino1EventManager(EventManager): # Easily replaced in order to ease testing + _session_adapter = SessionAdapter _event_adapter = EventAdapter _history_adapter = HistoryAdapter _log_adapter = LogAdapter removed_ids: Set[int] = set() + @property + def is_authenticated(self): + session_ok = self._verify_session(quiet=True) + return self.session.request.authenticated if session_ok else False + + @property + def is_connected(self): + session_ok = self._verify_session(quiet=True) + return self.session.request.connected if session_ok else False + def rename_exception(self, function, *args): "Replace the original exception with our own" try: @@ -248,20 +387,46 @@ def rename_exception(self, function, *args): except ProtocolError as e: raise self.ManagerException(e) + def _verify_session(self, quiet=False): + if not getattr(self.session, 'request', None): + if quiet: + return False + raise ValueError + return True + + @classmethod + def configure(cls, config): + session = cls._session_adapter.create_session(config) + return cls(session) + + def connect(self): + self._verify_session() + self.session = self._session_adapter.connect_session(self.session) + + def authenticate(self, username=None, password=None): + try: + self.session = self._session_adapter.authenticate(username, password) + except (ProtocolError, ValueError) as e: + raise self.ManagerException(e) + + def disconnect(self): + self._verify_session() + self.session = self._session_adapter.close_session(self.session) + def clear_flapping(self, event: EventType): """Clear flapping state of a PortStateEvent Usage: c = ritz_session.case(123) - c.clear_clapping() + c.clear_flapping() """ if event.type == Event.Type.PortState: - return self.session.clear_flapping(event.router, event.ifindex) + return self.session.request.clear_flapping(event.router, event.ifindex) return None def get_events(self): - self.check_session() - for event_id in self._event_adapter.get_event_ids(self.session): + self._verify_session() + for event_id in self._event_adapter.get_event_ids(self.session.request): try: event = self.create_event_from_id(event_id) except self.ManagerException: @@ -270,8 +435,8 @@ def get_events(self): self.events[event_id] = event def create_event_from_id(self, event_id: int): - self.check_session() - attrlist = self.rename_exception(self._event_adapter.get_attrlist, self.session, event_id) + self._verify_session() + attrlist = self.rename_exception(self._event_adapter.get_attrlist, self.session.request, event_id) attrdict = self._event_adapter.attrlist_to_attrdict(attrlist) attrdict = self._event_adapter.convert_values(attrdict) return Event.create(attrdict) @@ -285,9 +450,9 @@ def get_updated_event_for_id(self, event_id): return event def change_admin_state_for_id(self, event_id, admin_state: AdmState) -> Optional[Event]: - self.check_session() + self._verify_session() event = self._get_event(event_id) - success = self._event_adapter.set_admin_state(self.session, event, admin_state) + success = self._event_adapter.set_admin_state(self.session.request, event, admin_state) if success: event = self.get_updated_event_for_id(event_id) self._set_event(event) @@ -295,15 +460,15 @@ def change_admin_state_for_id(self, event_id, admin_state: AdmState) -> Optional return None def get_history_for_id(self, event_id: int) -> list[HistoryEntry]: - self.check_session() - raw_history = self.rename_exception(self._history_adapter.get_history, self.session, event_id) + self._verify_session() + raw_history = self.rename_exception(self._history_adapter.get_history, self.session.request, event_id) parsed_history = self._history_adapter.parse_response(raw_history) return HistoryEntry.create_list(parsed_history) def add_history_entry_for_id(self, event_id: int, message) -> Optional[EventType]: - self.check_session() + self._verify_session() event = self._get_event(event_id) - success = self._history_adapter.add(self.session, message, event) + success = self._history_adapter.add(self.session.request, message, event) if success: event = self.get_updated_event_for_id(event_id) self._set_event(event) @@ -311,7 +476,7 @@ def add_history_entry_for_id(self, event_id: int, message) -> Optional[EventType return None def get_log_for_id(self, event_id: int) -> list[LogEntry]: - self.check_session() - raw_log = self.rename_exception(self._log_adapter.get_log, self.session, event_id) + self._verify_session() + raw_log = self.rename_exception(self._log_adapter.get_log, self.session.request, event_id) parsed_log = self._log_adapter.parse_response(raw_log) return LogEntry.create_list(parsed_log) diff --git a/tests/atest_mock.py b/tests/atest_mock.py deleted file mode 100644 index 21ac4e8..0000000 --- a/tests/atest_mock.py +++ /dev/null @@ -1,288 +0,0 @@ -from ritz import ritz, ProtocolError, AuthenticationError, caseState, caseType -from ritz.zino_emu import zinoemu -import logging -import datetime -import unittest -from ipaddress import ip_address -import unittest.mock - -ritzlog = logging.getLogger("ritz") -ritzlog.setLevel(logging.DEBUG) -ritzlog.addHandler(logging.FileHandler("test1.log")) - - -class mock_socket: - def __init__(self): - global _mockserver_data - self.output = [] - self.lines = [] - if _mockserver_data: - self.lines.append(_mockserver_data) - _mockserver_data = None - self.conn = None - self.timeout = None - - -def create_connection(address, timeout=3): - try: - int(address[1]) - except ValueError: - raise ValueError("Unable to convert tcp port to integer") - mock_socket() - - -def dict_diff(x, y): - # keys in x not in y - a = [k for k in x.keys() if k not in y.keys()] - if a: - raise KeyError("keys %s not in y" % repr(a)) - - b = [k for k in y.keys() if k not in x.keys()] - if b: - raise KeyError("keys %s not in x" % repr(b)) - - for k in x.keys(): - if x[k] != y[k]: - raise ValueError( - "Values in %s differs, x=%s, y=%s" % (k, repr(x[k]), repr(y[k])) - ) - return True - - -def executor(client): - d = { - "user testuser 7f53cac4ffa877616b8472d3b33a44cbba1907ad -\r\n": ["200 ok\r\n"], - "user auth-failure 7f53cac4ffa877616b8472d3b33a44cbba1907ad -\r\n": [ - "500 Authentication failure\r\n" - ], - "user illegal-first-response 85970fa23a2f5aa06c22b60f04013fe072319ebd -\r\n": [ - "something-gurba-happened" - ], - "user no_login_response 87101b4944f4200fce90f519ddae5eacffeeadf6 -\r\n": [""], - "caseids\r\n": [ - "304 list of active cases follows, terminated with '.'\r\n", - "32802\r\n34978\r\n.\r\n", - ], - "getattrs 32802\r\n": [ - "303 simple attributes follow, terminated with '.'\r\n", - "state: working\r\nrouter: uninett-gsw2\r\ntype: bgp\r\nopened: 1524465142\r\nremote-addr: 2001:700:0:4515::5:11\r\nid: 32802\r\npeer-uptime: 0\r\nupdated: 1533116751\r\npolladdr: 128.39.103.25\r\npriority: 100\r\nbgpOS: down\r\nbgpAS: halted\r\nremote-AS: 64666\r\nlastevent: peer is admin turned off\r\n.\r\n", - ], - "getattrs 34978\r\n": [ - "303 simple attributes follow, terminated with '.'\r\n", - "router: bergen-sw1\r\nstate: working\r\ntype: alarm\r\nalarm-count: 1\r\nopened: 1529156235\r\nalarm-type: yellow\r\nid: 34978\r\nupdated: 1529156235\r\npolladdr: 158.38.234.180\r\npriority: 100\r\nlastevent: alarms went from 0 to 1\r\n.\r\n", - ], - "getattrs 40959\r\n": [ - "303 simple attributes follow, terminated with '.'\r\n", - "state: open\r\nrouter: oslo-gw1\r\ntype: bgp\r\nopened: 1539480952\r\nremote-addr: 193.108.152.34\r\nid: 40959\r\npeer-uptime: 503\r\nupdated: 1539485757\r\npolladdr: 128.39.0.1\r\npriority: 500\r\nbgpOS: established\r\nremote-AS: 21357\r\nbgpAS: running\r\nlastevent: peer was reset (now up)\r\n.\r\n", - ], - "gethist 40959\r\n": [ - "301 history follows, terminated with '.'\r\n", - "1539480952 state change embryonic -> open (monitor)\r\n1539509123 runarb\r\n Testmelding ifra pyRitz\r\n \r\n.\r\n", - ], - "getlog 40959\r\n": [ - "300 log follows, terminated with '.'\r\n", - "1539480952 oslo-gw1 peer 193.108.152.34 AS 21357 was reset (now up)\r\n1539484557 oslo-gw1 peer 193.108.152.34 AS 21357 was reset (now up)\r\n1539485757 oslo-gw1 peer 193.108.152.34 AS 21357 was reset (now up)\r\n.\r\n", - ], - "addhist 40959 -\r\n": [ - "302 please provide new history entry, termiate with '.'\r\n" - ], - "Testmelding ifra pyRitz\r\n\r\n.\r\n": ["200 ok\r\n"], - "setstate 40959 open\r\n": ["200 ok\r\n"], - "setstate 40959 working\r\n": ["200 ok\r\n"], - "setstate 40959 waiting\r\n": ["200 ok\r\n"], - "setstate 40959 confirm-wait\r\n": ["200 ok\r\n"], - "setstate 40959 ignored\r\n": ["200 ok\r\n"], - "setstate 40959 closed\r\n": ["200 ok\r\n"], - "setstate 40960 open\r\n": ["500 Cannot reopen closed event 40960\r\n"], - "clearflap uninett-tor-sw3 707\r\n": ["200 ok\r\n"], - "pollrtr uninett-tor-sw3": ["200 ok\r\n"], - "pollintf uninett-tor-sw3 707": ["200 ok\r\n"], - "ntie 909e90c2eda89a09819ee7fe9b3f67cadb31449f\r\n": ["200 ok\r\n"], - "ntie xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\r\n": [ - "500 Could not find your notify socket\r\n" - ], - } - - client.send("200 2f88fe9d496b1c1a33a8d69f5c3ff7e8c34a1069 Hello, there\r\n") - client.executor(d) - - -class DefaultTest(unittest.TestCase): - def test_A_connect_imediate_disconnect(self): - def client(client): - pass - - with zinoemu(client): - r = ritz("127.0.0.1", username="disconnect-on-connect", password="test") - with self.assertRaises(ProtocolError): - r.connect() - r.close() - - def test_B_connect_illegal_first_response(self): - def client(client): - client.send("This will crash.. :)") - - with zinoemu(client): - r = ritz( - "127.0.0.1", - username="illegal-first-response", - password="illegal-first-response", - ) - with self.assertRaises(ProtocolError): - r.connect() - r.close() - - @unittest.mock.patch("ritz.socket.create_connection") - def test_C_connect_no_login_response(self, create_connection_mock): - - r = ritz( - "127.0.0.1", username="no_login_response", password="no_login_response" - ) - # with self.assertRaises(ProtocolError): - try: - r.connect() - create_connection_mock.assert_called_once_with("127.0.0.1") - except ProtocolError: - pass - finally: - r.close() - - def test_D_connect_authentication_failed(self): - with zinoemu(executor): - r = ritz("127.0.0.1", username="auth-failure", password="test") - with self.assertRaises(AuthenticationError): - r.connect() - r.close() - - def test_E_connect_random_data_on_connect(self): - with zinoemu(executor): - r = ritz( - "127.0.0.1", - username="illegal-first-response", - password="illegal-first-response", - ) - with self.assertRaises(ProtocolError): - r.connect() - r.close() - - def test_F_with(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - self.assertTrue(sess) - - def test_G_get_attributes(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - caseids = sess.get_caseids() - self.assertTrue(32802 in caseids) - self.assertTrue(34978 in caseids) - self.assertFalse(999 in caseids) - test = { - "bgpAS": "halted", - "bgpOS": "down", - "id": 32802, - "lastevent": "peer is admin turned off", - "opened": datetime.datetime(2018, 4, 23, 8, 32, 22), - "peer-uptime": "0", - "polladdr": ip_address("128.39.103.25"), - "priority": 100, - "remote-AS": "64666", - "remote-addr": ip_address("2001:700:0:4515::5:11"), - "router": "uninett-gsw2", - "state": caseState.WORKING, - "type": caseType.BGP, - "updated": datetime.datetime(2018, 8, 1, 11, 45, 51), - } - self.assertTrue(dict_diff(sess.get_attributes(32802), test)) - - test = { - "alarm-count": "1", - "alarm-type": "yellow", - "id": 34978, - "lastevent": "alarms went from 0 to 1", - "opened": datetime.datetime(2018, 6, 16, 15, 37, 15), - "polladdr": ip_address("158.38.234.180"), - "priority": 100, - "router": "bergen-sw1", - "state": caseState.WORKING, - "type": caseType.ALARM, - "updated": datetime.datetime(2018, 6, 16, 15, 37, 15), - } - self.assertTrue(dict_diff(sess.get_attributes(34978), test)) - - def test_H_get_history(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - hist = sess.get_history(40959) - test = { - "date": datetime.datetime(2018, 10, 14, 3, 35, 52), - "header": [ - "1539480952", - "state change embryonic -> open (monitor)", - ], - "user": "monitor", - "log": ["state change embryonic -> open (monitor)"], - } - self.assertTrue(dict_diff(hist[0], test)) - - test = { - "date": datetime.datetime(2018, 10, 14, 11, 25, 23), - "header": ["1539509123", "runarb"], - "user": "runarb", - "log": ["Testmelding ifra pyRitz"], - } - self.assertTrue(dict_diff(hist[1], test)) - - def test_I_add_history(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - self.assertTrue(sess.add_history(40959, "Testmelding ifra pyRitz")) - - def test_J_get_log(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - test = [ - "1539480952 oslo-gw1 peer 193.108.152.34 AS 21357 was reset (now up)", - "1539484557 oslo-gw1 peer 193.108.152.34 AS 21357 was reset (now up)", - "1539485757 oslo-gw1 peer 193.108.152.34 AS 21357 was reset (now up)", - ] - self.assertTrue(sess.get_log(40959) == test) - - def test_K_set_state(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - self.assertTrue(sess.set_state(40959, "open")) - self.assertTrue(sess.set_state(40959, "working")) - self.assertTrue(sess.set_state(40959, "waiting")) - self.assertTrue(sess.set_state(40959, "confirm-wait")) - self.assertTrue(sess.set_state(40959, "ignored")) - self.assertTrue(sess.set_state(40959, "closed")) - with self.assertRaises(ValueError): - sess.set_state(40960, "open") - - def test_L_clear_flap(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - sess.clear_flapping("uninett-tor-sw3", 707) - - def test_M_poll_router(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - sess.poll_router("uninett-tor-sw3") - - def test_N_poll_interface(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - sess.poll_interface("uninett-tor-sw3", 707) - - def test_N_ntie(self): - with zinoemu(executor): - with ritz("127.0.0.1", username="testuser", password="test") as sess: - self.assertTrue(sess.ntie("909e90c2eda89a09819ee7fe9b3f67cadb31449f")) - self.assertTrue(sess.ntie(b"909e90c2eda89a09819ee7fe9b3f67cadb31449f")) - with self.assertRaises(ValueError): - sess.ntie(123456789) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_zinolib_controllers_zino1.py b/tests/test_zinolib_controllers_zino1.py index 11d8ef9..4010a6d 100644 --- a/tests/test_zinolib_controllers_zino1.py +++ b/tests/test_zinolib_controllers_zino1.py @@ -2,7 +2,8 @@ from datetime import datetime, timedelta, timezone from zinolib.event_types import AdmState, Event, HistoryEntry, LogEntry -from zinolib.controllers.zino1 import EventAdapter, HistoryAdapter, LogAdapter, Zino1EventManager +from zinolib.controllers.zino1 import EventAdapter, HistoryAdapter, LogAdapter, SessionAdapter, Zino1EventManager, UpdateHandler +from zinolib.ritz import NotifierResponse raw_event_id = 139110 raw_attrlist = [ @@ -41,8 +42,8 @@ class FakeEventAdapter: @staticmethod - def get_attrlist(session, event_id: int): - return raw_attrlist + def get_attrlist(request, event_id: int): + return raw_attrlist.copy() @classmethod def attrlist_to_attrdict(cls, attrlist): @@ -53,26 +54,39 @@ def convert_values(cls, attrdict): return EventAdapter.convert_values(attrdict) @staticmethod - def get_event_ids(session): + def get_event_ids(request): return [raw_event_id] class FakeHistoryAdapter(HistoryAdapter): @staticmethod - def get_history(session, event_id: int): - return raw_history + def get_history(request, event_id: int): + return raw_history.copy() class FakeLogAdapter(LogAdapter): @staticmethod - def get_log(session, event_id: int): - return raw_log + def get_log(request, event_id: int): + return raw_log.copy() + + +class FakeSessionAdapter(SessionAdapter): + + @classmethod + def _setup_config(cls, config): + pass + + @staticmethod + def _setup_request(session, config): + session.request = 'foo' # needs to be truthy + return session class FakeZino1EventManager(Zino1EventManager): _event_adapter = FakeEventAdapter _history_adapter = FakeHistoryAdapter _log_adapter = FakeLogAdapter + _session_adapter = FakeSessionAdapter def __init__(self, session=None): super().__init__(session) @@ -80,8 +94,12 @@ def __init__(self, session=None): class Zino1EventManagerTest(unittest.TestCase): + def init_manager(self): + zino1 = FakeZino1EventManager.configure(None) + return zino1 + def test_get_events(self): - zino1 = FakeZino1EventManager('foo') + zino1 = self.init_manager() self.assertEqual(len(zino1.events), 0) zino1.get_events() self.assertEqual(len(zino1.events), 1) @@ -89,7 +107,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 = FakeZino1EventManager('foo') + zino1 = self.init_manager() history_list = zino1.get_history_for_id(4567) expected_history_list = [ HistoryEntry( @@ -116,7 +134,7 @@ def test_get_history_for_id(self): self.assertEqual(history_list, expected_history_list) def test_get_log_for_id(self): - zino1 = FakeZino1EventManager('foo') + zino1 = self.init_manager() log_list = zino1.get_log_for_id(4567) expected_log_list = [ LogEntry( @@ -127,3 +145,101 @@ def test_get_log_for_id(self): log='some other log message') ] self.assertEqual(log_list, expected_log_list) + + +class UpdateHandlerTest(unittest.TestCase): + + def init_manager(self): + zino1 = FakeZino1EventManager.configure(None) + return zino1 + + def test_cmd_scavenged(self): + zino1 = self.init_manager() + zino1.get_events() + self.assertIn(raw_event_id, zino1.events) + self.assertNotIn(raw_event_id, zino1.removed_ids) + updates = UpdateHandler(zino1) + update = NotifierResponse(raw_event_id, "","") + ok = updates.cmd_scavenged(update) + self.assertTrue(ok) + self.assertNotIn(raw_event_id, zino1.events) + self.assertIn(raw_event_id, zino1.removed_ids) + + def test_cmd_attr(self): + zino1 = self.init_manager() + zino1.get_events() + old_events = zino1.events.copy() + old_events[raw_event_id].priority = 500 + updates = UpdateHandler(zino1) + update = NotifierResponse(raw_event_id, "","") + ok = updates.cmd_attr(update) + self.assertTrue(ok) + self.assertNotEqual(zino1.events[raw_event_id].priority, old_events[raw_event_id].priority) + + def test_cmd_state_is_closed_and_autoremove_is_on(self): + zino1 = self.init_manager() + zino1.get_events() + self.assertNotIn(raw_event_id, zino1.removed_ids) + self.assertIn(raw_event_id, zino1.events) + updates = UpdateHandler(zino1, autoremove=True) + update = NotifierResponse(raw_event_id, "", "X closed") + ok = updates.cmd_state(update) + self.assertTrue(ok) + self.assertIn(raw_event_id, zino1.removed_ids) + self.assertNotIn(raw_event_id, zino1.events) + + def test_cmd_state_is_closed_and_autoremove_is_off(self): + zino1 = self.init_manager() + zino1.get_events() + old_events = zino1.events.copy() + old_events[raw_event_id].priority = 500 + updates = UpdateHandler(zino1, autoremove=False) + update = NotifierResponse(raw_event_id, "","X closed") + ok = updates.cmd_state(update) + self.assertTrue(ok) + self.assertNotEqual(zino1.events[raw_event_id].priority, old_events[raw_event_id].priority) + + def test_cmd_state_is_not_closed(self): + zino1 = self.init_manager() + zino1.get_events() + old_events = zino1.events.copy() + old_events[raw_event_id].priority = 500 + updates = UpdateHandler(zino1, autoremove=False) + update = NotifierResponse(raw_event_id, "","x butterfly") + ok = updates.cmd_state(update) + self.assertTrue(ok) + self.assertNotEqual(zino1.events[raw_event_id].priority, old_events[raw_event_id].priority) + + def test_fallback(self): + zino1 = self.init_manager() + updates = UpdateHandler(zino1) + update = NotifierResponse(raw_event_id, "", "") + with self.assertLogs('zinolib.controllers.zino1', level='WARNING') as cm: + self.assertFalse(updates.fallback(update)) + + def test_handle_new_stateless_event_is_very_special(self): + zino1 = self.init_manager() + updates = UpdateHandler(zino1) + update = NotifierResponse(1337, "", "") + result = updates.handle(update) + self.assertEqual(result, None) + + def test_handle_known_type(self): + zino1 = self.init_manager() + zino1.get_events() + old_events = zino1.events.copy() + old_events[raw_event_id].priority = 500 + updates = UpdateHandler(zino1) + update = NotifierResponse(raw_event_id, updates.UpdateType.LOG, "") + ok = updates.handle(update) # will refetch events + self.assertTrue(ok) + self.assertNotEqual(zino1.events[raw_event_id].priority, old_events[raw_event_id].priority) + + def test_handle_unknown_type(self): + zino1 = self.init_manager() + zino1.get_events() + updates = UpdateHandler(zino1) + update = NotifierResponse(raw_event_id, "trout", "") + with self.assertLogs('zinolib.controllers.zino1', level='WARNING'): + ok = updates.handle(update) # will run fallback + self.assertFalse(ok) diff --git a/tests/test_zinolib_event_types.py b/tests/test_zinolib_event_types.py index cbc4507..d9b46bf 100644 --- a/tests/test_zinolib_event_types.py +++ b/tests/test_zinolib_event_types.py @@ -213,10 +213,10 @@ def setUp(self): event = Event.create(minimal_input) self.event = event - def test_check_session_wjen_no_session_should_fail_noisily(self): + def test__verify_session_wjen_no_session_should_fail_noisily(self): event_manager = EventManager() with self.assertRaises(ValueError): - event_manager.check_session() + event_manager._verify_session() def test_get_event_with_id_should_succeed(self): event = self.event