distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SAM utility functions."""
2
3import base64
4
5import pytest
6
7from i2p_sam.utils import parse_params, negotiate_version, generate_transient_destination
8
9
10class TestParseParams:
11 """Tests for parse_params()."""
12
13 def test_parse_params(self):
14 """Basic key=value parsing."""
15 result = parse_params("MIN=3.0 MAX=3.3")
16 assert result == {"MIN": "3.0", "MAX": "3.3"}
17
18 def test_parse_params_quoted(self):
19 """Quoted values with spaces."""
20 result = parse_params('ID="my session" STYLE=STREAM')
21 assert result["ID"] == "my session"
22 assert result["STYLE"] == "STREAM"
23
24 def test_parse_params_empty(self):
25 """Empty string returns empty dict."""
26 assert parse_params("") == {}
27
28 def test_parse_params_single(self):
29 """Single key=value pair."""
30 result = parse_params("RESULT=OK")
31 assert result == {"RESULT": "OK"}
32
33 def test_parse_params_preserves_case_in_values(self):
34 """Keys are uppercased, values preserve case."""
35 result = parse_params("style=Stream")
36 assert result["STYLE"] == "Stream"
37
38
39class TestNegotiateVersion:
40 """Tests for negotiate_version()."""
41
42 def test_negotiate_version_exact(self):
43 """Exact match — client wants exactly 3.3."""
44 result = negotiate_version("3.3", "3.3", ["3.0", "3.1", "3.2", "3.3"])
45 assert result == "3.3"
46
47 def test_negotiate_version_range(self):
48 """Range negotiation — pick highest mutual version."""
49 result = negotiate_version("3.0", "3.3", ["3.0", "3.1", "3.2", "3.3"])
50 assert result == "3.3"
51
52 def test_negotiate_version_range_limited(self):
53 """Server supports fewer versions than client requests."""
54 result = negotiate_version("3.0", "3.3", ["3.0", "3.1"])
55 assert result == "3.1"
56
57 def test_negotiate_version_no_match(self):
58 """No overlap returns None."""
59 result = negotiate_version("3.4", "3.5", ["3.0", "3.1", "3.2", "3.3"])
60 assert result is None
61
62 def test_negotiate_version_client_below(self):
63 """Client max below server min."""
64 result = negotiate_version("2.0", "2.9", ["3.0", "3.1", "3.2", "3.3"])
65 assert result is None
66
67
68class TestGenerateTransientDestination:
69 """Tests for generate_transient_destination()."""
70
71 def test_generate_transient_destination(self):
72 """Produces valid I2P base64 string and non-empty raw bytes."""
73 raw, b64 = generate_transient_destination()
74 assert isinstance(raw, bytes)
75 assert len(raw) > 0
76 assert isinstance(b64, str)
77 assert len(b64) > 0
78 # Verify b64 is decodable using I2P base64 alphabet (-~ instead of +/)
79 padding = 4 - (len(b64) % 4)
80 if padding != 4:
81 b64_padded = b64 + "=" * padding
82 else:
83 b64_padded = b64
84 decoded = base64.b64decode(b64_padded, altchars=b"-~")
85 assert decoded == raw
86
87 def test_generate_transient_destination_unique(self):
88 """Each call produces a different destination."""
89 _, b64_1 = generate_transient_destination()
90 _, b64_2 = generate_transient_destination()
91 assert b64_1 != b64_2