From 78eea76f6b60ab692ae2f36608d123f031d2f30e Mon Sep 17 00:00:00 2001 From: drewclauson Date: Fri, 21 May 2021 15:19:31 -0400 Subject: [PATCH 01/21] Initial start to adding door_control support --- axis/door_control.py | 171 +++++++++ axis/param_cgi.py | 5 + axis/vapix.py | 6 + tests/test_door_control.py | 704 +++++++++++++++++++++++++++++++++++++ 4 files changed, 886 insertions(+) create mode 100644 axis/door_control.py create mode 100644 tests/test_door_control.py diff --git a/axis/door_control.py b/axis/door_control.py new file mode 100644 index 00000000..873cb336 --- /dev/null +++ b/axis/door_control.py @@ -0,0 +1,171 @@ +"""Door Control API. + +The Axis Door control API makes it possible to control the behavior and functionality +of physical access controls in the Axis devices (e.g. A1001, A1601) +""" + +import attr + +from .api import APIItem, APIItems, Body + +URL = "/vapix/doorcontrol" + +API_DISCOVERY_ID = "door-control" +API_VERSION = "1.0" + + +class DoorControl(APIItems): + """Door control for Axis devices.""" + + def __init__(self, request: object) -> None: + super().__init__({}, request, URL, Door) + + # TODO: Question: Is this used to get status information for the door? Or to update the object properties? + async def update(self) -> None: + raw = await self.get_door_info_list() + self.process_raw(raw) + + @staticmethod + def pre_process_raw(raw: dict) -> dict: + """Return a dictionary of doors.""" + door_control_data = raw.get("DoorInfo", {}) + return {api["token"]: api for api in door_control_data} + + async def get_service_capabilities(self) -> dict: + """List the capabilities of the door controller.""" + return await self._request( + "post", + URL, + json={"tdc:GetServiceCapabilities": {}}, + ) + + async def get_door_info_list(self) -> dict: + """List the doors.""" + return await self._request( + "post", + URL, + json={"tdc:GetDoorInfoList": {}}, + ) + + async def get_door_info(self, door_token: str) -> dict: + """List the door information.""" + return await self._request( + "post", + URL, + json={"tdc:GetDoorInfo": {"Token": door_token}} + ) + + async def get_door_state(self, door_token: str) -> dict: + """List the door information.""" + return await self._request( + "post", + URL, + json={"tdc:GetDoorState": {"Token": door_token}} + ) + + # region Door Actions + async def access_door(self, door_token: str) -> None: + """Access the door. Use when a credential holder is granted access, + for example by swiping a card in a card reader.""" + await self._request( + "post", + URL, + json={"tdc:AccessDoor": {"Token": door_token}} + ) + + async def lock_door(self, door_token: str) -> None: + """Lock the door.""" + await self._request( + "post", + URL, + json={"tdc:LockDoor": {"Token": door_token}} + ) + + async def unlock_door(self, door_token: str) -> None: + """Unlock the door until it is explicitly locked again.""" + await self._request( + "post", + URL, + json={"tdc:UnlockDoor": {"Token": door_token}} + ) + + async def block_door(self, door_token: str) -> None: + """Block the door.""" + await self._request( + "post", + URL, + json={"tdc:BlockDoor": {"Token": door_token}} + ) + + async def lock_down_door(self, door_token: str) -> None: + """Lock the door and prevent all other commands until a LockDownReleaseDoor command is sent.""" + await self._request( + "post", + URL, + json={"tdc:LockDownDoor": {"Token": door_token}} + ) + + async def lock_down_release_door(self, door_token: str) -> None: + """Release the door from the LockedDown state.""" + await self._request( + "post", + URL, + json={"tdc:LockDownReleaseDoor": {"Token": door_token}} + ) + + async def lock_open_door(self, door_token: str) -> None: + """Unlock the door and prevent all other commands until a LockOpenReleaseDoor command is sent.""" + await self._request( + "post", + URL, + json={"tdc:LockOpenDoor": {"Token": door_token}} + ) + + async def lock_open_release_door(self, door_token: str) -> None: + """Release the door from the LockedOpen state.""" + await self._request( + "post", + URL, + json={"tdc:LockOpenReleaseDoor": {"Token": door_token}} + ) + + async def double_lock_door(self, door_token: str) -> None: + """Lock the door with a double lock.""" + await self._request( + "post", + URL, + json={"tdc:DoubleLockDoor": {"Token": door_token}} + ) + + async def release_door(self, door_token: str) -> None: + """Release the door from a priority level.""" + await self._request( + "post", + URL, + json={"axtdc:ReleaseDoor": {"Token": door_token}} + ) + # endregion + + +class Door(APIItem): + """API Discovery item.""" + + @property + def door_token(self) -> str: + """Token of Door.""" + return self.raw["token"] + + @property + def door_name(self) -> str: + """Name of Door.""" + return self.raw["Name"] + + @property + def door_description(self) -> str: + """Description of Door.""" + return self.raw["Description"] + + @property + def door_capabilities(self) -> dict: + """Capabilities of Door.""" + return self.raw["Capabilities"] diff --git a/axis/param_cgi.py b/axis/param_cgi.py index 29c19663..385834c4 100644 --- a/axis/param_cgi.py +++ b/axis/param_cgi.py @@ -343,6 +343,11 @@ def light_control(self) -> bool: """Support light control.""" return self.get(PROPERTIES, {}).get("LightControl.LightControl2") == "yes" + @property + def door_control(self) -> bool: + """FIXME: Figure out a different way to determine if door control is supported """ + return self.get(BRAND, {}).get("ProdNbr") == "A1001" + @property def ptz(self) -> bool: """Support PTZ control.""" diff --git a/axis/vapix.py b/axis/vapix.py index a51e61b6..2ca89f34 100644 --- a/axis/vapix.py +++ b/axis/vapix.py @@ -21,6 +21,7 @@ from .applications.vmd4 import Vmd4 from .basic_device_info import API_DISCOVERY_ID as BASIC_DEVICE_INFO_ID, BasicDeviceInfo from .configuration import Configuration +from .door_control import DoorControl, API_DISCOVERY_ID as DOOR_CONTROL_ID from .errors import PathNotFound, RequestError, Unauthorized, raise_error from .event_instances import EventInstances from .light_control import API_DISCOVERY_ID as LIGHT_CONTROL_ID, LightControl @@ -51,6 +52,7 @@ def __init__(self, config: Configuration) -> None: self.applications: Optional[Applications] = None self.basic_device_info: Optional[BasicDeviceInfo] = None self.event_instances: Optional[EventInstances] = None + self.door_control = Optional[DoorControl] = None self.fence_guard: Optional[FenceGuard] = None self.light_control: Optional[LightControl] = None self.loitering_guard: Optional[LoiteringGuard] = None @@ -139,6 +141,7 @@ async def initialize_api_discovery(self) -> None: for api_id, api_class, api_attr in ( (BASIC_DEVICE_INFO_ID, BasicDeviceInfo, "basic_device_info"), (IO_PORT_MANAGEMENT_ID, IoPortManagement, "ports"), + (DOOR_CONTROL_ID, DoorControl, "door_control"), (LIGHT_CONTROL_ID, LightControl, "light_control"), (MQTT_ID, MqttClient, "mqtt"), (STREAM_PROFILES_ID, StreamProfiles, "stream_profiles"), @@ -181,6 +184,9 @@ async def initialize_param_cgi(self, preload_data: bool = True) -> None: if not self.light_control and self.params.light_control: await self._initialize_api_attribute(LightControl, "light_control") + if not self.door_control and self.params.door_control: + await self._initialize_api_attribute(DoorControl, "door_control") + if not self.ports: self.ports = Ports(self.params, self.request) diff --git a/tests/test_door_control.py b/tests/test_door_control.py new file mode 100644 index 00000000..82f1bbd1 --- /dev/null +++ b/tests/test_door_control.py @@ -0,0 +1,704 @@ +"""Test Axis Door Control API. + +pytest --cov-report term-missing --cov=axis.light_control tests/test_light_control.py +""" + +import json +import pytest + +import respx + +from axis.door_control import DoorControl +from .conftest import HOST + + +@pytest.fixture +def door_control(axis_device) -> DoorControl: + """Returns the door_control mock object.""" + return DoorControl(axis_device.vapix.request) + + +@respx.mock +async def test_update(door_control): + """Test update method.""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={"DoorInfo": [ + { + "token": "Axis-accc8ea9abac:1550808050.595717000", + "Name": "Main Door", + "Description": "Main Door Description", + "Capabilities": { + "Access": True, + "Lock": True, + "Unlock": True, + "Block": True, + "DoubleLock": True, + "LockDown": True, + "LockOpen": True, + "DoorMonitor": True, + "LockMonitor": False, + "DoubleLockMonitor": False, + "Alarm": True, + "Tamper": False, + "Warning": True, + "Configurable": True + } + }, + { + "token": "Axis-accc8ee70fbe:1614733258.565014000", + "Name": "North Gym Door", + "Description": "North Gym Door Description", + "Capabilities": {} + } + ] + } + ) + + await door_control.update() + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:GetDoorInfoList": {}} + + assert len(door_control.values()) == 2 + + item = door_control["Axis-accc8ea9abac:1550808050.595717000"] + assert item.id == "Axis-accc8ea9abac:1550808050.595717000" + assert item.door_token == "Axis-accc8ea9abac:1550808050.595717000" + assert item.door_name == "Main Door" + assert item.door_description == "Main Door Description" + + # TODO: Check to see that comparing dict using "==" actually does what I want it to do + assert item.door_capabilities == { + "Access": True, + "Lock": True, + "Unlock": True, + "Block": True, + "DoubleLock": True, + "LockDown": True, + "LockOpen": True, + "DoorMonitor": True, + "LockMonitor": False, + "DoubleLockMonitor": False, + "Alarm": True, + "Tamper": False, + "Warning": True, + "Configurable": True + } + + item2 = door_control["Axis-accc8ee70fbe:1614733258.565014000"] + assert item2.id == "Axis-accc8ee70fbe:1614733258.565014000" + assert item2.door_token == "Axis-accc8ee70fbe:1614733258.565014000" + assert item2.door_name == "North Gym Door" + assert item2.description == "North Gym Door Description" + # TODO: Check to see that comparing dict using "==" actually does what I want it to do + assert item2.door_capabilities == {} + + +@respx.mock +async def test_get_service_capabilities(door_control): + """Test get service capabilities API.""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={ + "Capabilities": { + "MaxLimit": 100, + "GetDoorSupported": True, + "SetDoorSupported": True, + "PriorityConfigurationSupported": True + } + }, + ) + + response = await door_control.get_service_capabilities() + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:GetServiceCapabilities": {}} + + assert response["Capabilities"] == { + "MaxLimit": 100, + "GetDoorSupported": True, + "SetDoorSupported": True, + "PriorityConfigurationSupported": True + } + + +@respx.mock +async def test_get_light_information(light_control): + """Test get light information API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "getLightInformation", + "data": { + "items": [ + { + "lightID": "led0", + "lightType": "IR", + "enabled": True, + "synchronizeDayNightMode": True, + "lightState": False, + "automaticIntensityMode": False, + "automaticAngleOfIlluminationMode": False, + "nrOfLEDs": 1, + "error": False, + "errorInfo": "", + } + ] + }, + }, + ) + + response = await light_control.get_light_information() + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getLightInformation", + "apiVersion": "1.1", + "context": "Axis library", + } + + assert response["data"] == { + "items": [ + { + "lightID": "led0", + "lightType": "IR", + "enabled": True, + "synchronizeDayNightMode": True, + "lightState": False, + "automaticIntensityMode": False, + "automaticAngleOfIlluminationMode": False, + "nrOfLEDs": 1, + "error": False, + "errorInfo": "", + } + ] + } + + +@respx.mock +async def test_activate_light(light_control): + """Test activating light API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "activateLight", + "data": {}, + }, + ) + + await light_control.activate_light("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "activateLight", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + +@respx.mock +async def test_deactivate_light(light_control): + """Test deactivating light API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "deactivateLight", + "data": {}, + }, + ) + + await light_control.deactivate_light("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "deactivateLight", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + +@respx.mock +async def test_enable_light(light_control): + """Test enabling light API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "enableLight", + "data": {}, + }, + ) + + await light_control.enable_light("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "enableLight", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + +@respx.mock +async def test_disable_light(light_control): + """Test disabling light API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "disableLight", + "data": {}, + }, + ) + + await light_control.disable_light("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "disableLight", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + +@respx.mock +async def test_get_light_status(light_control): + """Test get light status API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "getLightStatus", + "data": {"status": False}, + }, + ) + + response = await light_control.get_light_status("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getLightStatus", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == {"status": False} + + +@respx.mock +async def test_set_automatic_intensity_mode(light_control): + """Test set automatic intensity mode API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "setAutomaticIntensityMode", + "data": {}, + }, + ) + + await light_control.set_automatic_intensity_mode("led0", True) + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "setAutomaticIntensityMode", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0", "enabled": True}, + } + + +@respx.mock +async def test_get_manual_intensity(light_control): + """Test get valid intensity API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "getManualIntensity", + "data": {"intensity": 1000}, + }, + ) + + response = await light_control.get_manual_intensity("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getManualIntensity", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == {"intensity": 1000} + + +@respx.mock +async def test_set_manual_intensity(light_control): + """Test set manual intensity API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "setManualIntensity", + "data": {}, + }, + ) + + await light_control.set_manual_intensity("led0", 1000) + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "setManualIntensity", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0", "intensity": 1000}, + } + + +@respx.mock +async def test_get_valid_intensity(light_control): + """Test get valid intensity API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "getValidIntensity", + "data": {"ranges": [{"low": 0, "high": 1000}]}, + }, + ) + + response = await light_control.get_valid_intensity("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getValidIntensity", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == {"ranges": [{"low": 0, "high": 1000}]} + + +@respx.mock +async def test_set_individual_intensity(light_control): + """Test set individual intensity API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "setIndividualIntensity", + "data": {}, + }, + ) + + await light_control.set_individual_intensity("led0", 1, 1000) + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "setIndividualIntensity", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0", "LEDID": 1, "intensity": 1000}, + } + + +@respx.mock +async def test_get_individual_intensity(light_control): + """Test get individual intensity API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "getIndividualIntensity", + "data": {"intensity": 1000}, + }, + ) + + response = await light_control.get_individual_intensity("led0", 1) + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getIndividualIntensity", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0", "LEDID": 1}, + } + + assert response["data"] == {"intensity": 1000} + + +@respx.mock +async def test_get_current_intensity(light_control): + """Test get current intensity API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "getCurrentIntensity", + "data": {"intensity": 1000}, + }, + ) + + response = await light_control.get_current_intensity("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getCurrentIntensity", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == {"intensity": 1000} + + +@respx.mock +async def test_set_automatic_angle_of_illumination_mode(light_control): + """Test set automatic angle of illumination mode API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "setAutomaticAngleOfIlluminationMode", + "data": {}, + }, + ) + + await light_control.set_automatic_angle_of_illumination_mode("led0", True) + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "setAutomaticAngleOfIlluminationMode", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0", "enabled": True}, + } + + +@respx.mock +async def test_get_valid_angle_of_illumination(light_control): + """Test get valid angle of illumination API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.0", + "context": "my context", + "method": "getValidAngleOfIllumination", + "data": {"ranges": [{"low": 10, "high": 30}, {"low": 20, "high": 50}]}, + }, + ) + + response = await light_control.get_valid_angle_of_illumination("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getValidAngleOfIllumination", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == { + "ranges": [{"low": 10, "high": 30}, {"low": 20, "high": 50}] + } + + +@respx.mock +async def test_set_manual_angle_of_illumination(light_control): + """Test set manual angle of illumination API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "setManualAngleOfIllumination", + "data": {}, + }, + ) + + await light_control.set_manual_angle_of_illumination("led0", 30) + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "setManualAngleOfIllumination", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0", "angleOfIllumination": 30}, + } + + +@respx.mock +async def test_get_manual_angle_of_illumination(light_control): + """Test get manual angle of illumination API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.0", + "context": "my context", + "method": "getManualAngleOfIllumination", + "data": {"angleOfIllumination": 30}, + }, + ) + + response = await light_control.get_manual_angle_of_illumination("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getManualAngleOfIllumination", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == {"angleOfIllumination": 30} + + +@respx.mock +async def test_get_current_angle_of_illumination(light_control): + """Test get current angle of illumination API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.0", + "context": "my context", + "method": "getCurrentAngleOfIllumination", + "data": {"angleOfIllumination": 20}, + }, + ) + + response = await light_control.get_current_angle_of_illumination("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getCurrentAngleOfIllumination", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == {"angleOfIllumination": 20} + + +@respx.mock +async def test_set_light_synchronization_day_night_mode(light_control): + """Test set light synchronization day night mode API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "method": "setLightSynchronizationDayNightMode", + "data": {}, + }, + ) + + await light_control.set_light_synchronization_day_night_mode("led0", True) + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "setLightSynchronizationDayNightMode", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0", "enabled": True}, + } + + +@respx.mock +async def test_get_light_synchronization_day_night_mode(light_control): + """Test get light synchronization day night mode API.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "apiVersion": "1.1", + "context": "my context", + "method": "getLightSynchronizeDayNightMode", + "data": {"enabled": True}, + }, + ) + + response = await light_control.get_light_synchronization_day_night_mode("led0") + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getLightSynchronizationDayNightMode", + "apiVersion": "1.1", + "context": "Axis library", + "params": {"lightID": "led0"}, + } + + assert response["data"] == {"enabled": True} + + +@respx.mock +async def test_get_supported_versions(light_control): + """Test get supported versions api.""" + route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( + json={ + "method": "getSupportedVersions", + "data": {"apiVersions": ["1.1"]}, + }, + ) + + response = await light_control.get_supported_versions() + + assert route.called + assert route.calls.last.request.method == "POST" + assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert json.loads(route.calls.last.request.content) == { + "method": "getSupportedVersions" + } + + assert response["data"] == {"apiVersions": ["1.1"]} + + +response_getLightInformation = { + "apiVersion": "1.1", + "method": "getLightInformation", + "data": { + "items": [ + { + "lightID": "led0", + "lightType": "IR", + "enabled": True, + "synchronizeDayNightMode": True, + "lightState": False, + "automaticIntensityMode": False, + "automaticAngleOfIlluminationMode": False, + "nrOfLEDs": 1, + "error": False, + "errorInfo": "", + } + ] + }, +} From 2a3a9cf38d7ae0e7a494adce3e7174a686f34ae8 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Thu, 27 May 2021 11:59:18 -0400 Subject: [PATCH 02/21] Updated tests and door_control --- axis/door_control.py | 4 +- tests/test_door_control.py | 677 +++++++++++++------------------------ 2 files changed, 233 insertions(+), 448 deletions(-) diff --git a/axis/door_control.py b/axis/door_control.py index 873cb336..6a1abf63 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -47,12 +47,12 @@ async def get_door_info_list(self) -> dict: json={"tdc:GetDoorInfoList": {}}, ) - async def get_door_info(self, door_token: str) -> dict: + async def get_door_info(self, door_tokens: list) -> dict: """List the door information.""" return await self._request( "post", URL, - json={"tdc:GetDoorInfo": {"Token": door_token}} + json={"tdc:GetDoorInfo": {"Token": door_tokens}} ) async def get_door_state(self, door_token: str) -> dict: diff --git a/tests/test_door_control.py b/tests/test_door_control.py index 82f1bbd1..a51edeb7 100644 --- a/tests/test_door_control.py +++ b/tests/test_door_control.py @@ -91,7 +91,7 @@ async def test_update(door_control): assert item2.id == "Axis-accc8ee70fbe:1614733258.565014000" assert item2.door_token == "Axis-accc8ee70fbe:1614733258.565014000" assert item2.door_name == "North Gym Door" - assert item2.description == "North Gym Door Description" + assert item2.door_description == "North Gym Door Description" # TODO: Check to see that comparing dict using "==" actually does what I want it to do assert item2.door_capabilities == {} @@ -126,579 +126,364 @@ async def test_get_service_capabilities(door_control): @respx.mock -async def test_get_light_information(light_control): - """Test get light information API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( +async def test_get_door_info_list(door_control): + """Test get door list.""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( json={ - "apiVersion": "1.1", - "method": "getLightInformation", - "data": { - "items": [ - { - "lightID": "led0", - "lightType": "IR", - "enabled": True, - "synchronizeDayNightMode": True, - "lightState": False, - "automaticIntensityMode": False, - "automaticAngleOfIlluminationMode": False, - "nrOfLEDs": 1, - "error": False, - "errorInfo": "", + "DoorInfo": [ + { + "token": "Axis-accc8ea9abac:1550808050.595717000", + "Name": "Main Door", + "Description": "", + "Capabilities": { + "Access": True, + "Lock": True, + "Unlock": True, + "Block": True, + "DoubleLock": True, + "LockDown": True, + "LockOpen": True, + "DoorMonitor": True, + "LockMonitor": False, + "DoubleLockMonitor": False, + "Alarm": True, + "Tamper": False, + "Warning": True, + "Configurable": True } - ] - }, - }, + }, + { + "token": "Axis-accc8ee70fbe:1614733258.565014000", + "Name": "North Gym Door", + "Description": "", + "Capabilities": {} + } + ] + } ) - response = await light_control.get_light_information() + response = await door_control.get_door_info_list() assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getLightInformation", - "apiVersion": "1.1", - "context": "Axis library", - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:GetDoorInfoList": {}} - assert response["data"] == { - "items": [ - { - "lightID": "led0", - "lightType": "IR", - "enabled": True, - "synchronizeDayNightMode": True, - "lightState": False, - "automaticIntensityMode": False, - "automaticAngleOfIlluminationMode": False, - "nrOfLEDs": 1, - "error": False, - "errorInfo": "", + assert response["DoorInfo"] == [ + { + "token": "Axis-accc8ea9abac:1550808050.595717000", + "Name": "Main Door", + "Description": "", + "Capabilities": { + "Access": True, + "Lock": True, + "Unlock": True, + "Block": True, + "DoubleLock": True, + "LockDown": True, + "LockOpen": True, + "DoorMonitor": True, + "LockMonitor": False, + "DoubleLockMonitor": False, + "Alarm": True, + "Tamper": False, + "Warning": True, + "Configurable": True } - ] - } - - -@respx.mock -async def test_activate_light(light_control): - """Test activating light API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "activateLight", - "data": {}, }, - ) - - await light_control.activate_light("led0") - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "activateLight", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + { + "token": "Axis-accc8ee70fbe:1614733258.565014000", + "Name": "North Gym Door", + "Description": "", + "Capabilities": {} + } + ] @respx.mock -async def test_deactivate_light(light_control): - """Test deactivating light API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( +async def test_get_door_info(door_control): + """Test get door info from token(s).""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( json={ - "apiVersion": "1.1", - "method": "deactivateLight", - "data": {}, - }, + "DoorInfo": [ + { + "token": "Axis-accc8ea9abac:1550808050.595717000", + "Name": "Main Door", + "Description": "", + "Capabilities": { + "Access": True, + "Lock": True, + "Unlock": True, + "Block": True, + "DoubleLock": True, + "LockDown": True, + "LockOpen": True, + "DoorMonitor": True, + "LockMonitor": False, + "DoubleLockMonitor": False, + "Alarm": True, + "Tamper": False, + "Warning": True, + "Configurable": True + } + } + ] + } ) - await light_control.deactivate_light("led0") + tokens = ["Axis-accc8ea9abac:1550808050.595717000"] + + response = await door_control.get_door_info(tokens) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert route.calls.last.request.url.path == "/vapix/doorcontrol" assert json.loads(route.calls.last.request.content) == { - "method": "deactivateLight", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, + "tdc:GetDoorInfo": {"Token": tokens} } + assert response["DoorInfo"] == [ + { + "token": "Axis-accc8ea9abac:1550808050.595717000", + "Name": "Main Door", + "Description": "", + "Capabilities": { + "Access": True, + "Lock": True, + "Unlock": True, + "Block": True, + "DoubleLock": True, + "LockDown": True, + "LockOpen": True, + "DoorMonitor": True, + "LockMonitor": False, + "DoubleLockMonitor": False, + "Alarm": True, + "Tamper": False, + "Warning": True, + "Configurable": True + } + } + ] + @respx.mock -async def test_enable_light(light_control): - """Test enabling light API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( +async def test_get_door_state(door_control): + """Test get door state(s).""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( json={ - "apiVersion": "1.1", - "method": "enableLight", - "data": {}, - }, + "DoorState": { + "DoorPhysicalState": "Closed", + "Alarm": "Normal", + "DoorMode": "Locked" + } + } ) - await light_control.enable_light("led0") + token = "Axis-accc8ea9abac:1550808050.595717000" + + response = await door_control.get_door_state(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" + assert route.calls.last.request.url.path == "/vapix/doorcontrol" assert json.loads(route.calls.last.request.content) == { - "method": "enableLight", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, + "tdc:GetDoorState": {"Token": token} } - -@respx.mock -async def test_disable_light(light_control): - """Test disabling light API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "disableLight", - "data": {}, - }, - ) - - await light_control.disable_light("led0") - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "disableLight", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, + assert response["DoorState"] == { + "DoorPhysicalState": "Closed", + "Alarm": "Normal", + "DoorMode": "Locked" } @respx.mock -async def test_get_light_status(light_control): - """Test get light status API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "getLightStatus", - "data": {"status": False}, - }, +async def test_access_door(door_control): + """Test access door """ + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - response = await light_control.get_light_status("led0") + token = "Axis-accc8ea9abac:1550808050.595717000" + + response = await door_control.access_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getLightStatus", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:AccessDoor": {"Token": token}} - assert response["data"] == {"status": False} + assert response == {} @respx.mock -async def test_set_automatic_intensity_mode(light_control): - """Test set automatic intensity mode API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "setAutomaticIntensityMode", - "data": {}, - }, +async def test_lock_door(door_control): + """Test lock door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - await light_control.set_automatic_intensity_mode("led0", True) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "setAutomaticIntensityMode", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0", "enabled": True}, - } + token = "Axis-accc8ea9abac:1550808050.595717000" - -@respx.mock -async def test_get_manual_intensity(light_control): - """Test get valid intensity API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "getManualIntensity", - "data": {"intensity": 1000}, - }, - ) - - response = await light_control.get_manual_intensity("led0") + response = await door_control.lock_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getManualIntensity", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:LockDoor": {"Token": token}} - assert response["data"] == {"intensity": 1000} + assert response == {} @respx.mock -async def test_set_manual_intensity(light_control): - """Test set manual intensity API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "setManualIntensity", - "data": {}, - }, +async def test_unlock_door(door_control): + """Test unlock door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - await light_control.set_manual_intensity("led0", 1000) + token = "Axis-accc8ea9abac:1550808050.595717000" - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "setManualIntensity", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0", "intensity": 1000}, - } - - -@respx.mock -async def test_get_valid_intensity(light_control): - """Test get valid intensity API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "getValidIntensity", - "data": {"ranges": [{"low": 0, "high": 1000}]}, - }, - ) - - response = await light_control.get_valid_intensity("led0") + response = await door_control.unlock_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getValidIntensity", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:UnlockDoor": {"Token": token}} - assert response["data"] == {"ranges": [{"low": 0, "high": 1000}]} + assert response == {} @respx.mock -async def test_set_individual_intensity(light_control): - """Test set individual intensity API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "setIndividualIntensity", - "data": {}, - }, +async def test_block_door(door_control): + """Test block door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - await light_control.set_individual_intensity("led0", 1, 1000) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "setIndividualIntensity", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0", "LEDID": 1, "intensity": 1000}, - } - - -@respx.mock -async def test_get_individual_intensity(light_control): - """Test get individual intensity API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "getIndividualIntensity", - "data": {"intensity": 1000}, - }, - ) + token = "Axis-accc8ea9abac:1550808050.595717000" - response = await light_control.get_individual_intensity("led0", 1) + response = await door_control.block_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getIndividualIntensity", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0", "LEDID": 1}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:BlockDoor": {"Token": token}} - assert response["data"] == {"intensity": 1000} + assert response == {} @respx.mock -async def test_get_current_intensity(light_control): - """Test get current intensity API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "getCurrentIntensity", - "data": {"intensity": 1000}, - }, +async def test_lock_down_door(door_control): + """Test lock down door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - response = await light_control.get_current_intensity("led0") + token = "Axis-accc8ea9abac:1550808050.595717000" + + response = await door_control.lock_down_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getCurrentIntensity", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:LockDownDoor": {"Token": token}} - assert response["data"] == {"intensity": 1000} + assert response == {} @respx.mock -async def test_set_automatic_angle_of_illumination_mode(light_control): - """Test set automatic angle of illumination mode API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "setAutomaticAngleOfIlluminationMode", - "data": {}, - }, +async def test_lock_down_release_door(door_control): + """Test lock down release door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - await light_control.set_automatic_angle_of_illumination_mode("led0", True) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "setAutomaticAngleOfIlluminationMode", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0", "enabled": True}, - } - + token = "Axis-accc8ea9abac:1550808050.595717000" -@respx.mock -async def test_get_valid_angle_of_illumination(light_control): - """Test get valid angle of illumination API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.0", - "context": "my context", - "method": "getValidAngleOfIllumination", - "data": {"ranges": [{"low": 10, "high": 30}, {"low": 20, "high": 50}]}, - }, - ) - - response = await light_control.get_valid_angle_of_illumination("led0") + response = await door_control.lock_down_release_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getValidAngleOfIllumination", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:LockDownReleaseDoor": {"Token": token}} - assert response["data"] == { - "ranges": [{"low": 10, "high": 30}, {"low": 20, "high": 50}] - } + assert response == {} @respx.mock -async def test_set_manual_angle_of_illumination(light_control): - """Test set manual angle of illumination API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "setManualAngleOfIllumination", - "data": {}, - }, +async def test_lock_open_door(door_control): + """Test lock open door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - await light_control.set_manual_angle_of_illumination("led0", 30) + token = "Axis-accc8ea9abac:1550808050.595717000" - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "setManualAngleOfIllumination", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0", "angleOfIllumination": 30}, - } - - -@respx.mock -async def test_get_manual_angle_of_illumination(light_control): - """Test get manual angle of illumination API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.0", - "context": "my context", - "method": "getManualAngleOfIllumination", - "data": {"angleOfIllumination": 30}, - }, - ) - - response = await light_control.get_manual_angle_of_illumination("led0") + response = await door_control.lock_open_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getManualAngleOfIllumination", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:LockOpenDoor": {"Token": token}} - assert response["data"] == {"angleOfIllumination": 30} + assert response == {} @respx.mock -async def test_get_current_angle_of_illumination(light_control): - """Test get current angle of illumination API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.0", - "context": "my context", - "method": "getCurrentAngleOfIllumination", - "data": {"angleOfIllumination": 20}, - }, +async def test_lock_open_release_door(door_control): + """Test lock open release door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - response = await light_control.get_current_angle_of_illumination("led0") + token = "Axis-accc8ea9abac:1550808050.595717000" + + response = await door_control.lock_open_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getCurrentAngleOfIllumination", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:LockOpenReleaseDoor": {"Token": token}} - assert response["data"] == {"angleOfIllumination": 20} + assert response == {} @respx.mock -async def test_set_light_synchronization_day_night_mode(light_control): - """Test set light synchronization day night mode API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "method": "setLightSynchronizationDayNightMode", - "data": {}, - }, +async def test_double_lock_door(door_control): + """Test double lock door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - await light_control.set_light_synchronization_day_night_mode("led0", True) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "setLightSynchronizationDayNightMode", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0", "enabled": True}, - } - - -@respx.mock -async def test_get_light_synchronization_day_night_mode(light_control): - """Test get light synchronization day night mode API.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "apiVersion": "1.1", - "context": "my context", - "method": "getLightSynchronizeDayNightMode", - "data": {"enabled": True}, - }, - ) + token = "Axis-accc8ea9abac:1550808050.595717000" - response = await light_control.get_light_synchronization_day_night_mode("led0") + response = await door_control.double_lock_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getLightSynchronizationDayNightMode", - "apiVersion": "1.1", - "context": "Axis library", - "params": {"lightID": "led0"}, - } + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:DoubleLockDoor": {"Token": token}} - assert response["data"] == {"enabled": True} + assert response == {} @respx.mock -async def test_get_supported_versions(light_control): - """Test get supported versions api.""" - route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( - json={ - "method": "getSupportedVersions", - "data": {"apiVersions": ["1.1"]}, - }, +async def test_release_door(door_control): + """Test release door""" + route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( + json={} ) - response = await light_control.get_supported_versions() + token = "Axis-accc8ea9abac:1550808050.595717000" + + response = await door_control.release_door(token) assert route.called assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" - assert json.loads(route.calls.last.request.content) == { - "method": "getSupportedVersions" - } - - assert response["data"] == {"apiVersions": ["1.1"]} + assert route.calls.last.request.url.path == "/vapix/doorcontrol" + assert json.loads(route.calls.last.request.content) == {"tdc:ReleaseDoor": {"Token": token}} + assert response == {} -response_getLightInformation = { - "apiVersion": "1.1", - "method": "getLightInformation", - "data": { - "items": [ - { - "lightID": "led0", - "lightType": "IR", - "enabled": True, - "synchronizeDayNightMode": True, - "lightState": False, - "automaticIntensityMode": False, - "automaticAngleOfIlluminationMode": False, - "nrOfLEDs": 1, - "error": False, - "errorInfo": "", - } - ] - }, -} From a03bd912833c9a1e5e579ddd370b74c26c0bb13d Mon Sep 17 00:00:00 2001 From: drewclauson Date: Fri, 28 May 2021 16:23:54 -0400 Subject: [PATCH 03/21] Removed - not needed --- axis/param_cgi.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/axis/param_cgi.py b/axis/param_cgi.py index 385834c4..29c19663 100644 --- a/axis/param_cgi.py +++ b/axis/param_cgi.py @@ -343,11 +343,6 @@ def light_control(self) -> bool: """Support light control.""" return self.get(PROPERTIES, {}).get("LightControl.LightControl2") == "yes" - @property - def door_control(self) -> bool: - """FIXME: Figure out a different way to determine if door control is supported """ - return self.get(BRAND, {}).get("ProdNbr") == "A1001" - @property def ptz(self) -> bool: """Support PTZ control.""" From 776c8bf81773f3e784c0f65f1d8ec8ad9cd57f2c Mon Sep 17 00:00:00 2001 From: drewclauson Date: Thu, 3 Jun 2021 09:20:50 -0400 Subject: [PATCH 04/21] Add capabilities globals --- axis/door_control.py | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/axis/door_control.py b/axis/door_control.py index 6a1abf63..b62f30cc 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -13,6 +13,37 @@ API_DISCOVERY_ID = "door-control" API_VERSION = "1.0" +CAPABILITY_ACCESS = "access" +CAPABILITY_LOCK = "lock" +CAPABILITY_UNLOCK = "unlock" +CAPABILITY_BLOCK = "block" +CAPABILITY_DOUBLE_LOCK = "doublelock" +CAPABILITY_LOCK_DOWN = "lockdown" +CAPABILITY_LOCK_OPEN = "lockopen" +CAPABILITY_DOOR_MONITOR = "doormonitor" +CAPABILITY_LOCK_MONITOR = "lockmonitor" +CAPABILITY_DOUBLE_LOCK_MONITOR = "doublelockmonitor" +CAPABILITY_ALARM = "alarm" +CAPABILITY_TAMPER = "tamper" +CAPABILITY_WARNING = "warning" +CAPABILITY_CONFIGURABLE = "configurable" + +SUPPORTED_CAPABILITIES = ( + CAPABILITY_ACCESS, + CAPABILITY_LOCK, + CAPABILITY_UNLOCK, + CAPABILITY_BLOCK, + CAPABILITY_DOUBLE_LOCK, + CAPABILITY_LOCK_DOWN, + CAPABILITY_LOCK_OPEN, + CAPABILITY_DOOR_MONITOR, + CAPABILITY_LOCK_MONITOR, + CAPABILITY_DOUBLE_LOCK_MONITOR, + CAPABILITY_ALARM, + CAPABILITY_TAMPER, + CAPABILITY_WARNING, + CAPABILITY_CONFIGURABLE +) class DoorControl(APIItems): """Door control for Axis devices.""" @@ -169,3 +200,5 @@ def door_description(self) -> str: def door_capabilities(self) -> dict: """Capabilities of Door.""" return self.raw["Capabilities"] + +#TODO: add method to show what this supports? \ No newline at end of file From bd92b6da59e5bd6d86536193a27c5cf535296d5c Mon Sep 17 00:00:00 2001 From: drewclauson Date: Fri, 4 Jun 2021 14:59:57 -0400 Subject: [PATCH 05/21] Fix formatting issues --- axis/door_control.py | 7 +++---- tests/test_door_control.py | 1 - 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/axis/door_control.py b/axis/door_control.py index b62f30cc..7dab38dd 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -4,9 +4,7 @@ of physical access controls in the Axis devices (e.g. A1001, A1601) """ -import attr - -from .api import APIItem, APIItems, Body +from .api import APIItem, APIItems URL = "/vapix/doorcontrol" @@ -45,6 +43,7 @@ CAPABILITY_CONFIGURABLE ) + class DoorControl(APIItems): """Door control for Axis devices.""" @@ -201,4 +200,4 @@ def door_capabilities(self) -> dict: """Capabilities of Door.""" return self.raw["Capabilities"] -#TODO: add method to show what this supports? \ No newline at end of file +# TODO: add method to show what this supports? diff --git a/tests/test_door_control.py b/tests/test_door_control.py index a51edeb7..462ab5ae 100644 --- a/tests/test_door_control.py +++ b/tests/test_door_control.py @@ -486,4 +486,3 @@ async def test_release_door(door_control): assert json.loads(route.calls.last.request.content) == {"tdc:ReleaseDoor": {"Token": token}} assert response == {} - From 01f4733273a649c5bf8ac436fa99d5249977dde7 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Mon, 7 Jun 2021 23:33:17 -0400 Subject: [PATCH 06/21] Docs update --- axis/door_control.py | 67 +++++++++++++++++++++++++++++--------------- 1 file changed, 44 insertions(+), 23 deletions(-) diff --git a/axis/door_control.py b/axis/door_control.py index 7dab38dd..7386ca37 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -11,20 +11,20 @@ API_DISCOVERY_ID = "door-control" API_VERSION = "1.0" -CAPABILITY_ACCESS = "access" -CAPABILITY_LOCK = "lock" -CAPABILITY_UNLOCK = "unlock" -CAPABILITY_BLOCK = "block" -CAPABILITY_DOUBLE_LOCK = "doublelock" -CAPABILITY_LOCK_DOWN = "lockdown" -CAPABILITY_LOCK_OPEN = "lockopen" -CAPABILITY_DOOR_MONITOR = "doormonitor" -CAPABILITY_LOCK_MONITOR = "lockmonitor" -CAPABILITY_DOUBLE_LOCK_MONITOR = "doublelockmonitor" -CAPABILITY_ALARM = "alarm" -CAPABILITY_TAMPER = "tamper" -CAPABILITY_WARNING = "warning" -CAPABILITY_CONFIGURABLE = "configurable" +CAPABILITY_ACCESS = "Access" +CAPABILITY_LOCK = "Lock" +CAPABILITY_UNLOCK = "Unlock" +CAPABILITY_BLOCK = "Block" +CAPABILITY_DOUBLE_LOCK = "DoubleLock" +CAPABILITY_LOCK_DOWN = "LockDown" +CAPABILITY_LOCK_OPEN = "LockOpen" +CAPABILITY_DOOR_MONITOR = "DoorMonitor" +CAPABILITY_LOCK_MONITOR = "LockMonitor" +CAPABILITY_DOUBLE_LOCK_MONITOR = "DoubleLockMonitor" +CAPABILITY_ALARM = "Alarm" +CAPABILITY_TAMPER = "Tamper" +CAPABILITY_WARNING = "Warning" +CAPABILITY_CONFIGURABLE = "Configurable" SUPPORTED_CAPABILITIES = ( CAPABILITY_ACCESS, @@ -95,8 +95,12 @@ async def get_door_state(self, door_token: str) -> dict: # region Door Actions async def access_door(self, door_token: str) -> None: - """Access the door. Use when a credential holder is granted access, - for example by swiping a card in a card reader.""" + """This operation allows momentarily accessing a Door. It invokes the functionality typically used when a card holder + presents a card to a card reader at the door and is granted access. + The DoorMode shall change to Accessed. + The Door shall remain accessible for the defined time as configured in the device. + When the time span elapses, the DoorMode shall change back to its previous state. + A device must have the Lock capability to utilize this method.""" await self._request( "post", URL, @@ -104,7 +108,8 @@ async def access_door(self, door_token: str) -> None: ) async def lock_door(self, door_token: str) -> None: - """Lock the door.""" + """This operation allows locking a Door. The DoorMode shall change to Locked. + A device must have the Lock capability to utilize this method.""" await self._request( "post", URL, @@ -112,7 +117,9 @@ async def lock_door(self, door_token: str) -> None: ) async def unlock_door(self, door_token: str) -> None: - """Unlock the door until it is explicitly locked again.""" + """This operation allows unlocking a Door until it is explicitly locked again. + The DoorMode shall change to Unlocked. + A device must have the Unlock capability to utilize this method.""" await self._request( "post", URL, @@ -120,7 +127,9 @@ async def unlock_door(self, door_token: str) -> None: ) async def block_door(self, door_token: str) -> None: - """Block the door.""" + """This operation allows blocking a Door and preventing momentary access (AccessDoor command). + The DoorMode shall change to Blocked. + A device must have the Block capability to utilize this method.""" await self._request( "post", URL, @@ -128,7 +137,10 @@ async def block_door(self, door_token: str) -> None: ) async def lock_down_door(self, door_token: str) -> None: - """Lock the door and prevent all other commands until a LockDownReleaseDoor command is sent.""" + """This operation allows locking and preventing other actions until a LockDownReleaseDoor command is invoked. + The DoorMode shall change to LockedDown. + The device shall ignore other door control commands until a LockDownReleaseDoor command is performed. + A device must have the LockDown capability to utilize this method.""" await self._request( "post", URL, @@ -136,7 +148,10 @@ async def lock_down_door(self, door_token: str) -> None: ) async def lock_down_release_door(self, door_token: str) -> None: - """Release the door from the LockedDown state.""" + """This operation allows releasing the LockedDown state of a Door. + The DoorMode shall change back to its previous/next state. + It is not defined what the previous/next state shall be, but typically - Locked. + This method will only succeed if the current DoorMode is LockedDown.""" await self._request( "post", URL, @@ -144,7 +159,10 @@ async def lock_down_release_door(self, door_token: str) -> None: ) async def lock_open_door(self, door_token: str) -> None: - """Unlock the door and prevent all other commands until a LockOpenReleaseDoor command is sent.""" + """This operation allows unlocking a Door and preventing other actions until LockOpenReleaseDoor method is invoked. + The DoorMode shall change to LockedOpen. + The device shall ignore other door control commands until a LockOpenReleaseDoor command is performed. + A device must have the LockOpen capability to utilize this method.""" await self._request( "post", URL, @@ -152,7 +170,10 @@ async def lock_open_door(self, door_token: str) -> None: ) async def lock_open_release_door(self, door_token: str) -> None: - """Release the door from the LockedOpen state.""" + """This operation allows releasing the LockedOpen state of a Door. + The DoorMode shall change state from the LockedOpen state back to its previous/next state. + It is not defined what the previous/next state shall be, but typically - Unlocked. + This method shall only succeed if the current DoorMode is LockedOpen.""" await self._request( "post", URL, From 6ba6f11361b4a88e96e64bf8193ef3ec4231d28c Mon Sep 17 00:00:00 2001 From: drewclauson Date: Mon, 7 Jun 2021 23:33:36 -0400 Subject: [PATCH 07/21] Quick pass at a check for capabilities --- axis/door_control.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/axis/door_control.py b/axis/door_control.py index 7386ca37..aeab83e2 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -221,4 +221,8 @@ def door_capabilities(self) -> dict: """Capabilities of Door.""" return self.raw["Capabilities"] -# TODO: add method to show what this supports? + async def is_capable_of(self, capability: str) -> bool: + """Retrieve whether door has the specified capability.""" + if capability not in SUPPORTED_CAPABILITIES: + return False + return self.raw["Capabilities"][capability] \ No newline at end of file From 8762d20110781f2b48a3bab7802f522cb697be1d Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 14:41:20 -0400 Subject: [PATCH 08/21] Add door events --- axis/event_stream.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/axis/event_stream.py b/axis/event_stream.py index 195a2c0c..27f44045 100644 --- a/axis/event_stream.py +++ b/axis/event_stream.py @@ -9,6 +9,7 @@ LOGGER = logging.getLogger(__name__) +CLASS_DOOR = "door" CLASS_INPUT = "input" CLASS_LIGHT = "light" CLASS_MOTION = "motion" @@ -188,6 +189,22 @@ class DayNight(AxisBinaryEvent): TYPE = "DayNight" +class DoorMode(AxisBinaryEvent): + """Door Mode Changes.""" + + TOPIC = "tns1:Door/State/DoorMode" + CLASS = CLASS_DOOR + TYPE = "DoorMode" + + +class DoorPhysicalState(AxisBinaryEvent): + """Door Physical State Changes.""" + + TOPIC = "tns1:Door/State/DoorPhysicalState" + CLASS = CLASS_DOOR + TYPE = "DoorPhysicalState" + + class FenceGuard(AxisBinaryEvent): """Fence Guard trigger event.""" @@ -338,6 +355,8 @@ def id(self) -> str: EVENT_CLASSES = ( Audio, DayNight, + DoorMode, + DoorPhysicalState, FenceGuard, Input, Light, From ecca63829f4032e5fe609e94c95c2bf1a0bff29c Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 14:41:39 -0400 Subject: [PATCH 09/21] Remove initialization --- axis/vapix.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/axis/vapix.py b/axis/vapix.py index 2ca89f34..a4dedb04 100644 --- a/axis/vapix.py +++ b/axis/vapix.py @@ -184,9 +184,6 @@ async def initialize_param_cgi(self, preload_data: bool = True) -> None: if not self.light_control and self.params.light_control: await self._initialize_api_attribute(LightControl, "light_control") - if not self.door_control and self.params.door_control: - await self._initialize_api_attribute(DoorControl, "door_control") - if not self.ports: self.ports = Ports(self.params, self.request) From 4a4c1fa8060df2ef983bf4cf4e521212d40f435a Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 15:00:47 -0400 Subject: [PATCH 10/21] Fix assignment --- axis/vapix.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/axis/vapix.py b/axis/vapix.py index a4dedb04..91750045 100644 --- a/axis/vapix.py +++ b/axis/vapix.py @@ -21,7 +21,7 @@ from .applications.vmd4 import Vmd4 from .basic_device_info import API_DISCOVERY_ID as BASIC_DEVICE_INFO_ID, BasicDeviceInfo from .configuration import Configuration -from .door_control import DoorControl, API_DISCOVERY_ID as DOOR_CONTROL_ID +from .door_control import DoorControl from .errors import PathNotFound, RequestError, Unauthorized, raise_error from .event_instances import EventInstances from .light_control import API_DISCOVERY_ID as LIGHT_CONTROL_ID, LightControl @@ -52,7 +52,7 @@ def __init__(self, config: Configuration) -> None: self.applications: Optional[Applications] = None self.basic_device_info: Optional[BasicDeviceInfo] = None self.event_instances: Optional[EventInstances] = None - self.door_control = Optional[DoorControl] = None + self.door_control: Optional[DoorControl] = None self.fence_guard: Optional[FenceGuard] = None self.light_control: Optional[LightControl] = None self.loitering_guard: Optional[LoiteringGuard] = None From 31135451957ed9aae1bf2c89c36c9f9c2c119f7a Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 15:00:57 -0400 Subject: [PATCH 11/21] Remove global --- axis/door_control.py | 1 - 1 file changed, 1 deletion(-) diff --git a/axis/door_control.py b/axis/door_control.py index aeab83e2..1da23b90 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -9,7 +9,6 @@ URL = "/vapix/doorcontrol" API_DISCOVERY_ID = "door-control" -API_VERSION = "1.0" CAPABILITY_ACCESS = "Access" CAPABILITY_LOCK = "Lock" From 02d7ea1fa5ff9cf957df092691787a02fc0a11f4 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 16:09:43 -0400 Subject: [PATCH 12/21] Add events --- axis/event_stream.py | 62 +++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 58 insertions(+), 4 deletions(-) diff --git a/axis/event_stream.py b/axis/event_stream.py index 27f44045..7204102b 100644 --- a/axis/event_stream.py +++ b/axis/event_stream.py @@ -189,20 +189,68 @@ class DayNight(AxisBinaryEvent): TYPE = "DayNight" +class DoorAlarm(AxisBinaryEvent): + """Door Alarm Changes.""" + + TOPIC = "tns1:Door/State/DoorAlarm" + CLASS = CLASS_DOOR + TYPE = "Door Alarm" + + +class DoorFault(AxisBinaryEvent): + """Door Fault Changes.""" + + TOPIC = "tns1:Door/State/DoorFault" + CLASS = CLASS_DOOR + TYPE = "Door Falt" + + +class DoorLockPhysical(AxisBinaryEvent): + """Door Lock State Changes.""" + + TOPIC = "tns1:Door/State/LockPhysicalState" + CLASS = CLASS_DOOR + TYPE = "Door Lock Physical State" + + class DoorMode(AxisBinaryEvent): """Door Mode Changes.""" TOPIC = "tns1:Door/State/DoorMode" CLASS = CLASS_DOOR - TYPE = "DoorMode" + TYPE = "Door Mode" -class DoorPhysicalState(AxisBinaryEvent): +class DoorPhysical(AxisBinaryEvent): """Door Physical State Changes.""" TOPIC = "tns1:Door/State/DoorPhysicalState" CLASS = CLASS_DOOR - TYPE = "DoorPhysicalState" + TYPE = "Door Physical State" + + +class DoorTamper(AxisBinaryEvent): + """Door Tamper State Changes.""" + + TOPIC = "tns1:Door/State/DoorTamper" + CLASS = CLASS_DOOR + TYPE = "Door Tamper State" + + +class DoorWarning(AxisBinaryEvent): + """Door Warning State Changes.""" + + TOPIC = "tns1:Door/State/DoorWarning" + CLASS = CLASS_DOOR + TYPE = "Door Warning State" + + +class DoorDoubleLockPhysical(AxisBinaryEvent): + """Door Double Lock State Changes.""" + + TOPIC = "tns1:Door/State/DoubleLockPhysicalState" + CLASS = CLASS_DOOR + TYPE = "Door Double Lock Physical State" class FenceGuard(AxisBinaryEvent): @@ -355,8 +403,14 @@ def id(self) -> str: EVENT_CLASSES = ( Audio, DayNight, + DoorAlarm, + DoorDoubleLockPhysical, + DoorFault, + DoorLockPhysical, DoorMode, - DoorPhysicalState, + DoorPhysical, + DoorWarning, + DoorTamper, FenceGuard, Input, Light, From b3bf6f6f63287ce085170d70db6dddcc50d2d719 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 16:24:32 -0400 Subject: [PATCH 13/21] Fix Typo --- axis/event_stream.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/axis/event_stream.py b/axis/event_stream.py index 7204102b..fdb2927c 100644 --- a/axis/event_stream.py +++ b/axis/event_stream.py @@ -202,7 +202,7 @@ class DoorFault(AxisBinaryEvent): TOPIC = "tns1:Door/State/DoorFault" CLASS = CLASS_DOOR - TYPE = "Door Falt" + TYPE = "Door Fault" class DoorLockPhysical(AxisBinaryEvent): From d133c1bd69c423c8d02adcb65f1eda4a422630bf Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 16:27:31 -0400 Subject: [PATCH 14/21] Add API_DISCOVERY_ID back in --- axis/vapix.py | 56 +++++++++++++++++++++++++-------------------------- 1 file changed, 28 insertions(+), 28 deletions(-) diff --git a/axis/vapix.py b/axis/vapix.py index 91750045..3cec45c9 100644 --- a/axis/vapix.py +++ b/axis/vapix.py @@ -21,7 +21,7 @@ from .applications.vmd4 import Vmd4 from .basic_device_info import API_DISCOVERY_ID as BASIC_DEVICE_INFO_ID, BasicDeviceInfo from .configuration import Configuration -from .door_control import DoorControl +from .door_control import DoorControl, API_DISCOVERY_ID as DOOR_CONTROL_ID from .errors import PathNotFound, RequestError, Unauthorized, raise_error from .event_instances import EventInstances from .light_control import API_DISCOVERY_ID as LIGHT_CONTROL_ID, LightControl @@ -117,7 +117,7 @@ async def initialize(self) -> None: await self.initialize_applications() async def _initialize_api_attribute( - self, api_class: Callable, api_attr: str + self, api_class: Callable, api_attr: str ) -> None: """Initialize API and load data.""" api_instance = api_class(self.request) @@ -139,13 +139,13 @@ async def initialize_api_discovery(self) -> None: tasks = [] for api_id, api_class, api_attr in ( - (BASIC_DEVICE_INFO_ID, BasicDeviceInfo, "basic_device_info"), - (IO_PORT_MANAGEMENT_ID, IoPortManagement, "ports"), - (DOOR_CONTROL_ID, DoorControl, "door_control"), - (LIGHT_CONTROL_ID, LightControl, "light_control"), - (MQTT_ID, MqttClient, "mqtt"), - (STREAM_PROFILES_ID, StreamProfiles, "stream_profiles"), - (VIEW_AREAS_ID, ViewAreas, "view_areas"), + (BASIC_DEVICE_INFO_ID, BasicDeviceInfo, "basic_device_info"), + (IO_PORT_MANAGEMENT_ID, IoPortManagement, "ports"), + (DOOR_CONTROL_ID, DoorControl, "door_control"), + (LIGHT_CONTROL_ID, LightControl, "light_control"), + (MQTT_ID, MqttClient, "mqtt"), + (STREAM_PROFILES_ID, StreamProfiles, "stream_profiles"), + (VIEW_AREAS_ID, ViewAreas, "view_areas"), ): if api_id in self.api_discovery: tasks.append(self._initialize_api_attribute(api_class, api_attr)) @@ -194,7 +194,7 @@ async def initialize_applications(self) -> None: """Load data for applications on device.""" self.applications = Applications(self.request) if self.params and version.parse( - self.params.embedded_development + self.params.embedded_development ) >= version.parse(APPLICATIONS_MINIMUM_VERSION): try: await self.applications.update() @@ -204,16 +204,16 @@ async def initialize_applications(self) -> None: tasks = [] for app_class, app_attr in ( - (FenceGuard, "fence_guard"), - (LoiteringGuard, "loitering_guard"), - (MotionGuard, "motion_guard"), - (ObjectAnalytics, "object_analytics"), - (Vmd4, "vmd4"), + (FenceGuard, "fence_guard"), + (LoiteringGuard, "loitering_guard"), + (MotionGuard, "motion_guard"), + (ObjectAnalytics, "object_analytics"), + (Vmd4, "vmd4"), ): if ( - app_class.APPLICATION_NAME in self.applications # type: ignore[attr-defined] - and self.applications[app_class.APPLICATION_NAME].status # type: ignore[attr-defined] - == APPLICATION_STATE_RUNNING + app_class.APPLICATION_NAME in self.applications # type: ignore[attr-defined] + and self.applications[app_class.APPLICATION_NAME].status # type: ignore[attr-defined] + == APPLICATION_STATE_RUNNING ): tasks.append(self._initialize_api_attribute(app_class, app_attr)) @@ -240,11 +240,11 @@ async def load_user_groups(self) -> None: if self.users and self.config.username in self.users: user = self.users[self.config.username] user_groups = ( - f"{user.name}\n" # type: ignore[attr-defined] - + ("admin " if user.admin else "") # type: ignore[attr-defined] - + ("operator " if user.operator else "") # type: ignore[attr-defined] - + ("viewer " if user.viewer else "") # type: ignore[attr-defined] - + ("ptz" if user.ptz else "") # type: ignore[attr-defined] + f"{user.name}\n" # type: ignore[attr-defined] + + ("admin " if user.admin else "") # type: ignore[attr-defined] + + ("operator " if user.operator else "") # type: ignore[attr-defined] + + ("viewer " if user.viewer else "") # type: ignore[attr-defined] + + ("ptz" if user.ptz else "") # type: ignore[attr-defined] ) else: @@ -256,11 +256,11 @@ async def load_user_groups(self) -> None: self.user_groups = UserGroups(user_groups, self.request) async def request( - self, - method: str, - path: str, - kwargs_xmltodict: Optional[dict] = None, - **kwargs: dict, + self, + method: str, + path: str, + kwargs_xmltodict: Optional[dict] = None, + **kwargs: dict, ) -> Union[dict, str]: """Make a request to the API.""" url = self.config.url + path From fed1fdfd71815f42299a95aaa78088d253209103 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 16:28:04 -0400 Subject: [PATCH 15/21] Add newline at end of file --- axis/door_control.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/axis/door_control.py b/axis/door_control.py index 1da23b90..71237179 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -224,4 +224,4 @@ async def is_capable_of(self, capability: str) -> bool: """Retrieve whether door has the specified capability.""" if capability not in SUPPORTED_CAPABILITIES: return False - return self.raw["Capabilities"][capability] \ No newline at end of file + return self.raw["Capabilities"][capability] From 98d2a758b6d08d347cbc6793d75c1f218a212304 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 22:11:14 -0400 Subject: [PATCH 16/21] No more flake8 complaining. --- axis/door_control.py | 55 +++++++++++++++++++++++++++++--------------- 1 file changed, 37 insertions(+), 18 deletions(-) diff --git a/axis/door_control.py b/axis/door_control.py index 71237179..7914faf7 100644 --- a/axis/door_control.py +++ b/axis/door_control.py @@ -47,10 +47,12 @@ class DoorControl(APIItems): """Door control for Axis devices.""" def __init__(self, request: object) -> None: + """Initialize door control manager.""" super().__init__({}, request, URL, Door) # TODO: Question: Is this used to get status information for the door? Or to update the object properties? async def update(self) -> None: + """Refresh data.""" raw = await self.get_door_info_list() self.process_raw(raw) @@ -94,12 +96,14 @@ async def get_door_state(self, door_token: str) -> dict: # region Door Actions async def access_door(self, door_token: str) -> None: - """This operation allows momentarily accessing a Door. It invokes the functionality typically used when a card holder - presents a card to a card reader at the door and is granted access. + """Access a Door. + + It invokes the functionality typically used when a card holder presents a card to a card reader at the door and is granted access. The DoorMode shall change to Accessed. The Door shall remain accessible for the defined time as configured in the device. When the time span elapses, the DoorMode shall change back to its previous state. - A device must have the Lock capability to utilize this method.""" + A device must have the Lock capability to utilize this method. + """ await self._request( "post", URL, @@ -107,8 +111,11 @@ async def access_door(self, door_token: str) -> None: ) async def lock_door(self, door_token: str) -> None: - """This operation allows locking a Door. The DoorMode shall change to Locked. - A device must have the Lock capability to utilize this method.""" + """Lock a Door. + + The DoorMode shall change to Locked. + A device must have the Lock capability to utilize this method. + """ await self._request( "post", URL, @@ -116,9 +123,11 @@ async def lock_door(self, door_token: str) -> None: ) async def unlock_door(self, door_token: str) -> None: - """This operation allows unlocking a Door until it is explicitly locked again. + """Unlock a Door until it is explicitly locked again. + The DoorMode shall change to Unlocked. - A device must have the Unlock capability to utilize this method.""" + A device must have the Unlock capability to utilize this method. + """ await self._request( "post", URL, @@ -126,9 +135,11 @@ async def unlock_door(self, door_token: str) -> None: ) async def block_door(self, door_token: str) -> None: - """This operation allows blocking a Door and preventing momentary access (AccessDoor command). + """Block a Door and prevent momentary access (AccessDoor command). + The DoorMode shall change to Blocked. - A device must have the Block capability to utilize this method.""" + A device must have the Block capability to utilize this method. + """ await self._request( "post", URL, @@ -136,10 +147,12 @@ async def block_door(self, door_token: str) -> None: ) async def lock_down_door(self, door_token: str) -> None: - """This operation allows locking and preventing other actions until a LockDownReleaseDoor command is invoked. + """Locks down a door and prevents other actions until a LockDownReleaseDoor command is invoked. + The DoorMode shall change to LockedDown. The device shall ignore other door control commands until a LockDownReleaseDoor command is performed. - A device must have the LockDown capability to utilize this method.""" + A device must have the LockDown capability to utilize this method. + """ await self._request( "post", URL, @@ -147,10 +160,12 @@ async def lock_down_door(self, door_token: str) -> None: ) async def lock_down_release_door(self, door_token: str) -> None: - """This operation allows releasing the LockedDown state of a Door. + """Releases the LockedDown state of a Door. + The DoorMode shall change back to its previous/next state. It is not defined what the previous/next state shall be, but typically - Locked. - This method will only succeed if the current DoorMode is LockedDown.""" + This method will only succeed if the current DoorMode is LockedDown. + """ await self._request( "post", URL, @@ -158,10 +173,12 @@ async def lock_down_release_door(self, door_token: str) -> None: ) async def lock_open_door(self, door_token: str) -> None: - """This operation allows unlocking a Door and preventing other actions until LockOpenReleaseDoor method is invoked. + """Unlocks a Door and prevents other actions until LockOpenReleaseDoor method is invoked. + The DoorMode shall change to LockedOpen. The device shall ignore other door control commands until a LockOpenReleaseDoor command is performed. - A device must have the LockOpen capability to utilize this method.""" + A device must have the LockOpen capability to utilize this method. + """ await self._request( "post", URL, @@ -169,10 +186,12 @@ async def lock_open_door(self, door_token: str) -> None: ) async def lock_open_release_door(self, door_token: str) -> None: - """This operation allows releasing the LockedOpen state of a Door. + """Releases the LockedOpen state of a Door. + The DoorMode shall change state from the LockedOpen state back to its previous/next state. It is not defined what the previous/next state shall be, but typically - Unlocked. - This method shall only succeed if the current DoorMode is LockedOpen.""" + This method shall only succeed if the current DoorMode is LockedOpen. + """ await self._request( "post", URL, @@ -212,7 +231,7 @@ def door_name(self) -> str: @property def door_description(self) -> str: - """Description of Door.""" + """Door Description.""" return self.raw["Description"] @property From 5fa502631b6799d5ee2c26e613419afac57a39c2 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 23:01:06 -0400 Subject: [PATCH 17/21] Parameterize tests! --- tests/test_door_control.py | 194 ++++--------------------------------- 1 file changed, 18 insertions(+), 176 deletions(-) diff --git a/tests/test_door_control.py b/tests/test_door_control.py index 462ab5ae..99d504f0 100644 --- a/tests/test_door_control.py +++ b/tests/test_door_control.py @@ -298,191 +298,33 @@ async def test_get_door_state(door_control): } +@pytest.mark.parametrize( + "input,expected", [ + ({"api_function": "tdc:AccessDoor", "method_name": "access_door"}, None), + ({"api_function": "tdc:LockDoor", "method_name": "lock_door"}, None), + ({"api_function": "tdc:UnlockDoor", "method_name": "unlock_door"}, None), + ({"api_function": "tdc:BlockDoor", "method_name": "block_door"}, None), + ({"api_function": "tdc:LockDownDoor", "method_name": "lock_down_door"}, None), + ({"api_function": "tdc:LockDownReleaseDoor", "method_name": "lock_down_release_door"}, None), + ({"api_function": "tdc:LockOpenDoor", "method_name": "lock_open_door"}, None), + ({"api_function": "tdc:LockOpenReleaseDoor", "method_name": "lock_open_release_door"}, None), + ({"api_function": "tdc:DoubleLockDoor", "method_name": "double_lock_door"}, None), + ({"api_function": "axtdc:ReleaseDoor", "method_name": "release_door"}, None), + ] +) @respx.mock -async def test_access_door(door_control): - """Test access door """ - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.access_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:AccessDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_lock_door(door_control): - """Test lock door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.lock_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:LockDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_unlock_door(door_control): - """Test unlock door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.unlock_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:UnlockDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_block_door(door_control): - """Test block door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.block_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:BlockDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_lock_down_door(door_control): - """Test lock down door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.lock_down_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:LockDownDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_lock_down_release_door(door_control): - """Test lock down release door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.lock_down_release_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:LockDownReleaseDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_lock_open_door(door_control): - """Test lock open door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.lock_open_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:LockOpenDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_lock_open_release_door(door_control): - """Test lock open release door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.lock_open_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:LockOpenReleaseDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_double_lock_door(door_control): - """Test double lock door""" - route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( - json={} - ) - - token = "Axis-accc8ea9abac:1550808050.595717000" - - response = await door_control.double_lock_door(token) - - assert route.called - assert route.calls.last.request.method == "POST" - assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:DoubleLockDoor": {"Token": token}} - - assert response == {} - - -@respx.mock -async def test_release_door(door_control): - """Test release door""" +async def test_door_requests(door_control, input: dict, expected: str): route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( json={} ) token = "Axis-accc8ea9abac:1550808050.595717000" - response = await door_control.release_door(token) + response = await getattr(door_control, input["method_name"])(token) assert route.called assert route.calls.last.request.method == "POST" assert route.calls.last.request.url.path == "/vapix/doorcontrol" - assert json.loads(route.calls.last.request.content) == {"tdc:ReleaseDoor": {"Token": token}} + assert json.loads(route.calls.last.request.content) == {input["api_function"]: {"Token": token}} - assert response == {} + assert response == expected From f2fe2c66cd34da46a29ca6d8786ba0b06380ba1e Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 23:34:32 -0400 Subject: [PATCH 18/21] Anonymize door token --- tests/test_door_control.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/tests/test_door_control.py b/tests/test_door_control.py index 99d504f0..45721886 100644 --- a/tests/test_door_control.py +++ b/tests/test_door_control.py @@ -24,7 +24,7 @@ async def test_update(door_control): route = respx.post(f"http://{HOST}:80/vapix/doorcontrol").respond( json={"DoorInfo": [ { - "token": "Axis-accc8ea9abac:1550808050.595717000", + "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", "Name": "Main Door", "Description": "Main Door Description", "Capabilities": { @@ -63,9 +63,9 @@ async def test_update(door_control): assert len(door_control.values()) == 2 - item = door_control["Axis-accc8ea9abac:1550808050.595717000"] - assert item.id == "Axis-accc8ea9abac:1550808050.595717000" - assert item.door_token == "Axis-accc8ea9abac:1550808050.595717000" + item = door_control["Axis-5fba94a4-8601-4627-bdda-cc408f69e026"] + assert item.id == "Axis-5fba94a4-8601-4627-bdda-cc408f69e026" + assert item.door_token == "Axis-5fba94a4-8601-4627-bdda-cc408f69e026" assert item.door_name == "Main Door" assert item.door_description == "Main Door Description" @@ -132,7 +132,7 @@ async def test_get_door_info_list(door_control): json={ "DoorInfo": [ { - "token": "Axis-accc8ea9abac:1550808050.595717000", + "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", "Name": "Main Door", "Description": "", "Capabilities": { @@ -171,7 +171,7 @@ async def test_get_door_info_list(door_control): assert response["DoorInfo"] == [ { - "token": "Axis-accc8ea9abac:1550808050.595717000", + "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", "Name": "Main Door", "Description": "", "Capabilities": { @@ -207,7 +207,7 @@ async def test_get_door_info(door_control): json={ "DoorInfo": [ { - "token": "Axis-accc8ea9abac:1550808050.595717000", + "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", "Name": "Main Door", "Description": "", "Capabilities": { @@ -231,7 +231,7 @@ async def test_get_door_info(door_control): } ) - tokens = ["Axis-accc8ea9abac:1550808050.595717000"] + tokens = ["Axis-5fba94a4-8601-4627-bdda-cc408f69e026"] response = await door_control.get_door_info(tokens) @@ -244,7 +244,7 @@ async def test_get_door_info(door_control): assert response["DoorInfo"] == [ { - "token": "Axis-accc8ea9abac:1550808050.595717000", + "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", "Name": "Main Door", "Description": "", "Capabilities": { @@ -280,7 +280,7 @@ async def test_get_door_state(door_control): } ) - token = "Axis-accc8ea9abac:1550808050.595717000" + token = "Axis-5fba94a4-8601-4627-bdda-cc408f69e026" response = await door_control.get_door_state(token) @@ -318,7 +318,7 @@ async def test_door_requests(door_control, input: dict, expected: str): json={} ) - token = "Axis-accc8ea9abac:1550808050.595717000" + token = "Axis-5fba94a4-8601-4627-bdda-cc408f69e026" response = await getattr(door_control, input["method_name"])(token) From 07a9dd7a6ad43e723c7532f526fa596a7e2f9640 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 23:36:56 -0400 Subject: [PATCH 19/21] Anonymize more --- tests/test_door_control.py | 40 +++++++++++++++++++------------------- 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/tests/test_door_control.py b/tests/test_door_control.py index 45721886..41746716 100644 --- a/tests/test_door_control.py +++ b/tests/test_door_control.py @@ -25,8 +25,8 @@ async def test_update(door_control): json={"DoorInfo": [ { "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", - "Name": "Main Door", - "Description": "Main Door Description", + "Name": "Test Door 1", + "Description": "Test Door 1 Description", "Capabilities": { "Access": True, "Lock": True, @@ -45,9 +45,9 @@ async def test_update(door_control): } }, { - "token": "Axis-accc8ee70fbe:1614733258.565014000", - "Name": "North Gym Door", - "Description": "North Gym Door Description", + "token": "Axis-c2987ee0-28d5-4b53-8493-52977af927cf", + "Name": "Test Door 2", + "Description": "Test Door 2 Description", "Capabilities": {} } ] @@ -66,8 +66,8 @@ async def test_update(door_control): item = door_control["Axis-5fba94a4-8601-4627-bdda-cc408f69e026"] assert item.id == "Axis-5fba94a4-8601-4627-bdda-cc408f69e026" assert item.door_token == "Axis-5fba94a4-8601-4627-bdda-cc408f69e026" - assert item.door_name == "Main Door" - assert item.door_description == "Main Door Description" + assert item.door_name == "Test Door 1" + assert item.door_description == "Test Door 1 Description" # TODO: Check to see that comparing dict using "==" actually does what I want it to do assert item.door_capabilities == { @@ -87,11 +87,11 @@ async def test_update(door_control): "Configurable": True } - item2 = door_control["Axis-accc8ee70fbe:1614733258.565014000"] - assert item2.id == "Axis-accc8ee70fbe:1614733258.565014000" - assert item2.door_token == "Axis-accc8ee70fbe:1614733258.565014000" - assert item2.door_name == "North Gym Door" - assert item2.door_description == "North Gym Door Description" + item2 = door_control["Axis-c2987ee0-28d5-4b53-8493-52977af927cf"] + assert item2.id == "Axis-c2987ee0-28d5-4b53-8493-52977af927cf" + assert item2.door_token == "Axis-c2987ee0-28d5-4b53-8493-52977af927cf" + assert item2.door_name == "Test Door 2" + assert item2.door_description == "Test Door 2 Description" # TODO: Check to see that comparing dict using "==" actually does what I want it to do assert item2.door_capabilities == {} @@ -133,7 +133,7 @@ async def test_get_door_info_list(door_control): "DoorInfo": [ { "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", - "Name": "Main Door", + "Name": "Test Door 1", "Description": "", "Capabilities": { "Access": True, @@ -153,8 +153,8 @@ async def test_get_door_info_list(door_control): } }, { - "token": "Axis-accc8ee70fbe:1614733258.565014000", - "Name": "North Gym Door", + "token": "Axis-c2987ee0-28d5-4b53-8493-52977af927cf", + "Name": "Test Door 2", "Description": "", "Capabilities": {} } @@ -172,7 +172,7 @@ async def test_get_door_info_list(door_control): assert response["DoorInfo"] == [ { "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", - "Name": "Main Door", + "Name": "Test Door 1", "Description": "", "Capabilities": { "Access": True, @@ -192,8 +192,8 @@ async def test_get_door_info_list(door_control): } }, { - "token": "Axis-accc8ee70fbe:1614733258.565014000", - "Name": "North Gym Door", + "token": "Axis-c2987ee0-28d5-4b53-8493-52977af927cf", + "Name": "Test Door 2", "Description": "", "Capabilities": {} } @@ -208,7 +208,7 @@ async def test_get_door_info(door_control): "DoorInfo": [ { "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", - "Name": "Main Door", + "Name": "Test Door 1", "Description": "", "Capabilities": { "Access": True, @@ -245,7 +245,7 @@ async def test_get_door_info(door_control): assert response["DoorInfo"] == [ { "token": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", - "Name": "Main Door", + "Name": "Test Door 1", "Description": "", "Capabilities": { "Access": True, From 8015c6a02cf98282ffb13992e2d99510cd747103 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Wed, 9 Jun 2021 23:53:16 -0400 Subject: [PATCH 20/21] Some event testing added --- tests/event_fixtures.py | 2 + tests/test_event_stream.py | 413 +++++++++++++++++++------------------ 2 files changed, 215 insertions(+), 200 deletions(-) diff --git a/tests/event_fixtures.py b/tests/event_fixtures.py index 55a9cf65..6658f05b 100644 --- a/tests/event_fixtures.py +++ b/tests/event_fixtures.py @@ -4,6 +4,8 @@ DAYNIGHT_INIT = b'\n\ntns1:VideoSource/tnsaxis:DayNightVisionuri://1c8ae81b-3b00-46cf-bf76-79cc3fa533dc/ProducerReference\n' +DOOR_MODE_INIT = b'\n\n tns1:Door/State/DoorModeuri://755cc9bb-cf3a-410b-bd1b-0ec97c6d6256/ProducerReference\n' + FENCE_GUARD_INIT = b'\n\ntnsaxis:CameraApplicationPlatform/FenceGuard/Camera1Profile1uri://755cc9bb-cf3a-410b-bd1b-0ec97c6d6256/ProducerReference\n' GLOBAL_SCENE_CHANGE = b'\n\ntns1:VideoSource/GlobalSceneChange/ImagingServiceuri://755cc9bb-cf3a-410b-bd1b-0ec97c6d6256/ProducerReference\n' diff --git a/tests/test_event_stream.py b/tests/test_event_stream.py index 9517278e..24b3cef3 100644 --- a/tests/test_event_stream.py +++ b/tests/test_event_stream.py @@ -12,6 +12,7 @@ FIRST_MESSAGE, AUDIO_INIT, DAYNIGHT_INIT, + DOOR_MODE_INIT, FENCE_GUARD_INIT, GLOBAL_SCENE_CHANGE, LIGHT_STATUS_INIT, @@ -55,66 +56,66 @@ def event_manager(axis_device) -> EventManager: [ (FIRST_MESSAGE, {}), ( - PIR_INIT, - { - "operation": "Initialized", - "topic": "tns1:Device/tnsaxis:Sensor/PIR", - "source": "sensor", - "source_idx": "0", - "type": "state", - "value": "0", - }, + PIR_INIT, + { + "operation": "Initialized", + "topic": "tns1:Device/tnsaxis:Sensor/PIR", + "source": "sensor", + "source_idx": "0", + "type": "state", + "value": "0", + }, ), ( - PIR_CHANGE, - { - "operation": "Changed", - "topic": "tns1:Device/tnsaxis:Sensor/PIR", - "source": "sensor", - "source_idx": "0", - "type": "state", - "value": "1", - }, + PIR_CHANGE, + { + "operation": "Changed", + "topic": "tns1:Device/tnsaxis:Sensor/PIR", + "source": "sensor", + "source_idx": "0", + "type": "state", + "value": "1", + }, ), ( - RULE_ENGINE_REGION_DETECTOR_INIT, - { - "operation": "Initialized", - "source": "VideoSource", - "source_idx": "0", - "topic": "tns1:RuleEngine/MotionRegionDetector/Motion", - "type": "State", - "value": "0", - }, + RULE_ENGINE_REGION_DETECTOR_INIT, + { + "operation": "Initialized", + "source": "VideoSource", + "source_idx": "0", + "topic": "tns1:RuleEngine/MotionRegionDetector/Motion", + "type": "State", + "value": "0", + }, ), ( - STORAGE_ALERT_INIT, - { - "operation": "Initialized", - "source": "disk_id", - "source_idx": "NetworkShare", - "topic": "tnsaxis:Storage/Alert", - "type": "overall_health", - "value": "-3", - }, + STORAGE_ALERT_INIT, + { + "operation": "Initialized", + "source": "disk_id", + "source_idx": "NetworkShare", + "topic": "tnsaxis:Storage/Alert", + "type": "overall_health", + "value": "-3", + }, ), ( - VMD4_ANY_INIT, - { - "operation": "Initialized", - "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", - "type": "active", - "value": "0", - }, + VMD4_ANY_INIT, + { + "operation": "Initialized", + "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", + "type": "active", + "value": "0", + }, ), ( - VMD4_ANY_CHANGE, - { - "operation": "Changed", - "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", - "type": "active", - "value": "1", - }, + VMD4_ANY_CHANGE, + { + "operation": "Changed", + "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", + "type": "active", + "value": "1", + }, ), ], ) @@ -127,184 +128,196 @@ def test_parse_event_xml(event_manager, input: bytes, expected: dict): "input,expected", [ ( - AUDIO_INIT, - { - "topic": "tns1:AudioSource/tnsaxis:TriggerLevel", - "source": "channel", - "source_idx": "1", - "class": "sound", - "type": "Sound", - "state": "0", - "tripped": False, - }, + AUDIO_INIT, + { + "topic": "tns1:AudioSource/tnsaxis:TriggerLevel", + "source": "channel", + "source_idx": "1", + "class": "sound", + "type": "Sound", + "state": "0", + "tripped": False, + }, ), ( - DAYNIGHT_INIT, - { - "topic": "tns1:VideoSource/tnsaxis:DayNightVision", - "source": "VideoSourceConfigurationToken", - "source_idx": "1", - "class": "light", - "type": "DayNight", - "state": "1", - "tripped": True, - }, + DAYNIGHT_INIT, + { + "topic": "tns1:VideoSource/tnsaxis:DayNightVision", + "source": "VideoSourceConfigurationToken", + "source_idx": "1", + "class": "light", + "type": "DayNight", + "state": "1", + "tripped": True, + }, ), ( - FENCE_GUARD_INIT, - { - "topic": "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1Profile1", - "source": "", - "source_idx": "Camera1Profile1", - "class": "motion", - "type": "Fence Guard", - "state": "0", - "tripped": False, - }, + DOOR_MODE_INIT, + { + "topic": "tns1:Door/State/DoorMode", + "source": "DoorToken", + "source_idx": "Axis-5fba94a4-8601-4627-bdda-cc408f69e026", + "class": "door", + "type": "Door Mode", + "state": "Accessed", + "tripped": False, + }, ), ( - LIGHT_STATUS_INIT, - { - "topic": "tns1:Device/tnsaxis:Light/Status", - "source": "id", - "source_idx": "0", - "class": "light", - "type": "Light", - "state": "OFF", - "tripped": False, - }, + FENCE_GUARD_INIT, + { + "topic": "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1Profile1", + "source": "", + "source_idx": "Camera1Profile1", + "class": "motion", + "type": "Fence Guard", + "state": "0", + "tripped": False, + }, ), ( - LOITERING_GUARD_INIT, - { - "topic": "tnsaxis:CameraApplicationPlatform/LoiteringGuard/Camera1Profile1", - "source": "", - "source_idx": "Camera1Profile1", - "class": "motion", - "type": "Loitering Guard", - "state": "0", - "tripped": False, - }, + LIGHT_STATUS_INIT, + { + "topic": "tns1:Device/tnsaxis:Light/Status", + "source": "id", + "source_idx": "0", + "class": "light", + "type": "Light", + "state": "OFF", + "tripped": False, + }, ), ( - MOTION_GUARD_INIT, - { - "topic": "tnsaxis:CameraApplicationPlatform/MotionGuard/Camera1ProfileANY", - "source": "", - "source_idx": "Camera1ProfileANY", - "class": "motion", - "type": "Motion Guard", - "state": "0", - "tripped": False, - }, + LOITERING_GUARD_INIT, + { + "topic": "tnsaxis:CameraApplicationPlatform/LoiteringGuard/Camera1Profile1", + "source": "", + "source_idx": "Camera1Profile1", + "class": "motion", + "type": "Loitering Guard", + "state": "0", + "tripped": False, + }, ), ( - OBJECT_ANALYTICS_INIT, - { - "topic": "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1", - "source": "", - "source_idx": "Device1Scenario1", - "class": "motion", - "type": "Object Analytics", - "state": "0", - "tripped": False, - }, + MOTION_GUARD_INIT, + { + "topic": "tnsaxis:CameraApplicationPlatform/MotionGuard/Camera1ProfileANY", + "source": "", + "source_idx": "Camera1ProfileANY", + "class": "motion", + "type": "Motion Guard", + "state": "0", + "tripped": False, + }, ), ( - PIR_INIT, - { - "topic": "tns1:Device/tnsaxis:Sensor/PIR", - "source": "sensor", - "source_idx": "0", - "class": "motion", - "type": "PIR", - "state": "0", - "tripped": False, - }, + OBJECT_ANALYTICS_INIT, + { + "topic": "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1", + "source": "", + "source_idx": "Device1Scenario1", + "class": "motion", + "type": "Object Analytics", + "state": "0", + "tripped": False, + }, ), ( - PORT_0_INIT, - { - "topic": "tns1:Device/tnsaxis:IO/Port", - "source": "port", - "source_idx": "1", - "class": "input", - "type": "Input", - "state": "0", - "tripped": False, - }, + PIR_INIT, + { + "topic": "tns1:Device/tnsaxis:Sensor/PIR", + "source": "sensor", + "source_idx": "0", + "class": "motion", + "type": "PIR", + "state": "0", + "tripped": False, + }, ), ( - PORT_ANY_INIT, - { - "topic": "tns1:Device/tnsaxis:IO/Port", - "source": "port", - "source_idx": "", - "class": "input", - "type": "Input", - "state": "0", - "tripped": False, - }, + PORT_0_INIT, + { + "topic": "tns1:Device/tnsaxis:IO/Port", + "source": "port", + "source_idx": "1", + "class": "input", + "type": "Input", + "state": "0", + "tripped": False, + }, ), ( - PTZ_MOVE_INIT, - { - "topic": "tns1:PTZController/tnsaxis:Move/Channel_1", - "source": "PTZConfigurationToken", - "source_idx": "1", - "class": "ptz", - "type": "is_moving", - "state": "0", - "tripped": False, - }, + PORT_ANY_INIT, + { + "topic": "tns1:Device/tnsaxis:IO/Port", + "source": "port", + "source_idx": "", + "class": "input", + "type": "Input", + "state": "0", + "tripped": False, + }, ), ( - PTZ_PRESET_INIT_1, - { - "topic": "tns1:PTZController/tnsaxis:PTZPresets/Channel_1", - "source": "PresetToken", - "source_idx": "1", - "class": "ptz", - "type": "on_preset", - "state": "1", - "tripped": True, - }, + PTZ_MOVE_INIT, + { + "topic": "tns1:PTZController/tnsaxis:Move/Channel_1", + "source": "PTZConfigurationToken", + "source_idx": "1", + "class": "ptz", + "type": "is_moving", + "state": "0", + "tripped": False, + }, ), ( - RELAY_INIT, - { - "topic": "tns1:Device/Trigger/Relay", - "source": "RelayToken", - "source_idx": "3", - "class": "output", - "type": "Relay", - "state": "inactive", - "tripped": False, - }, + PTZ_PRESET_INIT_1, + { + "topic": "tns1:PTZController/tnsaxis:PTZPresets/Channel_1", + "source": "PresetToken", + "source_idx": "1", + "class": "ptz", + "type": "on_preset", + "state": "1", + "tripped": True, + }, ), ( - VMD3_INIT, - { - "topic": "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", - "source": "areaid", - "source_idx": "0", - "class": "motion", - "type": "VMD3", - "state": "0", - "tripped": False, - }, + RELAY_INIT, + { + "topic": "tns1:Device/Trigger/Relay", + "source": "RelayToken", + "source_idx": "3", + "class": "output", + "type": "Relay", + "state": "inactive", + "tripped": False, + }, ), ( - VMD4_ANY_INIT, - { - "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", - "source": "", - "source_idx": "Camera1ProfileANY", - "class": "motion", - "type": "VMD4", - "state": "0", - "tripped": False, - }, + VMD3_INIT, + { + "topic": "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", + "source": "areaid", + "source_idx": "0", + "class": "motion", + "type": "VMD3", + "state": "0", + "tripped": False, + }, + ), + ( + VMD4_ANY_INIT, + { + "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", + "source": "", + "source_idx": "Camera1ProfileANY", + "class": "motion", + "type": "VMD4", + "state": "0", + "tripped": False, + }, ), ], ) From ed9eafc9db4d6f6f94d45e44e309dc149e2fe911 Mon Sep 17 00:00:00 2001 From: drewclauson Date: Sat, 31 Jul 2021 14:30:51 -0400 Subject: [PATCH 21/21] Fix indentation --- axis/vapix.py | 54 +++++++++++++++++++++++++-------------------------- 1 file changed, 27 insertions(+), 27 deletions(-) diff --git a/axis/vapix.py b/axis/vapix.py index 3cec45c9..202cb9ec 100644 --- a/axis/vapix.py +++ b/axis/vapix.py @@ -117,7 +117,7 @@ async def initialize(self) -> None: await self.initialize_applications() async def _initialize_api_attribute( - self, api_class: Callable, api_attr: str + self, api_class: Callable, api_attr: str ) -> None: """Initialize API and load data.""" api_instance = api_class(self.request) @@ -139,13 +139,13 @@ async def initialize_api_discovery(self) -> None: tasks = [] for api_id, api_class, api_attr in ( - (BASIC_DEVICE_INFO_ID, BasicDeviceInfo, "basic_device_info"), - (IO_PORT_MANAGEMENT_ID, IoPortManagement, "ports"), - (DOOR_CONTROL_ID, DoorControl, "door_control"), - (LIGHT_CONTROL_ID, LightControl, "light_control"), - (MQTT_ID, MqttClient, "mqtt"), - (STREAM_PROFILES_ID, StreamProfiles, "stream_profiles"), - (VIEW_AREAS_ID, ViewAreas, "view_areas"), + (BASIC_DEVICE_INFO_ID, BasicDeviceInfo, "basic_device_info"), + (IO_PORT_MANAGEMENT_ID, IoPortManagement, "ports"), + (DOOR_CONTROL_ID, DoorControl, "door_control"), + (LIGHT_CONTROL_ID, LightControl, "light_control"), + (MQTT_ID, MqttClient, "mqtt"), + (STREAM_PROFILES_ID, StreamProfiles, "stream_profiles"), + (VIEW_AREAS_ID, ViewAreas, "view_areas"), ): if api_id in self.api_discovery: tasks.append(self._initialize_api_attribute(api_class, api_attr)) @@ -194,7 +194,7 @@ async def initialize_applications(self) -> None: """Load data for applications on device.""" self.applications = Applications(self.request) if self.params and version.parse( - self.params.embedded_development + self.params.embedded_development ) >= version.parse(APPLICATIONS_MINIMUM_VERSION): try: await self.applications.update() @@ -204,16 +204,16 @@ async def initialize_applications(self) -> None: tasks = [] for app_class, app_attr in ( - (FenceGuard, "fence_guard"), - (LoiteringGuard, "loitering_guard"), - (MotionGuard, "motion_guard"), - (ObjectAnalytics, "object_analytics"), - (Vmd4, "vmd4"), + (FenceGuard, "fence_guard"), + (LoiteringGuard, "loitering_guard"), + (MotionGuard, "motion_guard"), + (ObjectAnalytics, "object_analytics"), + (Vmd4, "vmd4"), ): if ( - app_class.APPLICATION_NAME in self.applications # type: ignore[attr-defined] - and self.applications[app_class.APPLICATION_NAME].status # type: ignore[attr-defined] - == APPLICATION_STATE_RUNNING + app_class.APPLICATION_NAME in self.applications # type: ignore[attr-defined] + and self.applications[app_class.APPLICATION_NAME].status # type: ignore[attr-defined] + == APPLICATION_STATE_RUNNING ): tasks.append(self._initialize_api_attribute(app_class, app_attr)) @@ -240,11 +240,11 @@ async def load_user_groups(self) -> None: if self.users and self.config.username in self.users: user = self.users[self.config.username] user_groups = ( - f"{user.name}\n" # type: ignore[attr-defined] - + ("admin " if user.admin else "") # type: ignore[attr-defined] - + ("operator " if user.operator else "") # type: ignore[attr-defined] - + ("viewer " if user.viewer else "") # type: ignore[attr-defined] - + ("ptz" if user.ptz else "") # type: ignore[attr-defined] + f"{user.name}\n" # type: ignore[attr-defined] + + ("admin " if user.admin else "") # type: ignore[attr-defined] + + ("operator " if user.operator else "") # type: ignore[attr-defined] + + ("viewer " if user.viewer else "") # type: ignore[attr-defined] + + ("ptz" if user.ptz else "") # type: ignore[attr-defined] ) else: @@ -256,11 +256,11 @@ async def load_user_groups(self) -> None: self.user_groups = UserGroups(user_groups, self.request) async def request( - self, - method: str, - path: str, - kwargs_xmltodict: Optional[dict] = None, - **kwargs: dict, + self, + method: str, + path: str, + kwargs_xmltodict: Optional[dict] = None, + **kwargs: dict, ) -> Union[dict, str]: """Make a request to the API.""" url = self.config.url + path