distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for piece management — TDD: tests before implementation."""
2
3import pytest
4from i2p_apps.snark.pieces import PieceManager, PieceState
5
6
7class TestPieceManager:
8 def test_creation(self):
9 pm = PieceManager(total_pieces=10, piece_length=16384, total_size=163840)
10 assert pm.total_pieces == 10
11 assert pm.completed_count == 0
12
13 def test_initial_state(self):
14 pm = PieceManager(total_pieces=5, piece_length=16384, total_size=81920)
15 for i in range(5):
16 assert pm.state(i) == PieceState.MISSING
17
18 def test_mark_complete(self):
19 pm = PieceManager(total_pieces=5, piece_length=16384, total_size=81920)
20 pm.mark_complete(0)
21 assert pm.state(0) == PieceState.COMPLETE
22 assert pm.completed_count == 1
23
24 def test_mark_requested(self):
25 pm = PieceManager(total_pieces=5, piece_length=16384, total_size=81920)
26 pm.mark_requested(2)
27 assert pm.state(2) == PieceState.REQUESTED
28
29 def test_next_needed(self):
30 pm = PieceManager(total_pieces=3, piece_length=16384, total_size=49152)
31 pm.mark_complete(0)
32 available = {0, 1, 2}
33 needed = pm.next_needed(available)
34 assert needed is not None
35 assert needed != 0 # Already complete
36
37 def test_next_needed_none(self):
38 pm = PieceManager(total_pieces=2, piece_length=16384, total_size=32768)
39 pm.mark_complete(0)
40 pm.mark_complete(1)
41 assert pm.next_needed({0, 1}) is None
42
43 def test_is_complete(self):
44 pm = PieceManager(total_pieces=2, piece_length=16384, total_size=32768)
45 assert pm.is_complete is False
46 pm.mark_complete(0)
47 pm.mark_complete(1)
48 assert pm.is_complete is True
49
50 def test_bitfield(self):
51 pm = PieceManager(total_pieces=8, piece_length=16384, total_size=131072)
52 pm.mark_complete(0)
53 pm.mark_complete(3)
54 pm.mark_complete(7)
55 bf = pm.bitfield()
56 assert len(bf) == 1 # 8 pieces = 1 byte
57 assert bf[0] & 0x80 # bit 0
58 assert bf[0] & 0x10 # bit 3
59 assert bf[0] & 0x01 # bit 7
60
61 def test_percent_complete(self):
62 pm = PieceManager(total_pieces=4, piece_length=16384, total_size=65536)
63 pm.mark_complete(0)
64 pm.mark_complete(1)
65 assert pm.percent_complete == 50.0