distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SAM session registry (SessionsDB)."""
2
3import asyncio
4from unittest.mock import MagicMock
5
6import pytest
7
8from i2p_sam.sessions_db import SessionRecord, SessionsDB
9
10
11def _make_record(nickname="test", style="STREAM"):
12 return SessionRecord(
13 nickname=nickname,
14 style=style,
15 destination=b"\x00" * 32,
16 destination_b64="AAAA",
17 handler=MagicMock(),
18 )
19
20
21class TestSessionRecord:
22 def test_creation(self):
23 handler = MagicMock()
24 r = SessionRecord(
25 nickname="mysession",
26 style="STREAM",
27 destination=b"\x01" * 32,
28 destination_b64="AQEB",
29 handler=handler,
30 )
31 assert r.nickname == "mysession"
32 assert r.style == "STREAM"
33 assert r.destination == b"\x01" * 32
34 assert r.destination_b64 == "AQEB"
35 assert r.handler is handler
36 assert r.subsessions == {}
37
38 def test_subsessions_default(self):
39 r = _make_record()
40 assert r.subsessions == {}
41
42 def test_subsessions_isolated(self):
43 r1 = _make_record("s1")
44 r2 = _make_record("s2")
45 r1.subsessions["sub"] = r2
46 assert "sub" not in _make_record("s3").subsessions
47
48
49class TestSessionsDB:
50 @pytest.mark.asyncio
51 async def test_add_and_get(self):
52 db = SessionsDB()
53 r = _make_record("test1")
54 assert await db.add(r) is True
55 found = await db.get("test1")
56 assert found is r
57
58 @pytest.mark.asyncio
59 async def test_add_duplicate(self):
60 db = SessionsDB()
61 r1 = _make_record("dup")
62 r2 = _make_record("dup")
63 assert await db.add(r1) is True
64 assert await db.add(r2) is False
65
66 @pytest.mark.asyncio
67 async def test_get_missing(self):
68 db = SessionsDB()
69 assert await db.get("nope") is None
70
71 @pytest.mark.asyncio
72 async def test_remove(self):
73 db = SessionsDB()
74 r = _make_record("rem")
75 await db.add(r)
76 removed = await db.remove("rem")
77 assert removed is r
78 assert await db.get("rem") is None
79
80 @pytest.mark.asyncio
81 async def test_remove_missing(self):
82 db = SessionsDB()
83 assert await db.remove("nope") is None
84
85 @pytest.mark.asyncio
86 async def test_has(self):
87 db = SessionsDB()
88 r = _make_record("check")
89 await db.add(r)
90 assert await db.has("check") is True
91 assert await db.has("nope") is False
92
93 @pytest.mark.asyncio
94 async def test_count(self):
95 db = SessionsDB()
96 assert db.count == 0
97 await db.add(_make_record("a"))
98 assert db.count == 1
99 await db.add(_make_record("b"))
100 assert db.count == 2
101 await db.remove("a")
102 assert db.count == 1
103
104 @pytest.mark.asyncio
105 async def test_multiple_styles(self):
106 db = SessionsDB()
107 await db.add(_make_record("stream", "STREAM"))
108 await db.add(_make_record("dgram", "DATAGRAM"))
109 await db.add(_make_record("raw", "RAW"))
110 assert db.count == 3
111 s = await db.get("dgram")
112 assert s.style == "DATAGRAM"