import datetime import responses from responses import matchers from unifi_access.schemas import ( AccessPolicyId, NfcCardToken, UserGroupId, UserId, UserStatus, ) from .base import UnifiAccessTests class UserTests(UnifiAccessTests): @responses.activate def test_register_user(self) -> None: """3.2 User Registration""" responses.post( f"https://{self.host}/api/v1/developer/users", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "first_name": "First Name", "last_name": "Last Name", "employee_number": "100000", "onboard_time": 1689150139, "user_email": "example@*.com", } ), ], json={ "code": "SUCCESS", "msg": "success", "data": { "id": "37f2b996-c2c5-487b-aa22-8b453ff14a4b", "first_name": "First Name", "last_name": "Last Name", "employee_number": "100000", "onboard_time": 1689150139, "user_email": "example@*.com", # NOTE: following fields missing from example, but present in real responses "alias": "", "avatar_relative_path": "", "email": "", "email_status": "UNVERIFIED", "full_name": "First Name Last Name", "phone": "", "status": "ACTIVE", "username": "", }, }, ) resp = self.client.register_user( first_name="First Name", last_name="Last Name", employee_number="100000", onboard_time=datetime.datetime.fromtimestamp(1689150139, tz=datetime.UTC), user_email="example@*.com", ) assert resp.id == "37f2b996-c2c5-487b-aa22-8b453ff14a4b" @responses.activate def test_update_user(self) -> None: """3.3 Update User""" user_id = UserId("37f2b996-c2c5-487b-aa22-8b453ff14a4b") responses.put( f"https://{self.host}/api/v1/developer/users/{user_id}", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "first_name": "H", "last_name": "L", "employee_number": "", "user_email": "example@*.com", # XXX: TODO: this is not listed as a valid param in their docs... # "pin_code": "", "onboard_time": 1689150139, "status": "ACTIVE", } ), ], json={"code": "SUCCESS", "msg": "success", "data": None}, ) self.client.update_user( user_id=user_id, first_name="H", last_name="L", employee_number="", onboard_time=datetime.datetime.fromtimestamp(1689150139, tz=datetime.UTC), user_email="example@*.com", status=UserStatus.ACTIVE, ) @responses.activate def test_fetch_user(self) -> None: """3.4 Fetch User""" user_id = UserId("37f2b996-c2c5-487b-aa22-8b453ff14a4b") responses.get( f"https://{self.host}/api/v1/developer/users/{user_id}", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher({"expand[]": "access_policy"}), ], json={ "code": "SUCCESS", "data": { "access_policies": [ { "id": "edbc80df-3698-49fd-8b53-f1867f104947", "name": "test", "resources": [ { "id": "d5573467-d6b3-4e8f-8e48-8a322b91664a", "type": "door_group", }, { "id": "5c496423-6d25-4e4f-8cdf-95ad5135188a", "type": "door_group", }, { "id": "6ff875d2-af87-470b-9cb5-774c6596afc8", "type": "door", }, ], "schedule_id": "73facd6c-839e-4521-a4f4-c07e1d44e748", } ], "access_policy_ids": ["edbc80df-3698-49fd-8b53-f1867f104947"], "employee_number": "", "first_name": "***", "id": "37f2b996-c2c5-487b-aa22-8b453ff14a4b", "last_name": "L", "user_email": "example@*.com", "nfc_cards": [ { "id": "100001", "token": "d27822fc682b478dc637c6db01813e465174df6e54ca515d8427db623cfda1d0", "type": "ua_card", } ], "onboard_time": 1689047588, "pin_code": { "token": "5f742ee4424e5a7dd265de3461009b9ebafa1fb9d6b15018842055cc0466ac56" }, "status": "ACTIVE", # NOTE: following fields missing from example, but present in real responses "alias": "", "avatar_relative_path": "", "email": "", "email_status": "UNVERIFIED", "full_name": "*** L", "phone": "", "username": "", }, "msg": "success", }, ) resp = self.client.fetch_user(user_id=user_id, expand_access_policies=True) # TODO: verify correctness of data? @responses.activate def test_fetch_all_users(self) -> None: """3.5 Fetch All Users""" responses.get( f"https://{self.host}/api/v1/developer/users", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher( {"expand[]": "access_policy", "page_num": 1, "page_size": 25} ), ], json={ "code": "SUCCESS", "data": [ { "access_policies": [ { "id": "73f15cab-c725-4a76-a419-a4026d131e96", "name": "Default Admin Policy", "resources": [ { "id": "d5573467-d6b3-4e8f-8e48-8a322b91664a", "type": "door_group", }, { "id": "5c496423-6d25-4e4f-8cdf-95ad5135188a", "type": "door_group", }, ], "schedule_id": "73facd6c-839e-4521-a4f4-c07e1d44e748", } ], "access_policy_ids": ["73f15cab-c725-4a76-a419-a4026d131e96"], "employee_number": "", "first_name": "UniFi", "id": "83569f9b-0096-48ab-b2e4-5c9a598568a8", "last_name": "User", "user_email": "", "nfc_cards": [], "onboard_time": 0, "pin_code": None, "status": "ACTIVE", # NOTE: following fields missing from example, but present in real responses "alias": "", "avatar_relative_path": "", "email": "", "email_status": "UNVERIFIED", "full_name": "UniFi User", "phone": "", "username": "", }, { "access_policies": [ { "id": "c1682fb8-ef6e-4fe2-aa8a-b6f29df753ff", "name": "policy_1690272668035", "resources": [ { "id": "6ff875d2-af87-470b-9cb5-774c6596afc8", "type": "door", } ], "schedule_id": "0616ef06-b807-4372-9ae0-7a87e12e4019", } ], "access_policy_ids": ["c1682fb8-ef6e-4fe2-aa8a-b6f29df753ff"], "employee_number": "", "first_name": "Ttttt", "id": "3a3ba57a-796e-46e0-b8f3-478bb70a114d", "last_name": "Tttt", "nfc_cards": [], "onboard_time": 1689048000, "pin_code": None, "status": "ACTIVE", # NOTE: following fields missing from example, but present in real responses "alias": "", "avatar_relative_path": "", "user_email": "", "email": "", "email_status": "UNVERIFIED", "full_name": "Ttttt Tttt", "phone": "", "username": "", }, ], "msg": "success", "pagination": {"page_num": 1, "page_size": 97, "total": 97}, }, ) resp = self.client.fetch_all_users( expand_access_policies=True, page_num=1, page_size=25 ) assert resp.pagination # TODO: verify correctness of data? # NOTE: not taken from API docs examples @responses.activate def test_fetch_all_users__unpaged(self) -> None: """3.5 Fetch All Users, with pagination""" responses.get( f"https://{self.host}/api/v1/developer/users", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher({"page_num": 1, "page_size": 1}), ], json={ "code": "SUCCESS", "data": [ { "access_policy_ids": ["73f15cab-c725-4a76-a419-a4026d131e96"], "employee_number": "", "first_name": "UniFi", "id": "83569f9b-0096-48ab-b2e4-5c9a598568a8", "last_name": "User", "user_email": "", "nfc_cards": [], "onboard_time": 0, "pin_code": None, "status": "ACTIVE", "alias": "", "avatar_relative_path": "", "email": "", "email_status": "UNVERIFIED", "full_name": "UniFi User", "phone": "", "username": "", }, ], "msg": "success", "pagination": {"page_num": 1, "page_size": 1, "total": 2}, }, ) responses.get( f"https://{self.host}/api/v1/developer/users", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher({"page_num": 2, "page_size": 1}), ], json={ "code": "SUCCESS", "data": [ { "access_policy_ids": ["c1682fb8-ef6e-4fe2-aa8a-b6f29df753ff"], "employee_number": "", "first_name": "Ttttt", "id": "3a3ba57a-796e-46e0-b8f3-478bb70a114d", "last_name": "Tttt", "nfc_cards": [], "onboard_time": 1689048000, "pin_code": None, "status": "ACTIVE", "alias": "", "avatar_relative_path": "", "user_email": "", "email": "", "email_status": "UNVERIFIED", "full_name": "Ttttt Tttt", "phone": "", "username": "", }, ], "msg": "success", "pagination": {"page_num": 2, "page_size": 1, "total": 2}, }, ) resp = list( self.client.fetch_all_users__unpaged( expand_access_policies=False, page_size=1 ) ) assert resp[0].id == "83569f9b-0096-48ab-b2e4-5c9a598568a8" assert resp[1].id == "3a3ba57a-796e-46e0-b8f3-478bb70a114d" @responses.activate def test_assign_access_policy_to_user(self) -> None: """3.6 Assign Access Policy to User""" user_id = UserId("37f2b996-c2c5-487b-aa22-8b453ff14a4b") responses.put( f"https://{self.host}/api/v1/developer/users/{user_id}/access_policies", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "access_policy_ids": [ "03895c7f-9f53-4334-812b-5db9c122c109", "3b6bcb0c-7498-44cf-8615-00a96d824cbe", ] } ), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.assign_access_policy_to_user( user_id=user_id, access_policy_ids=[ AccessPolicyId("03895c7f-9f53-4334-812b-5db9c122c109"), AccessPolicyId("3b6bcb0c-7498-44cf-8615-00a96d824cbe"), ], ) @responses.activate def test_assign_nfc_card_to_user(self) -> None: """3.7 Assign NFC Card to User""" user_id = UserId("17d2f099-99df-429b-becb-1399a6937e5a") responses.put( f"https://{self.host}/api/v1/developer/users/{user_id}/nfc_cards", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "token": "d27822fc682b478dc637c6db01813e465174df6e54ca515d8427db623cfda1d0", "force_add": True, } ), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.assign_nfc_card_to_user( user_id=user_id, token=NfcCardToken( "d27822fc682b478dc637c6db01813e465174df6e54ca515d8427db623cfda1d0" ), force_add=True, ) @responses.activate def test_unassign_nfc_card_from_user(self) -> None: """3.8 Unassign NFC Card from User""" user_id = UserId("17d2f099-99df-429b-becb-1399a6937e5a") responses.put( f"https://{self.host}/api/v1/developer/users/{user_id}/nfc_cards/delete", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "token": "d27822fc682b478dc637c6db01813e465174df6e54ca515d8427db623cfda1d0", } ), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.unassign_nfc_card_from_user( user_id=user_id, token=NfcCardToken( "d27822fc682b478dc637c6db01813e465174df6e54ca515d8427db623cfda1d0" ), ) @responses.activate def test_assign_pin_code_to_user(self) -> None: """3.9 Assign PIN Code to User""" user_id = UserId("17d2f099-99df-429b-becb-1399a6937e5a") responses.put( f"https://{self.host}/api/v1/developer/users/{user_id}/pin_codes", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher({"pin_code": "57301208"}), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.assign_pin_code_to_user(user_id=user_id, pin_code="57301208") @responses.activate def test_unassign_pin_code_from_user(self) -> None: """3.10 Unassign PIN Code from User""" user_id = UserId("17d2f099-99df-429b-becb-1399a6937e5a") responses.delete( f"https://{self.host}/api/v1/developer/users/{user_id}/pin_codes", match=[ matchers.header_matcher(self.common_headers), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.unassign_pin_code_from_user( user_id=user_id, ) @responses.activate def test_create_user_group(self) -> None: """3.11 Create User Group""" responses.post( f"https://{self.host}/api/v1/developer/user_groups", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "name": "Group Name", "up_id": "013d05d3-7262-4908-ba69-badbbbf8f5a6", } ), ], json={ "code": "SUCCESS", # NOTE: completely missing from docs "data": { "full_name": "UniFi-CloudKey-Gen2-Plus / Group Name", "id": "7a681dd8-4d38-4bbf-a061-2fd5d71db5c1", "name": "Group Name", "up_id": "bdec1374-9547-4541-9761-d9e66cb1c367", "up_ids": ["bdec1374-9547-4541-9761-d9e66cb1c367"], }, "msg": "success", }, ) resp = self.client.create_user_group( name="Group Name", up_id=UserGroupId("013d05d3-7262-4908-ba69-badbbbf8f5a6") ) @responses.activate def test_fetch_all_user_groups(self) -> None: """3.12 Fetch All User Groups""" responses.get( f"https://{self.host}/api/v1/developer/user_groups", match=[matchers.header_matcher(self.common_headers)], json={ "code": "SUCCESS", "data": [ { "full_name": "Group Name", "id": "75011ee6-b7ab-4927-9d9f-dd08ef0a3199", "name": "Group Name", "up_id": "a27899fc-a2d1-4797-8d4d-86118f8555f3", "up_ids": ["a27899fc-a2d1-4797-8d4d-86118f8555f3"], } ], "msg": "success", }, ) resp = self.client.fetch_all_user_groups() @responses.activate def test_fetch_user_group(self) -> None: """3.13 Fetch User Group""" user_group_id = UserGroupId("75011ee6-b7ab-4927-9d9f-dd08ef0a3199") responses.get( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}", match=[matchers.header_matcher(self.common_headers)], json={ "code": "SUCCESS", "data": { "full_name": "Group Name", "id": "75011ee6-b7ab-4927-9d9f-dd08ef0a3199", "name": "Group Name", "up_id": "a27899fc-a2d1-4797-8d4d-86118f8555f3", "up_ids": ["a27899fc-a2d1-4797-8d4d-86118f8555f3"], }, "msg": "success", }, ) resp = self.client.fetch_user_group(user_group_id=user_group_id) @responses.activate def test_update_user_group(self) -> None: """3.14 Update User Group""" user_group_id = UserGroupId("75011ee6-b7ab-4927-9d9f-dd08ef0a3199") responses.put( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( { "name": "Group Name", "up_id": "013d05d3-7262-4908-ba69-badbbbf8f5a6", } ), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.update_user_group( user_group_id=user_group_id, name="Group Name", up_id=UserGroupId("013d05d3-7262-4908-ba69-badbbbf8f5a6"), ) @responses.activate def test_delete_user_group(self) -> None: """3.15 Delete User Group""" user_group_id = UserGroupId("75011ee6-b7ab-4927-9d9f-dd08ef0a3199") responses.delete( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}", match=[matchers.header_matcher(self.common_headers)], json={"code": "SUCCESS", "msg": "success"}, ) self.client.delete_user_group(user_group_id=user_group_id) @responses.activate def test_assign_user_to_user_group(self) -> None: """3.16 Assign User to User Group""" user_group_id = UserGroupId("75011ee6-b7ab-4927-9d9f-dd08ef0a3199") responses.post( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}/users", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( [ "7c6e9102-acb7-4b89-8ed4-7561e6fb706c", "fd63bc4c-52e0-4dbf-a699-e1233339c73b", ] ), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.assign_user_to_user_group( user_group_id=user_group_id, users=[ UserId("7c6e9102-acb7-4b89-8ed4-7561e6fb706c"), UserId("fd63bc4c-52e0-4dbf-a699-e1233339c73b"), ], ) @responses.activate def test_unassign_user_from_user_group(self) -> None: """3.17 Unassign User from User Group""" user_group_id = UserGroupId("75011ee6-b7ab-4927-9d9f-dd08ef0a3199") responses.post( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}/users/delete", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( [ "7c6e9102-acb7-4b89-8ed4-7561e6fb706c", "fd63bc4c-52e0-4dbf-a699-e1233339c73b", ] ), ], json={"code": "SUCCESS", "msg": "success"}, ) self.client.unassign_user_from_user_group( user_group_id=user_group_id, users=[ UserId("7c6e9102-acb7-4b89-8ed4-7561e6fb706c"), UserId("fd63bc4c-52e0-4dbf-a699-e1233339c73b"), ], ) @responses.activate def test_fetch_users_in_a_user_group(self) -> None: """3.18 Fetch Users in a User Group""" user_group_id = UserGroupId("23676a54-382e-4121-aa80-878d2d9bacaa") responses.get( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}/users", match=[matchers.header_matcher(self.common_headers)], json={ "code": "SUCCESS", "data": [ { "alias": "", "avatar_relative_path": "", "email": "*@*.com", "email_status": "UNVERIFIED", "employee_number": "1000000", "first_name": "", "full_name": "", "id": "27aa91ac-2924-43d4-82e1-24b6a570d29e", "last_name": "Chen", "onboard_time": 1689150139, "phone": "", "status": "ACTIVE", "user_email": "", "username": "", } ], "msg": "success", }, ) resp = self.client.fetch_users_in_a_user_group( user_group_id=user_group_id, ) @responses.activate def test_fetch_all_users_in_a_user_group(self) -> None: """3.19 Fetch All Users in a User Group""" user_group_id = UserGroupId("23676a54-382e-4121-aa80-878d2d9bacaa") responses.get( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}/users/all", match=[matchers.header_matcher(self.common_headers)], json={ "code": "SUCCESS", "data": [ { "alias": "", "avatar_relative_path": "", "email": "*@*.com", "email_status": "UNVERIFIED", "employee_number": "1000000", "first_name": "", "full_name": "", "id": "27aa91ac-2924-43d4-82e1-24b6a570d29e", "last_name": "Chen", "onboard_time": 1689150139, "phone": "", "status": "ACTIVE", "user_email": "", "username": "", } ], "msg": "success", }, ) resp = self.client.fetch_all_users_in_a_user_group( user_group_id=user_group_id, ) @responses.activate def test_fetch_the_access_policies_assigned_to_a_user(self) -> None: """3.20 Fetch the Access Policies Assigned to a User""" user_id = UserId("27aa91ac-2924-43d4-82e1-24b6a570d29e") responses.get( f"https://{self.host}/api/v1/developer/users/{user_id}/access_policies", match=[ matchers.header_matcher(self.common_headers), matchers.query_param_matcher({"only_user_policies": "false"}), ], json={ "code": "SUCCESS", "data": [ { "id": "89a4ca95-1502-4ae7-954f-d986b67afe5c", "name": "Default Site Policy", "resources": [ { "id": "fd2a06e2-81af-4cf4-9bd5-8bceb5e7b7d7", "type": "door_group", } ], "schedule_id": "6b79d12a-2a6e-4463-949c-f1a98fff40d2", }, { "id": "bbe48a65-2ac1-4bf6-bd65-bc8f9ee7fb75", "name": "Access Policy Name", "resources": [], "schedule_id": "f7414bcd-f0cc-4d3e-811a-b5ac75f7ddb8", }, ], "msg": "success", }, ) resp = self.client.fetch_the_access_policies_assigned_to_a_user( user_id=user_id, only_user_policies=False ) @responses.activate def test_assign_access_policy_to_user_group(self) -> None: """3.21 Assign Access Policy to User Group""" user_group_id = UserGroupId("23676a54-382e-4121-aa80-878d2d9bacaa") responses.put( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}/access_policies", match=[ matchers.header_matcher(self.common_headers), matchers.json_params_matcher( {"access_policy_ids": ["bbe48a65-2ac1-4bf6-bd65-bc8f9ee7fb75"]} ), ], json={"code": "SUCCESS", "data": None, "msg": "success"}, ) self.client.assign_access_policy_to_user_group( user_group_id=user_group_id, access_policy_ids=[AccessPolicyId("bbe48a65-2ac1-4bf6-bd65-bc8f9ee7fb75")], ) @responses.activate def test_fetch_the_access_policies_assigned_to_a_user_group(self) -> None: """3.22 Fetch the Access Policies Assigned to a User Group""" user_group_id = UserGroupId("23676a54-382e-4121-aa80-878d2d9bacaa") responses.get( f"https://{self.host}/api/v1/developer/user_groups/{user_group_id}/access_policies", match=[matchers.header_matcher(self.common_headers)], json={ "code": "SUCCESS", "data": [ { "id": "89a4ca95-1502-4ae7-954f-d986b67afe5c", "name": "Default Site Policy", "resources": [ { "id": "fd2a06e2-81af-4cf4-9bd5-8bceb5e7b7d7", "type": "door_group", } ], "schedule_id": "6b79d12a-2a6e-4463-949c-f1a98fff40d2", }, { "id": "bbe48a65-2ac1-4bf6-bd65-bc8f9ee7fb75", "name": "Access Policy Name", "resources": [], "schedule_id": "f7414bcd-f0cc-4d3e-811a-b5ac75f7ddb8", }, ], "msg": "success", }, ) resp = self.client.fetch_the_access_policies_assigned_to_a_user_group( user_group_id=user_group_id, )