distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for UPnP port mapping.
2
3TDD: tests written before implementation.
4"""
5
6import pytest
7
8from i2p_transport.upnp import UPnPManager, UPnPMapping
9
10
11class TestUPnPMapping:
12
13 def test_mapping_creation(self):
14 """UPnPMapping fields are stored correctly."""
15 m = UPnPMapping(
16 external_port=15000,
17 internal_port=15000,
18 protocol="TCP",
19 description="I2P NTCP2",
20 )
21 assert m.external_port == 15000
22 assert m.internal_port == 15000
23 assert m.protocol == "TCP"
24 assert m.description == "I2P NTCP2"
25
26 def test_mapping_default_description(self):
27 m = UPnPMapping(external_port=15001, internal_port=15001, protocol="UDP")
28 assert m.description == "I2P"
29
30
31class TestUPnPManager:
32
33 def test_not_available_before_discover(self):
34 """is_available is False before discovery."""
35 mgr = UPnPManager()
36 assert mgr.is_available is False
37
38 def test_ssdp_request_format(self):
39 """SSDP M-SEARCH request has correct format."""
40 mgr = UPnPManager()
41 req = mgr.build_ssdp_request()
42 text = req.decode("utf-8")
43 assert text.startswith("M-SEARCH * HTTP/1.1\r\n")
44 assert "HOST: 239.255.255.250:1900\r\n" in text
45 assert "MAN: \"ssdp:discover\"\r\n" in text
46 assert "MX:" in text
47 assert "ST:" in text
48 assert text.endswith("\r\n\r\n")
49
50 def test_parse_ssdp_response(self):
51 """Extracts LOCATION URL from SSDP response."""
52 mgr = UPnPManager()
53 response = (
54 b"HTTP/1.1 200 OK\r\n"
55 b"CACHE-CONTROL: max-age=1800\r\n"
56 b"LOCATION: http://192.168.1.1:5000/rootDesc.xml\r\n"
57 b"ST: urn:schemas-upnp-org:device:InternetGatewayDevice:1\r\n"
58 b"\r\n"
59 )
60 location = mgr.parse_ssdp_response(response)
61 assert location == "http://192.168.1.1:5000/rootDesc.xml"
62
63 def test_parse_ssdp_response_no_location(self):
64 """Returns None if no LOCATION header."""
65 mgr = UPnPManager()
66 response = (
67 b"HTTP/1.1 200 OK\r\n"
68 b"CACHE-CONTROL: max-age=1800\r\n"
69 b"\r\n"
70 )
71 assert mgr.parse_ssdp_response(response) is None
72
73 def test_soap_request_format(self):
74 """SOAP XML for AddPortMapping is well-formed."""
75 mgr = UPnPManager()
76 mgr._service_type = "urn:schemas-upnp-org:service:WANIPConnection:1"
77 xml_str = mgr.build_soap_request(
78 action="AddPortMapping",
79 args={
80 "NewRemoteHost": "",
81 "NewExternalPort": "15000",
82 "NewProtocol": "TCP",
83 "NewInternalPort": "15000",
84 "NewInternalClient": "192.168.1.100",
85 "NewEnabled": "1",
86 "NewPortMappingDescription": "I2P",
87 "NewLeaseDuration": "0",
88 },
89 )
90 assert "AddPortMapping" in xml_str
91 assert "NewExternalPort" in xml_str
92 assert "15000" in xml_str
93 assert "urn:schemas-upnp-org:service:WANIPConnection:1" in xml_str
94 # Basic XML well-formedness check
95 import xml.etree.ElementTree as ET
96 ET.fromstring(xml_str) # Should not raise
97
98 def test_parse_ssdp_case_insensitive(self):
99 """LOCATION header parsing is case-insensitive."""
100 mgr = UPnPManager()
101 response = (
102 b"HTTP/1.1 200 OK\r\n"
103 b"location: http://10.0.0.1:1234/desc.xml\r\n"
104 b"\r\n"
105 )
106 location = mgr.parse_ssdp_response(response)
107 assert location == "http://10.0.0.1:1234/desc.xml"
108
109 def test_parse_ssdp_empty(self):
110 mgr = UPnPManager()
111 assert mgr.parse_ssdp_response(b"") is None
112
113 def test_parse_ssdp_binary_garbage(self):
114 mgr = UPnPManager()
115 assert mgr.parse_ssdp_response(b"\xff\xfe\xfd") is None
116
117
118class TestUPnPManagerState:
119 def test_initial_state(self):
120 mgr = UPnPManager()
121 assert mgr._control_url is None
122 assert mgr._service_type is None
123 assert mgr._mappings == []
124 assert mgr._external_ip is None
125
126 def test_available_after_control_url(self):
127 mgr = UPnPManager()
128 mgr._control_url = "http://192.168.1.1/control"
129 assert mgr.is_available is True
130
131
132class TestUPnPManagerWithoutDiscovery:
133 """Operations that require discovery should fail gracefully."""
134
135 @pytest.mark.asyncio
136 async def test_add_mapping_returns_false(self):
137 mgr = UPnPManager()
138 m = UPnPMapping(9700, 9700, "TCP")
139 assert await mgr.add_mapping(m) is False
140
141 @pytest.mark.asyncio
142 async def test_remove_mapping_returns_false(self):
143 mgr = UPnPManager()
144 m = UPnPMapping(9700, 9700, "TCP")
145 assert await mgr.remove_mapping(m) is False
146
147 @pytest.mark.asyncio
148 async def test_get_external_ip_returns_none(self):
149 mgr = UPnPManager()
150 assert await mgr.get_external_ip() is None
151
152 @pytest.mark.asyncio
153 async def test_remove_all_empty(self):
154 mgr = UPnPManager()
155 await mgr.remove_all_mappings() # no crash
156
157 @pytest.mark.asyncio
158 async def test_soap_call_returns_none(self):
159 mgr = UPnPManager()
160 assert await mgr._soap_call("<xml/>", "TestAction") is None
161
162
163class TestGetLocalIP:
164 def test_returns_string(self):
165 ip = UPnPManager._get_local_ip()
166 assert isinstance(ip, str)
167 # Real IP or fallback
168 assert "." in ip
169
170
171class TestSOAPRequestVariants:
172 def test_delete_port_mapping(self):
173 mgr = UPnPManager()
174 mgr._service_type = "urn:schemas-upnp-org:service:WANIPConnection:1"
175 xml = mgr.build_soap_request("DeletePortMapping", {
176 "NewRemoteHost": "",
177 "NewExternalPort": "9700",
178 "NewProtocol": "TCP",
179 })
180 assert "DeletePortMapping" in xml
181 assert "<NewExternalPort>9700</NewExternalPort>" in xml
182
183 def test_get_external_ip_request(self):
184 mgr = UPnPManager()
185 mgr._service_type = "urn:schemas-upnp-org:service:WANIPConnection:1"
186 xml = mgr.build_soap_request("GetExternalIPAddress", {})
187 assert "GetExternalIPAddress" in xml
188 assert "s:Envelope" in xml
189
190 def test_uses_default_service_type(self):
191 mgr = UPnPManager()
192 mgr._service_type = None
193 xml = mgr.build_soap_request("TestAction", {})
194 # Should use first service type as default
195 from i2p_transport.upnp import _SERVICE_TYPES
196 assert _SERVICE_TYPES[0] in xml
197
198 def test_soap_xml_is_valid(self):
199 import xml.etree.ElementTree as ET
200 mgr = UPnPManager()
201 mgr._service_type = "urn:schemas-upnp-org:service:WANPPPConnection:1"
202 xml = mgr.build_soap_request("GetExternalIPAddress", {})
203 ET.fromstring(xml) # must not raise