distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SAM v3.3 text protocol parser and formatter."""
2
3import pytest
4
5from i2p_sam.protocol import SAMCommand, SAMReply, SUPPORTED_VERSIONS, MAX_VERSION
6
7
8class TestSAMCommandParse:
9 """Tests for SAMCommand.parse()."""
10
11 def test_parse_hello(self):
12 """Parse HELLO VERSION MIN=3.0 MAX=3.3."""
13 cmd = SAMCommand.parse("HELLO VERSION MIN=3.0 MAX=3.3")
14 assert cmd.verb == "HELLO"
15 assert cmd.opcode == "VERSION"
16 assert cmd.params["MIN"] == "3.0"
17 assert cmd.params["MAX"] == "3.3"
18
19 def test_parse_session_create(self):
20 """Parse SESSION CREATE with multiple params."""
21 cmd = SAMCommand.parse("SESSION CREATE ID=test STYLE=STREAM DESTINATION=TRANSIENT")
22 assert cmd.verb == "SESSION"
23 assert cmd.opcode == "CREATE"
24 assert cmd.params["ID"] == "test"
25 assert cmd.params["STYLE"] == "STREAM"
26 assert cmd.params["DESTINATION"] == "TRANSIENT"
27
28 def test_parse_stream_connect(self):
29 """Parse STREAM CONNECT with base64 destination."""
30 dest_b64 = "AAAA" * 86 + "AA" # ~346 chars, typical base64 dest
31 cmd = SAMCommand.parse(f"STREAM CONNECT ID=test DESTINATION={dest_b64}")
32 assert cmd.verb == "STREAM"
33 assert cmd.opcode == "CONNECT"
34 assert cmd.params["ID"] == "test"
35 assert cmd.params["DESTINATION"] == dest_b64
36
37 def test_parse_ping(self):
38 """Parse PING with data payload."""
39 cmd = SAMCommand.parse("PING keepalive123")
40 assert cmd.verb == "PING"
41 assert cmd.opcode == "keepalive123"
42
43 def test_parse_quoted_values(self):
44 """Handle KEY="value with spaces" in params."""
45 cmd = SAMCommand.parse('SESSION CREATE ID="my session" STYLE=STREAM DESTINATION=TRANSIENT')
46 assert cmd.params["ID"] == "my session"
47 assert cmd.params["STYLE"] == "STREAM"
48
49 def test_parse_case_insensitive_keys(self):
50 """Keys are normalized to uppercase."""
51 cmd = SAMCommand.parse("HELLO VERSION min=3.0 max=3.3")
52 assert "MIN" in cmd.params
53 assert "MAX" in cmd.params
54
55 def test_parse_empty_raises(self):
56 """Empty or whitespace-only input raises ValueError."""
57 with pytest.raises(ValueError):
58 SAMCommand.parse("")
59 with pytest.raises(ValueError):
60 SAMCommand.parse(" ")
61
62 def test_parse_verb_only(self):
63 """Single-word command gets empty opcode and params."""
64 cmd = SAMCommand.parse("QUIT")
65 assert cmd.verb == "QUIT"
66 assert cmd.opcode == ""
67 assert cmd.params == {}
68
69
70class TestSAMCommandStr:
71 """Tests for SAMCommand.__str__()."""
72
73 def test_command_str_roundtrip(self):
74 """Parse then format, verify consistency."""
75 original = "HELLO VERSION MIN=3.0 MAX=3.3"
76 cmd = SAMCommand.parse(original)
77 formatted = str(cmd)
78 # Re-parse the formatted string
79 cmd2 = SAMCommand.parse(formatted)
80 assert cmd2.verb == cmd.verb
81 assert cmd2.opcode == cmd.opcode
82 assert cmd2.params == cmd.params
83
84 def test_str_no_params(self):
85 """Format command with no params."""
86 cmd = SAMCommand("QUIT", "", {})
87 assert str(cmd).strip() == "QUIT"
88
89
90class TestSAMReply:
91 """Tests for SAMReply static methods."""
92
93 def test_hello_ok(self):
94 assert SAMReply.hello_ok() == "HELLO REPLY RESULT=OK VERSION=3.3\n"
95 assert SAMReply.hello_ok("3.1") == "HELLO REPLY RESULT=OK VERSION=3.1\n"
96
97 def test_hello_noversion(self):
98 assert SAMReply.hello_noversion() == "HELLO REPLY RESULT=NOVERSION\n"
99
100 def test_session_ok(self):
101 reply = SAMReply.session_ok("AAAA")
102 assert reply == "SESSION STATUS RESULT=OK DESTINATION=AAAA\n"
103
104 def test_session_error(self):
105 reply = SAMReply.session_error("DUPLICATED_ID", "Already exists")
106 assert "RESULT=DUPLICATED_ID" in reply
107 assert "MESSAGE=" in reply
108 assert reply.endswith("\n")
109
110 def test_stream_ok(self):
111 assert SAMReply.stream_ok() == "STREAM STATUS RESULT=OK\n"
112
113 def test_stream_error(self):
114 reply = SAMReply.stream_error("CANT_REACH_PEER", "Timeout")
115 assert "RESULT=CANT_REACH_PEER" in reply
116 assert reply.endswith("\n")
117
118 def test_dest_reply(self):
119 reply = SAMReply.dest_reply("test.i2p", "AAAA")
120 assert reply == "DEST REPLY RESULT=OK NAME=test.i2p DESTINATION=AAAA\n"
121
122 def test_dest_not_found(self):
123 reply = SAMReply.dest_not_found("unknown.i2p")
124 assert "KEY_NOT_FOUND" in reply
125 assert "unknown.i2p" in reply
126
127 def test_naming_reply(self):
128 reply = SAMReply.naming_reply("test.i2p", "AAAA")
129 assert reply == "NAMING REPLY RESULT=OK NAME=test.i2p VALUE=AAAA\n"
130
131 def test_pong(self):
132 assert SAMReply.pong("keepalive123") == "PONG keepalive123\n"
133
134
135class TestConstants:
136 """Tests for SAM protocol constants."""
137
138 def test_supported_versions(self):
139 assert "3.0" in SUPPORTED_VERSIONS
140 assert "3.3" in SUPPORTED_VERSIONS
141 assert MAX_VERSION == "3.3"