Update tests/test_app.py for media verify endpoint
This commit is contained in:
parent
cadcf8766f
commit
8230e03745
1 changed files with 84 additions and 0 deletions
|
|
@ -901,6 +901,90 @@ def test_uptime_monitor_remove_rejects_invalid_expected_name_before_ssh(monkeypa
|
||||||
assert response.status_code == 400
|
assert response.status_code == 400
|
||||||
|
|
||||||
|
|
||||||
|
def test_media_verify_returns_duration_and_not_suspect_when_within_tolerance(monkeypatch):
|
||||||
|
monkeypatch.setattr(app, "_find_inventory_host", lambda host: {"user": "sascha", "ip": "10.2.1.100"})
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
def fake_ssh(host, command, timeout=30):
|
||||||
|
calls.append((host, command, timeout))
|
||||||
|
return 0, "2967.355000\n", ""
|
||||||
|
|
||||||
|
monkeypatch.setattr(app, "_ssh", fake_ssh)
|
||||||
|
with TestClient(app.app) as client:
|
||||||
|
response = client.post(
|
||||||
|
"/media/verify",
|
||||||
|
headers={"Authorization": "Bearer test-token"},
|
||||||
|
json={"host": "arrapps", "path": "/data/UHD/serien/Show/ep.mkv", "expected_minutes": 49.5},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
body = response.json()
|
||||||
|
assert body["duration_minutes"] == 49.46
|
||||||
|
assert body["suspect"] is False
|
||||||
|
assert calls[0][0] == "sascha@10.2.1.100"
|
||||||
|
assert "bazarrUHD" in calls[0][1]
|
||||||
|
assert "ffprobe" in calls[0][1]
|
||||||
|
|
||||||
|
|
||||||
|
def test_media_verify_flags_truncated_file_as_suspect(monkeypatch):
|
||||||
|
monkeypatch.setattr(app, "_find_inventory_host", lambda host: {"user": "sascha", "ip": "10.2.1.100"})
|
||||||
|
monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (0, "1836.0\n", ""))
|
||||||
|
with TestClient(app.app) as client:
|
||||||
|
response = client.post(
|
||||||
|
"/media/verify",
|
||||||
|
headers={"Authorization": "Bearer test-token"},
|
||||||
|
json={"host": "arrapps", "path": "/data/UHD/serien/Show/ep.mkv", "expected_minutes": 49.5},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
body = response.json()
|
||||||
|
assert body["suspect"] is True
|
||||||
|
assert body["deviation_pct"] > 20
|
||||||
|
|
||||||
|
|
||||||
|
def test_media_verify_rejects_path_outside_data_before_ssh(monkeypatch):
|
||||||
|
monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
|
||||||
|
with TestClient(app.app) as client:
|
||||||
|
response = client.post(
|
||||||
|
"/media/verify",
|
||||||
|
headers={"Authorization": "Bearer test-token"},
|
||||||
|
json={"host": "arrapps", "path": "/etc/passwd"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 400
|
||||||
|
|
||||||
|
|
||||||
|
def test_media_verify_rejects_shell_metacharacters_before_ssh(monkeypatch):
|
||||||
|
monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
|
||||||
|
with TestClient(app.app) as client:
|
||||||
|
response = client.post(
|
||||||
|
"/media/verify",
|
||||||
|
headers={"Authorization": "Bearer test-token"},
|
||||||
|
json={"host": "arrapps", "path": "/data/UHD/serien/a'; rm -rf /'.mkv"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 400
|
||||||
|
|
||||||
|
|
||||||
|
def test_media_verify_rejects_unknown_host_before_ssh(monkeypatch):
|
||||||
|
monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
|
||||||
|
with TestClient(app.app) as client:
|
||||||
|
response = client.post(
|
||||||
|
"/media/verify",
|
||||||
|
headers={"Authorization": "Bearer test-token"},
|
||||||
|
json={"host": "unknown-host", "path": "/data/UHD/serien/ep.mkv"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 400
|
||||||
|
|
||||||
|
|
||||||
|
def test_media_verify_returns_502_on_unparsable_ffprobe_output(monkeypatch):
|
||||||
|
monkeypatch.setattr(app, "_find_inventory_host", lambda host: {"user": "sascha", "ip": "10.2.1.100"})
|
||||||
|
monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (0, "N/A\n", ""))
|
||||||
|
with TestClient(app.app) as client:
|
||||||
|
response = client.post(
|
||||||
|
"/media/verify",
|
||||||
|
headers={"Authorization": "Bearer test-token"},
|
||||||
|
json={"host": "arrapps", "path": "/data/UHD/serien/ep.mkv"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 502
|
||||||
|
|
||||||
|
|
||||||
def test_invalid_log_target_is_rejected_before_ssh():
|
def test_invalid_log_target_is_rejected_before_ssh():
|
||||||
with TestClient(app.app) as client:
|
with TestClient(app.app) as client:
|
||||||
response = client.get(
|
response = client.get(
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue