unifi_access/tests/test_user.py

786 lines
30 KiB
Python

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,
)