110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
from fastapi.testclient import TestClient
|
|
from pathlib import Path
|
|
from app.main import app
|
|
|
|
c = TestClient(app)
|
|
|
|
|
|
def test_overrides_fall_back_to_server_config(tmp_path, monkeypatch):
|
|
import shutil
|
|
|
|
import streamrip
|
|
from app.core import settings
|
|
from app.services.streamrip_service import _apply_job_overrides, _load_config
|
|
|
|
dl = tmp_path / "dl"
|
|
dl.mkdir()
|
|
cfg = tmp_path / "config.toml"
|
|
pkg = Path(streamrip.__file__).parent / "config.toml"
|
|
text = pkg.read_text()
|
|
# point folder + quality at test values (keep all other sections intact)
|
|
import re
|
|
|
|
text = re.sub(r'folder = ".*"', f'folder = "{dl}"', text, count=1)
|
|
cfg.write_text(text)
|
|
monkeypatch.setattr(settings, "CONFIG_PATH", str(cfg))
|
|
|
|
config = _load_config()
|
|
folder = _apply_job_overrides(config, None, None, None)
|
|
assert folder == str(dl)
|
|
assert config.session.conversion.enabled is False
|
|
|
|
folder2 = _apply_job_overrides(config, 0, "flac", str(dl))
|
|
assert folder2 == str(dl)
|
|
assert config.session.deezer.quality == 0
|
|
assert config.session.conversion.enabled is True
|
|
assert config.session.conversion.codec == "FLAC"
|
|
|
|
|
|
def test_config_arl_masked_roundtrip(tmp_path, monkeypatch):
|
|
from app.core import settings
|
|
|
|
cfg = tmp_path / "config.toml"
|
|
cfg.write_text('[deezer]\narl = "super-secret"\nquality = 1\n')
|
|
monkeypatch.setattr(settings, "CONFIG_PATH", str(cfg))
|
|
|
|
body = c.get("/api/config").json()
|
|
assert "super-secret" not in body["toml"]
|
|
assert "__MASKED__" in body["toml"]
|
|
|
|
# PUT back the masked text: real ARL must survive
|
|
r = c.put("/api/config", json={"toml": body["toml"]})
|
|
assert r.status_code == 200
|
|
assert 'arl = "super-secret"' in cfg.read_text()
|
|
|
|
|
|
def test_jobstore_prunes_old_terminal():
|
|
from app.core import jobstore
|
|
|
|
for _ in range(5):
|
|
j = jobstore.new_job({"x": 1})
|
|
jobstore.update_job(j["job_id"], status="done")
|
|
jobstore._prune(keep=3)
|
|
remaining = [j for j in jobstore.list_jobs() if j["params"] == {"x": 1}]
|
|
assert len(remaining) == 3
|
|
# cleanup test jobs
|
|
for j in remaining:
|
|
jobstore._jobs.pop(j["job_id"], None)
|
|
|
|
|
|
def test_download_rejects_bad_codec_and_folder():
|
|
assert c.post("/api/downloads", json={"media_type": "track", "id": "1", "codec": "WAV"}).status_code == 422
|
|
assert c.post("/api/downloads", json={"media_type": "track", "id": "1", "folder": "/does/not/exist"}).status_code == 422
|
|
assert c.post("/api/downloads/batch", json={"items": [{"media_type": "track", "id": "1"}], "codec": "WAV"}).status_code == 422
|
|
|
|
|
|
def test_corrupt_config_never_leaks_arl(tmp_path, monkeypatch):
|
|
from app.core import settings
|
|
|
|
cfg = tmp_path / "config.toml"
|
|
cfg.write_text('[deezer]\narl = "live-secret"\nBROKEN = = =\n')
|
|
monkeypatch.setattr(settings, "CONFIG_PATH", str(cfg))
|
|
r = c.get("/api/config")
|
|
assert r.status_code == 500
|
|
assert "live-secret" not in r.text
|
|
|
|
|
|
def test_failed_write_keeps_original(tmp_path, monkeypatch):
|
|
import os
|
|
|
|
from app.api import config as cfgmod
|
|
from app.core import settings
|
|
|
|
cfg = tmp_path / "config.toml"
|
|
cfg.write_text('[deezer]\narl = "keep-me"\n')
|
|
monkeypatch.setattr(settings, "CONFIG_PATH", str(cfg))
|
|
monkeypatch.setattr(os, "replace", lambda *a, **k: (_ for _ in ()).throw(OSError("disk")))
|
|
with __import__("pytest").raises(Exception):
|
|
cfgmod._write_config({})
|
|
assert 'arl = "keep-me"' in cfg.read_text()
|
|
|
|
|
|
def test_log_lines_capped():
|
|
from app.core import jobstore
|
|
|
|
job = jobstore.new_job({"x": 2})
|
|
for i in range(600):
|
|
jobstore.append_log(job["job_id"], f"line {i}")
|
|
assert len(jobstore.get_job(job["job_id"])["logs"]) == 500
|
|
jobstore._jobs.pop(job["job_id"], None)
|