distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for net.i2p.util Tier 1 — file & system utilities.
2
3TDD — tests for FileUtil, PasswordManager, ShellCommand, TempDirScanner.
4"""
5
6from __future__ import annotations
7
8import os
9import tempfile
10import zipfile
11
12import pytest
13
14from i2p_util.file_util import (
15 copy_file,
16 rm_dir,
17 extract_zip,
18 verify_zip,
19 file_checksum,
20 read_file,
21 write_file,
22)
23from i2p_util.password import PasswordManager
24from i2p_util.shell import shell_command
25from i2p_util.temp_scanner import scan_temp_dir
26
27
28# ---------------------------------------------------------------------------
29# FileUtil
30# ---------------------------------------------------------------------------
31
32
33class TestFileUtil:
34 """File utility operations."""
35
36 def test_copy_file(self):
37 with tempfile.TemporaryDirectory() as td:
38 src = os.path.join(td, "src.txt")
39 dst = os.path.join(td, "dst.txt")
40 with open(src, "w") as f:
41 f.write("hello")
42 assert copy_file(src, dst) is True
43 assert os.path.exists(dst)
44 with open(dst) as f:
45 assert f.read() == "hello"
46
47 def test_copy_file_missing_source(self):
48 with tempfile.TemporaryDirectory() as td:
49 assert copy_file("/nonexistent", os.path.join(td, "dst")) is False
50
51 def test_write_and_read_file(self):
52 with tempfile.TemporaryDirectory() as td:
53 path = os.path.join(td, "data.bin")
54 data = b"binary data here"
55 assert write_file(path, data) is True
56 result = read_file(path)
57 assert result == data
58
59 def test_read_file_missing(self):
60 assert read_file("/nonexistent/file.bin") is None
61
62 def test_rm_dir(self):
63 with tempfile.TemporaryDirectory() as td:
64 sub = os.path.join(td, "sub")
65 os.makedirs(sub)
66 with open(os.path.join(sub, "f.txt"), "w") as f:
67 f.write("x")
68 assert rm_dir(sub) is True
69 assert not os.path.exists(sub)
70
71 def test_rm_dir_missing(self):
72 assert rm_dir("/nonexistent/dir") is False
73
74 def test_extract_zip(self):
75 with tempfile.TemporaryDirectory() as td:
76 zp = os.path.join(td, "test.zip")
77 with zipfile.ZipFile(zp, "w") as zf:
78 zf.writestr("hello.txt", "world")
79 dest = os.path.join(td, "out")
80 os.makedirs(dest)
81 assert extract_zip(zp, dest) is True
82 assert os.path.exists(os.path.join(dest, "hello.txt"))
83
84 def test_verify_zip_valid(self):
85 with tempfile.TemporaryDirectory() as td:
86 zp = os.path.join(td, "good.zip")
87 with zipfile.ZipFile(zp, "w") as zf:
88 zf.writestr("a.txt", "data")
89 assert verify_zip(zp) is True
90
91 def test_verify_zip_invalid(self):
92 with tempfile.TemporaryDirectory() as td:
93 bad = os.path.join(td, "bad.zip")
94 with open(bad, "wb") as f:
95 f.write(b"not a zip")
96 assert verify_zip(bad) is False
97
98 def test_file_checksum_sha256(self):
99 with tempfile.TemporaryDirectory() as td:
100 path = os.path.join(td, "f.txt")
101 with open(path, "wb") as f:
102 f.write(b"test data")
103 cs = file_checksum(path, "sha256")
104 assert isinstance(cs, str)
105 assert len(cs) == 64 # hex SHA-256
106
107 def test_file_checksum_missing(self):
108 assert file_checksum("/nonexistent", "sha256") is None
109
110
111# ---------------------------------------------------------------------------
112# PasswordManager
113# ---------------------------------------------------------------------------
114
115
116class TestPasswordManager:
117 """Password hashing and verification."""
118
119 def test_create_and_check(self):
120 pm = PasswordManager()
121 hashed = pm.create_hash("my_password")
122 assert hashed is not None
123 assert pm.check_password("my_password", hashed) is True
124
125 def test_wrong_password_fails(self):
126 pm = PasswordManager()
127 hashed = pm.create_hash("correct")
128 assert pm.check_password("wrong", hashed) is False
129
130 def test_generate_password(self):
131 pm = PasswordManager()
132 pwd = pm.generate_password(length=32)
133 assert isinstance(pwd, str)
134 assert len(pwd) >= 32
135
136 def test_different_hashes_for_same_password(self):
137 """Salt should make hashes unique."""
138 pm = PasswordManager()
139 h1 = pm.create_hash("same")
140 h2 = pm.create_hash("same")
141 assert h1 != h2
142
143
144# ---------------------------------------------------------------------------
145# ShellCommand
146# ---------------------------------------------------------------------------
147
148
149class TestShellCommand:
150 """Shell command execution."""
151
152 def test_echo(self):
153 rc, stdout, stderr = shell_command("echo hello")
154 assert rc == 0
155 assert "hello" in stdout
156
157 def test_exit_code(self):
158 rc, _, _ = shell_command("false")
159 assert rc != 0
160
161 def test_timeout(self):
162 rc, _, _ = shell_command("sleep 60", timeout=1.0)
163 assert rc != 0 # should be killed
164
165 def test_stderr(self):
166 rc, _, stderr = shell_command("echo err >&2")
167 assert rc == 0
168 assert "err" in stderr
169
170
171# ---------------------------------------------------------------------------
172# TempDirScanner
173# ---------------------------------------------------------------------------
174
175
176class TestTempDirScanner:
177 """Temp directory cleanup."""
178
179 def test_scan_removes_old_files(self):
180 with tempfile.TemporaryDirectory() as td:
181 old = os.path.join(td, "old.tmp")
182 with open(old, "w") as f:
183 f.write("x")
184 # Set mtime to 1 hour ago
185 old_time = os.path.getmtime(old) - 3600
186 os.utime(old, (old_time, old_time))
187 removed = scan_temp_dir(td, max_age_secs=60)
188 assert old in removed
189 assert not os.path.exists(old)
190
191 def test_scan_keeps_recent_files(self):
192 with tempfile.TemporaryDirectory() as td:
193 recent = os.path.join(td, "recent.tmp")
194 with open(recent, "w") as f:
195 f.write("x")
196 removed = scan_temp_dir(td, max_age_secs=3600)
197 assert len(removed) == 0
198 assert os.path.exists(recent)
199
200 def test_scan_empty_dir(self):
201 with tempfile.TemporaryDirectory() as td:
202 removed = scan_temp_dir(td, max_age_secs=60)
203 assert removed == []