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
|
from helpers import faker
from nio.crypto import (
DeviceStore,
GroupSessionStore,
InboundGroupSession,
OlmAccount,
OutboundGroupSession,
OutboundSession,
SessionStore,
)
BOB_ID = "@bob:example.org"
BOB_DEVICE = "AGMTSWVYML"
BOB_CURVE = "T9tOKF+TShsn6mk1zisW2IBsBbTtzDNvw99RBFMJOgI"
BOB_ONETIME = "6QlQw3mGUveS735k/JDaviuoaih5eEi6S1J65iHjfgU"
TEST_ROOM = "!test:example.org"
class TestClass:
def test_session_store(self):
account = OlmAccount()
store = SessionStore()
session = OutboundSession(account, BOB_CURVE, BOB_ONETIME)
assert session not in store
assert len(store.values()) == 0
assert not store.get(BOB_CURVE)
assert store.add(BOB_CURVE, session)
assert len(store.values()) == 1
assert session in store
assert not store.add(BOB_CURVE, session)
assert len(store.values()) == 1
assert session in store
assert (BOB_CURVE, [session]) == list(store.items())[0]
def test_session_store_order(self):
alice = OlmAccount()
bob = OlmAccount()
bob_curve = bob.identity_keys["curve25519"]
bob.generate_one_time_keys(2)
store = SessionStore()
first, second = bob.one_time_keys["curve25519"].values()
session2 = OutboundSession(alice, bob_curve, second)
session = OutboundSession(alice, bob_curve, first)
assert session.id != session2.id
assert session not in store
assert store.add(bob_curve, session)
assert len(store[bob_curve]) == 1
assert session in store
assert store.add(bob_curve, session2) is True
print(store.values())
assert len(store[bob_curve]) == 2
session_a, session_b = store[bob_curve]
assert session_a.use_time > session_b.use_time
def test_device_get_by_sender_key(self):
store = DeviceStore()
for _ in range(10):
store.add(faker.olm_device())
device = faker.olm_device()
store.add(device)
fetched_device = store.device_from_sender_key(device.user_id, device.curve25519)
assert fetched_device == device
def test_group_session_store(self):
store = GroupSessionStore()
account = OlmAccount()
out_group = OutboundGroupSession()
session = InboundGroupSession(
out_group.session_key,
account.identity_keys["ed25519"],
BOB_CURVE,
TEST_ROOM,
)
assert session not in store
assert not store.get(TEST_ROOM, BOB_CURVE, session.id)
assert store.add(session)
assert store.get(TEST_ROOM, BOB_CURVE, session.id)
assert session in store
assert not store.add(session)
assert store[TEST_ROOM] == {BOB_CURVE: {session.id: session}}
|
