unifi_access/tests/test_system_log.py

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"),
        ).success_or_raise()

    @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