distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for DCC helper — TDD: tests before implementation."""
2
3import pytest
4import time
5
6from i2p_apps.i2ptunnel.dcc import DCCHelper, DCCTransfer
7
8
9class TestDCCTransfer:
10 def test_transfer_creation(self):
11 t = DCCTransfer(
12 remote_dest="A" * 516,
13 local_port=12345,
14 dcc_type="SEND",
15 filename="test.txt",
16 )
17 assert t.remote_dest == "A" * 516
18 assert t.local_port == 12345
19 assert t.dcc_type == "SEND"
20 assert t.filename == "test.txt"
21
22 def test_transfer_not_expired(self):
23 t = DCCTransfer(
24 remote_dest="A" * 516,
25 local_port=12345,
26 dcc_type="SEND",
27 expires_at=time.monotonic() + 1800,
28 )
29 assert t.is_expired is False
30
31 def test_transfer_expired(self):
32 t = DCCTransfer(
33 remote_dest="A" * 516,
34 local_port=12345,
35 dcc_type="SEND",
36 expires_at=time.monotonic() - 1,
37 )
38 assert t.is_expired is True
39
40
41class TestDCCHelper:
42 def test_creation(self):
43 h = DCCHelper(local_b32="test.b32.i2p", max_transfers=5)
44 assert h._max_transfers == 5
45 assert len(h._transfers) == 0
46
47 def test_parse_dcc_send(self):
48 h = DCCHelper(local_b32="test.b32.i2p")
49 parts = h._parse_dcc("DCC SEND file.txt 2130706433 1234 5678")
50 assert parts is not None
51 assert parts["type"] == "SEND"
52 assert parts["filename"] == "file.txt"
53 assert parts["port"] == "1234"
54
55 def test_parse_dcc_chat(self):
56 h = DCCHelper(local_b32="test.b32.i2p")
57 parts = h._parse_dcc("DCC CHAT chat 2130706433 1234")
58 assert parts is not None
59 assert parts["type"] == "CHAT"
60
61 def test_parse_invalid(self):
62 h = DCCHelper(local_b32="test.b32.i2p")
63 assert h._parse_dcc("not a dcc command") is None
64
65 def test_max_transfers_enforced(self):
66 h = DCCHelper(local_b32="test.b32.i2p", max_transfers=2)
67 h._transfers["t1"] = DCCTransfer("d1", 10001, "SEND", expires_at=time.monotonic() + 1800)
68 h._transfers["t2"] = DCCTransfer("d2", 10002, "SEND", expires_at=time.monotonic() + 1800)
69 assert h._can_add_transfer() is False
70
71 def test_cleanup_expired(self):
72 h = DCCHelper(local_b32="test.b32.i2p", max_transfers=2)
73 h._transfers["t1"] = DCCTransfer("d1", 10001, "SEND", expires_at=time.monotonic() - 1)
74 h._transfers["t2"] = DCCTransfer("d2", 10002, "SEND", expires_at=time.monotonic() + 1800)
75 h._cleanup_expired()
76 assert len(h._transfers) == 1
77 assert "t2" in h._transfers
78
79 def test_rewrite_outbound_send(self):
80 h = DCCHelper(local_b32="mysite.b32.i2p")
81 result = h.rewrite_outbound("DCC SEND file.txt 2130706433 1234 5678")
82 assert result is not None
83 assert "mysite.b32.i2p" in result
84 assert "file.txt" in result