import datetime import responses from responses import matchers from unifi_access.schemas import SystemLogTopic, UserId from .base import UnifiAccessTests class SystemLogTests(UnifiAccessTests): @responses.activate def test_fetch_system_logs(self) -> None: """9.2 Fetch System Logs""" responses.post( f"https://{self.host}/api/v1/developer/system/logs", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher({"page_size": 25, "page_num": 1}), matchers.json_params_matcher( { "topic": "door_openings", "since": 1690770546, "until": 1690771546, "actor_id": "3e1f196e-c97b-4748-aecb-eab5e9c251b2", } ), ], json={ "code": "SUCCESS", "data": { "hits": [ { "@timestamp": "2023-07-11T12:11:27Z", "_id": "", "_source": { "actor": { "alternate_id": "", "alternate_name": "", "display_name": "N/A", "id": "", "type": "user", }, "authentication": { "credential_provider": "NFC", "issuer": "6FC02554", }, "event": { "display_message": "Access Denied / Unknown (NFC)", "published": 1689077487000, "reason": "", "result": "BLOCKED", "type": "access.door.unlock", # NOTE: missing from example, but present in live api "log_key": "", }, "target": [ { "alternate_id": "", "alternate_name": "", "display_name": "UA-HUB-3855", "id": "7483c2773855", "type": "UAH", } ], }, "tag": "access", } ] }, # NOTE: example missing `msg` and `pagination`, had `page` and `total` instead "msg": "succ", "pagination": {"page_num": 1, "page_size": 10000, "total": 1}, }, ) resp = self.client.fetch_system_logs( page_num=1, page_size=25, topic=SystemLogTopic.DOOR_OPENINGS, since=datetime.datetime.fromtimestamp(1690770546, tz=datetime.UTC), until=datetime.datetime.fromtimestamp(1690771546, tz=datetime.UTC), actor_id=UserId("3e1f196e-c97b-4748-aecb-eab5e9c251b2"), ) # NOTE: not taken from API docs examples @responses.activate def test_fetch_system_logs__unpaged(self) -> None: """9.2 Fetch System Logs""" responses.post( f"https://{self.host}/api/v1/developer/system/logs", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher({"page_size": 1, "page_num": 1}), matchers.json_params_matcher({"topic": "door_openings"}), ], json={ "code": "SUCCESS", "data": { "hits": [ { "@timestamp": "2023-07-11T12:11:27Z", "_id": "", "_source": { "actor": { "alternate_id": "", "alternate_name": "", "display_name": "N/A", "id": "", "type": "user", }, "authentication": { "credential_provider": "NFC", "issuer": "6FC02554", }, "event": { "display_message": "Access Denied / Unknown (NFC)", "published": 1689077487000, "reason": "", "result": "BLOCKED", "type": "access.door.unlock", "log_key": "", }, "target": [ { "alternate_id": "", "alternate_name": "", "display_name": "UA-HUB-3855", "id": "7483c2773855", "type": "UAH", } ], }, "tag": "access", } ] }, "msg": "succ", "pagination": {"page_num": 1, "page_size": 1, "total": 2}, }, ) responses.post( f"https://{self.host}/api/v1/developer/system/logs", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher({"page_size": 1, "page_num": 2}), matchers.json_params_matcher({"topic": "door_openings"}), ], json={ "code": "SUCCESS", "data": { "hits": [ { "@timestamp": "2023-07-12T12:11:27Z", "_id": "", "_source": { "actor": { "alternate_id": "", "alternate_name": "", "display_name": "N/A", "id": "", "type": "user", }, "authentication": { "credential_provider": "NFC", "issuer": "6FC02554", }, "event": { "display_message": "Access Denied / Unknown (NFC)", "published": 1689077487000, "reason": "", "result": "BLOCKED", "type": "access.door.unlock", "log_key": "", }, "target": [ { "alternate_id": "", "alternate_name": "", "display_name": "UA-HUB-3855", "id": "7483c2773855", "type": "UAH", } ], }, "tag": "access", } ] }, "msg": "succ", "pagination": {"page_num": 2, "page_size": 1, "total": 2}, }, ) resp = list( self.client.fetch_system_logs__unpaged( page_size=1, topic=SystemLogTopic.DOOR_OPENINGS, ) ) assert resp[0].timestamp == datetime.datetime.fromisoformat( "2023-07-11T12:11:27Z" ) assert resp[1].timestamp == datetime.datetime.fromisoformat( "2023-07-12T12:11:27Z" ) @responses.activate def test_export_system_logs(self) -> None: """9.3 Export System Logs""" responses.post( f"https://{self.host}/api/v1/developer/system/logs/export", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "topic": "door_openings", "since": 1690770546, "until": 1690771546, "timezone": "America/New_York", "actor_id": "3e1f196e-c97b-4748-aecb-eab5e9c251b2", } ), ], # NOTE: example not provided by API # TODO: what does the API do on errors? body=b"time,actor.alternate_id,actor.alternate_name,actor.display_name,actor.id,actor.type,authentication.credential_provider,authentication.issuer,event.display_message,event.published,event.reason,event.result,event.type,target1.alternate_id,target1.alternate_name,target1.display_name,target1.id,target1.type,target2.alternate_id,target2.alternate_name,target2.display_name,target2.id,target2.type,target3.alternate_id,target3.alternate_name,target3.display_name,target3.id,target3.type,target4.alternate_id,target4.alternate_name,target4.display_name,target4.id,target4.type,target5.alternate_id,target5.alternate_name,target5.display_name,target5.id,target5.type\n", ) resp = self.client.export_system_logs( topic=SystemLogTopic.DOOR_OPENINGS, since=datetime.datetime.fromtimestamp(1690770546, tz=datetime.UTC), until=datetime.datetime.fromtimestamp(1690771546, tz=datetime.UTC), timezone="America/New_York", actor_id=UserId("3e1f196e-c97b-4748-aecb-eab5e9c251b2"), ) # TODO: # - 9.4 Fetch Resources in System Logs # - 9.5 Fetch Static Resources in System Logs