-
Notifications
You must be signed in to change notification settings - Fork 130
Expand file tree
/
Copy pathtest_encoder.py
More file actions
122 lines (112 loc) · 4.28 KB
/
Copy pathtest_encoder.py
File metadata and controls
122 lines (112 loc) · 4.28 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
"""Tests for pyhap.encoder."""
import json
import tempfile
import uuid
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ed25519
from pyhap import encoder
from pyhap.const import HAP_PERMISSIONS
from pyhap.state import State
from pyhap.util import generate_mac
def test_persist_and_load():
"""Stores an Accessory and then loads the stored state into another
Accessory. Tests if the two accessories have the same property values.
"""
mac = generate_mac()
_pk = ed25519.Ed25519PrivateKey.generate()
sample_client_pk = _pk.public_key()
state = State(mac=mac)
admin_client_uuid = uuid.uuid1()
admin_client_bytes = str(admin_client_uuid).upper().encode("utf-8")
state.add_paired_client(
admin_client_bytes,
sample_client_pk.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
),
HAP_PERMISSIONS.ADMIN,
)
assert state.is_admin(admin_client_uuid)
user_client_uuid = uuid.uuid1()
user_client_bytes = str(user_client_uuid).upper().encode("utf-8")
state.add_paired_client(
user_client_bytes,
sample_client_pk.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
),
HAP_PERMISSIONS.USER,
)
assert not state.is_admin(user_client_uuid)
config_loaded = State()
config_loaded.config_version += 2 # change the default state.
enc = encoder.AccessoryEncoder()
with tempfile.TemporaryFile(mode="r+") as fp:
enc.persist(fp, state)
fp.seek(0)
enc.load_into(fp, config_loaded)
assert state.mac == config_loaded.mac
assert state.private_key.private_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PrivateFormat.Raw,
encryption_algorithm=serialization.NoEncryption(),
) == config_loaded.private_key.private_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PrivateFormat.Raw,
encryption_algorithm=serialization.NoEncryption(),
)
assert state.public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
) == config_loaded.public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
)
assert state.config_version == config_loaded.config_version
assert state.paired_clients == config_loaded.paired_clients
assert state.client_properties == config_loaded.client_properties
def test_migration_to_include_client_properties():
"""Verify we build client properties if its missing since it was not present in older versions."""
mac = generate_mac()
_pk = ed25519.Ed25519PrivateKey.generate()
sample_client_pk = _pk.public_key()
state = State(mac=mac)
admin_client_uuid = uuid.uuid1()
admin_client_bytes = str(admin_client_uuid).upper().encode("utf-8")
state.add_paired_client(
admin_client_bytes,
sample_client_pk.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
),
HAP_PERMISSIONS.ADMIN,
)
assert state.is_admin(admin_client_uuid)
user_client_uuid = uuid.uuid1()
user_client_bytes = str(user_client_uuid).upper().encode("utf-8")
state.add_paired_client(
user_client_bytes,
sample_client_pk.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
),
HAP_PERMISSIONS.USER,
)
assert not state.is_admin(user_client_uuid)
config_loaded = State()
config_loaded.config_version += 2 # change the default state.
enc = encoder.AccessoryEncoder()
with tempfile.TemporaryFile(mode="r+") as fp:
enc.persist(fp, state)
fp.seek(0)
loaded = json.load(fp)
fp.seek(0)
del loaded["client_properties"]
json.dump(loaded, fp)
fp.truncate()
fp.seek(0)
enc.load_into(fp, config_loaded)
# When client_permissions are missing, all clients
# are imported as admins for backwards compatibility
assert config_loaded.is_admin(admin_client_uuid)
assert config_loaded.is_admin(user_client_uuid)