diff options
| author | historia <historiavg@proton.me> | 2026-09-07 06:47:47 -0400 |
|---|---|---|
| committer | historia <historiavg@proton.me> | 2026-09-07 06:47:47 -0400 |
| commit | 84dd2d068317998f6fb59400c534ef5be6b51b53 (patch) | |
| tree | 025293e9d9229e02960374771ae522d9de2628ce /lib/tests/test_ui.py | |
| parent | 39b0f2bbed74f6487a41b82501ae3c6799e4b5c4 (diff) | |
| download | producer-main.tar.gz | |
Diffstat (limited to 'lib/tests/test_ui.py')
| -rw-r--r-- | lib/tests/test_ui.py | 169 |
1 files changed, 0 insertions, 169 deletions
diff --git a/lib/tests/test_ui.py b/lib/tests/test_ui.py deleted file mode 100644 index 156b3e8..0000000 --- a/lib/tests/test_ui.py +++ /dev/null @@ -1,169 +0,0 @@ -import hashlib -import subprocess -import sys - -import pytest - -from producer import ui - - -class _FakeResp: - def __init__(self, payload: bytes, length: str | None = None): - self._payload = payload - self.headers = {"Content-Length": length} if length else {} - - def read(self, n=-1): - if not self._payload: - return b"" - out, self._payload = self._payload[:n], self._payload[n:] - return out - - def __enter__(self): - return self - - def __exit__(self, *_a): - return False - - -def test_fmt_bytes(): - assert ui.fmt_bytes(512) == "512 B" - assert ui.fmt_bytes(2048) == "2.0 KB" - assert ui.fmt_bytes(8 << 20) == "8.0 MB" - assert ui.fmt_bytes(None) == "?" - - -def test_fmt_secs(): - assert ui.fmt_secs(0) == "0:00" - assert ui.fmt_secs(59) == "0:59" - assert ui.fmt_secs(61) == "1:01" - assert ui.fmt_secs(3700) == "1:01:40" - assert ui.fmt_secs(None) == "?" - - -def test_bar_fills(): - assert ui._bar(0.0) == "[" + "-" * 22 + "]" - assert ui._bar(1.0) == "[" + "#" * 22 + "]" - half = ui._bar(0.5) - assert half.count("#") == 11 and half.count("-") == 11 - - -def test_download_writes_file(tmp_path, monkeypatch, capsys): - payload = b"x" * (2 << 20) - sha = hashlib.sha256(payload).hexdigest() - monkeypatch.setattr( - "urllib.request.urlopen", lambda req, timeout=None: _FakeResp(payload, str(len(payload))) - ) - dest = tmp_path / "big.bin" - out = ui.download("https://example.com/big.bin", dest, sha256=sha) - assert out == dest - assert dest.read_bytes() == payload - assert "done" in capsys.readouterr().out - - -def test_download_checksum_mismatch(tmp_path, monkeypatch): - monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=None: _FakeResp(b"abc")) - dest = tmp_path / "f.bin" - with pytest.raises(RuntimeError): - ui.download("https://example.com/f.bin", dest, sha256="0" * 64) - assert not dest.exists() - assert not dest.with_name(dest.name + ".part").exists() - - -def test_download_cached_skips(tmp_path, monkeypatch, capsys): - dest = tmp_path / "cached.bin" - dest.write_bytes(b"hello") - - def boom(*_a, **_k): - raise AssertionError("should not download") - - monkeypatch.setattr("urllib.request.urlopen", boom) - ui.download("https://example.com/cached.bin", dest) - assert dest.read_bytes() == b"hello" - assert "cached" in capsys.readouterr().out - - -def test_download_size_mismatch_redownloads(tmp_path, monkeypatch): - dest = tmp_path / "short.bin" - dest.write_bytes(b"too short") - monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=None: _FakeResp(b"abcd")) - ui.download("https://example.com/short.bin", dest, expected_size=4) - assert dest.read_bytes() == b"abcd" - - -def test_download_milestones_non_tty(tmp_path, monkeypatch, capsys): - payload = b"y" * (8 << 20) - monkeypatch.setattr( - "urllib.request.urlopen", lambda req, timeout=None: _FakeResp(payload, str(len(payload))) - ) - ui.download("https://example.com/m.bin", tmp_path / "m.bin") - out = capsys.readouterr().out - assert "8.0 MB/8.0 MB" in out - assert "done" in out - - -def test_run_piped_forwards_output(monkeypatch, capsys): - monkeypatch.setattr(ui, "is_tty", lambda: False) - rc = ui.run([sys.executable, "-c", "print('hello-uv-output')"], "test run") - assert rc == 0 - assert "hello-uv-output" in capsys.readouterr().out - - -def test_run_piped_raises_on_failure(monkeypatch): - monkeypatch.setattr(ui, "is_tty", lambda: False) - with pytest.raises(subprocess.CalledProcessError): - ui.run([sys.executable, "-c", "raise SystemExit(3)"], "test run") - - -def test_status_non_tty_milestones_and_done(capsys): - s = ui.Status(prefix="[producer] in.wav — ") - s.stage("denoise") - for i in range(1, 11): - s.tick(i, 10) - s.stage_done("denoise", 2.5) - s.stage("dsp") - s.stage_done("dsp", 0.3) - s.finish() - out = capsys.readouterr().out - assert "denoise 10/10 chunks (100%)" in out - assert "denoise done in 2.5s (10/10 chunks)" in out - assert "dsp done in 0.3s" in out - - -def test_status_tty_live_line(monkeypatch, capsys): - monkeypatch.setattr(ui, "is_tty", lambda: True) - s = ui.Status(prefix="p — ") - s.stage("denoise") - s.tick(10, 10) # completion redraws immediately (bypasses the 0.1 s throttle) - out = capsys.readouterr().out - assert "denoise" in out and "10/10" in out - s.finish() - - -def test_progress_milestones_non_tty(capsys): - p = ui.Progress("downloading x", total=1000) - for i in (100, 300, 1000): - p.update(i) - p.close() - out = capsys.readouterr().out - assert "1000 B/1000 B" in out - assert "done (1000 B" in out - - -def test_progress_tty_bar(monkeypatch, capsys): - monkeypatch.setattr(ui, "is_tty", lambda: True) - p = ui.Progress("downloading torch", total=100) - p.update(50) - p.close() - out = capsys.readouterr().out - assert "50 B/100 B (50%)" in out - - -def test_log_clears_live_line(monkeypatch, capsys): - monkeypatch.setattr(ui, "is_tty", lambda: True) - p = ui.Progress("downloading", total=100) - p.update(50) - ui.log("[producer] some message") - out = capsys.readouterr().out - assert "some message" in out - assert out.index("some message") > out.index("downloading") - p.close() |
