246 lines
8.5 KiB
Python
246 lines
8.5 KiB
Python
"""Clips: Marke setzen, Fenster ziehen, schneiden."""
|
|
|
|
import httpx
|
|
|
|
from app import clips, main, punkte
|
|
|
|
|
|
def marke(client):
|
|
antwort = client.post("/api/clips")
|
|
assert antwort.status_code == 200, antwort.text
|
|
return antwort.json()
|
|
|
|
|
|
def test_marke_ist_nur_ein_zeitpunkt(client, film):
|
|
"""Ohne Dreh und ohne Abend gehört die Marke keinem Film; das Fenster
|
|
hat Standardwerte."""
|
|
m = marke(client)
|
|
assert m["movie_id"] is None
|
|
assert (m["vorher"], m["dauer"]) == (clips.VORHER, clips.DAUER)
|
|
assert m["mine"] is True and m["geschnitten"] is False
|
|
assert m["name"] == "Anna"
|
|
|
|
|
|
def test_marke_haengt_am_gezogenen_film(client, zuschauer, hostmodus, film):
|
|
client.post("/api/suggestions", json={"movie_id": film(1)})
|
|
hostmodus(zuschauer)
|
|
zuschauer.post("/api/spin")
|
|
m = marke(client)
|
|
assert m["movie_id"] == film(1)
|
|
assert m["title"] == "Film 1"
|
|
main.letzter_dreh["ergebnis"] = None
|
|
zuschauer.delete("/api/suggestions/alle")
|
|
|
|
|
|
def test_ohne_dreh_zaehlt_der_neueste_abend(client, film):
|
|
client.post("/api/watched", json={"movie_id": film(1)})
|
|
client.post("/api/watched", json={"movie_id": film(2)})
|
|
assert marke(client)["movie_id"] == film(2)
|
|
|
|
|
|
def test_tage_mit_aufnahme(client, monkeypatch):
|
|
async def segmente(von, bis):
|
|
# Ein Abend mit Stück vor und nach Mitternacht, ein weiterer am Tag danach
|
|
return [
|
|
{"start": "2026-09-23T20:00:00Z", "duration": 30},
|
|
{"start": "2026-09-24T00:30:00Z", "duration": 30},
|
|
{"start": "2026-09-25T19:00:00Z", "duration": 30},
|
|
]
|
|
|
|
monkeypatch.setattr(clips, "_segmente", segmente)
|
|
assert client.get("/api/clips/aufnahmen").json()["tage"] == ["2026-09-23", "2026-09-25"]
|
|
|
|
|
|
def test_clip_zu_altem_abend(client, film, monkeypatch):
|
|
from datetime import datetime, timezone
|
|
|
|
client.post("/api/watched", json={"movie_id": film(4)})
|
|
# Kein Abend zu diesem Film
|
|
assert client.post("/api/clips/von-film", json={"movie_id": 999}).status_code == 404
|
|
|
|
# Keine Aufnahme an dem Tag
|
|
async def keine(watched_at):
|
|
return []
|
|
|
|
monkeypatch.setattr(clips, "aufnahmen_des_abends", keine)
|
|
assert client.get(f"/api/clips/von-film/{film(4)}").json()["stuecke"] == []
|
|
antwort = client.post("/api/clips/von-film", json={"movie_id": film(4)})
|
|
assert antwort.status_code == 404 and "Aufnahme" in antwort.json()["detail"]
|
|
|
|
# Zwei Stücke, weil der Stream neu gestartet wurde
|
|
stuecke = [
|
|
{"start": "2026-09-23T19:00:00+00:00", "dauer": 1800.0},
|
|
{"start": "2026-09-23T20:00:00+00:00", "dauer": 600.0},
|
|
]
|
|
|
|
async def da(watched_at):
|
|
return stuecke
|
|
|
|
monkeypatch.setattr(clips, "aufnahmen_des_abends", da)
|
|
assert client.get(f"/api/clips/von-film/{film(4)}").json()["stuecke"] == stuecke
|
|
|
|
# Ohne Wahl: die Marke liegt kurz nach dem Beginn des ersten Stücks
|
|
m = client.post("/api/clips/von-film", json={"movie_id": film(4)}).json()
|
|
assert m["movie_id"] == film(4)
|
|
assert m["marke_at"] == "2026-09-23T19:01:00+00:00"
|
|
|
|
# Mit Wahl: nach dem Beginn des gewählten Stücks
|
|
m = client.post(
|
|
"/api/clips/von-film", json={"movie_id": film(4), "start": stuecke[1]["start"]}
|
|
).json()
|
|
assert m["marke_at"] == "2026-09-23T20:01:00+00:00"
|
|
|
|
|
|
def test_stuecke_eines_abends_nach_beginn_geordnet(monkeypatch):
|
|
import asyncio
|
|
|
|
async def segmente(von, bis):
|
|
return [
|
|
{"start": "2026-09-23T21:00:00Z", "duration": 600.0},
|
|
{"start": "2026-09-23T19:00:00Z", "duration": 1800.04},
|
|
]
|
|
|
|
monkeypatch.setattr(clips, "_segmente", segmente)
|
|
assert asyncio.run(clips.aufnahmen_des_abends("2026-09-23T20:00:00+00:00")) == [
|
|
{"start": "2026-09-23T19:00:00+00:00", "dauer": 1800.0},
|
|
{"start": "2026-09-23T21:00:00+00:00", "dauer": 600.0},
|
|
]
|
|
|
|
|
|
def test_fremde_sehen_nichts_der_host_alles(client, client2, zuschauer, hostmodus):
|
|
m = marke(client)
|
|
assert [c for c in client2.get("/api/clips").json() if c["id"] == m["id"]] == []
|
|
assert client2.patch(f"/api/clips/{m['id']}", json={"titel": "x"}).status_code == 404
|
|
assert client2.get(f"/api/clips/{m['id']}/video").status_code == 404
|
|
hostmodus(zuschauer)
|
|
assert zuschauer.patch(f"/api/clips/{m['id']}", json={"titel": "Vom Host"}).json()["titel"] == "Vom Host"
|
|
assert zuschauer.get("/api/clips").json()[0]["id"] == m["id"]
|
|
|
|
|
|
def test_fenster_wird_begrenzt_und_macht_schnitt_ungueltig(client, monkeypatch):
|
|
m = marke(client)
|
|
geaendert = client.patch(f"/api/clips/{m['id']}", json={"vorher": 5000, "dauer": 1, "titel": " Hallo "}).json()
|
|
assert (geaendert["vorher"], geaendert["dauer"], geaendert["titel"]) == (clips.MAX_VORHER, clips.MIN_DAUER, "Hallo")
|
|
|
|
# Schneiden: der Playback-Server wird nachgestellt
|
|
class Antwort:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
async def aiter_bytes(self):
|
|
yield b"mp4-daten"
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
pass
|
|
|
|
aufrufe = []
|
|
|
|
class Client:
|
|
def __init__(self, **kw):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
pass
|
|
|
|
def stream(self, methode, url, params=None):
|
|
aufrufe.append(params)
|
|
return Antwort()
|
|
|
|
monkeypatch.setattr(clips.httpx, "AsyncClient", Client)
|
|
fertig = client.post(f"/api/clips/{m['id']}/schneiden").json()
|
|
assert fertig["geschnitten"] is True
|
|
assert aufrufe[0]["path"] == "filmabend" and aufrufe[0]["duration"] == clips.MIN_DAUER
|
|
assert (clips.CLIPS_DIR / f"{m['id']}.mp4").read_bytes() == b"mp4-daten"
|
|
video = client.get(f"/api/clips/{m['id']}/video")
|
|
assert video.status_code == 200 and video.content == b"mp4-daten"
|
|
|
|
# Neues Fenster: Datei weg, muss neu geschnitten werden
|
|
neu = client.patch(f"/api/clips/{m['id']}", json={"dauer": 20}).json()
|
|
assert neu["geschnitten"] is False
|
|
assert not (clips.CLIPS_DIR / f"{m['id']}.mp4").exists()
|
|
assert client.get(f"/api/clips/{m['id']}/video").status_code == 404
|
|
|
|
|
|
def test_schneiden_ohne_aufnahme_meldet_502(client, monkeypatch):
|
|
m = marke(client)
|
|
|
|
class Client:
|
|
def __init__(self, **kw):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
pass
|
|
|
|
def stream(self, *a, **kw):
|
|
raise httpx.ConnectError("weg")
|
|
|
|
monkeypatch.setattr(clips.httpx, "AsyncClient", Client)
|
|
assert client.post(f"/api/clips/{m['id']}/schneiden").status_code == 502
|
|
|
|
|
|
def test_loeschen_nimmt_die_datei_mit(client):
|
|
m = marke(client)
|
|
clips.CLIPS_DIR.mkdir(parents=True, exist_ok=True)
|
|
(clips.CLIPS_DIR / f"{m['id']}.mp4").write_bytes(b"x")
|
|
from app import db
|
|
|
|
with db.connect() as conn:
|
|
conn.execute("UPDATE clips SET datei = ? WHERE id = ?", (f"{m['id']}.mp4", m["id"]))
|
|
assert client.delete(f"/api/clips/{m['id']}").json()["removed"] == 1
|
|
assert not (clips.CLIPS_DIR / f"{m['id']}.mp4").exists()
|
|
assert client.get("/api/clips").json() == []
|
|
|
|
|
|
def test_zuschauer_darf_keine_marke_setzen(zuschauer):
|
|
assert zuschauer.post("/api/clips").status_code == 403
|
|
|
|
|
|
def test_segmente_eines_stuecks():
|
|
segmente = [
|
|
{"start": "2026-09-23T19:00:00.5Z"},
|
|
{"start": "2026-09-23T19:00:30.5Z"},
|
|
{"start": "2026-09-23T19:01:00.5Z"},
|
|
{"start": "2026-09-23T21:00:00Z"},
|
|
]
|
|
# Das Stück von 19:00 mit 61 s Dauer: drei Segmente, das von 21:00 nicht
|
|
assert clips._im_stueck(segmente, "2026-09-23T19:00:00+00:00", 61.0) == [
|
|
"2026-09-23T19:00:00.5Z",
|
|
"2026-09-23T19:00:30.5Z",
|
|
"2026-09-23T19:01:00.5Z",
|
|
]
|
|
|
|
|
|
def test_aufnahmen_verwaltet_nur_der_host(client, zuschauer, hostmodus, monkeypatch):
|
|
stuecke = [{"start": "2026-09-23T19:00:00+00:00", "dauer": 60.0}]
|
|
|
|
async def alle():
|
|
return stuecke
|
|
|
|
geloescht = []
|
|
|
|
async def loeschen(start, dauer):
|
|
geloescht.append((start, dauer))
|
|
return 2
|
|
|
|
monkeypatch.setattr(clips, "aufnahmen_alle", alle)
|
|
monkeypatch.setattr(clips, "aufnahme_loeschen", loeschen)
|
|
assert client.get("/api/aufnahmen").status_code == 403
|
|
assert client.delete("/api/aufnahmen", params={"start": stuecke[0]["start"], "dauer": 60}).status_code == 403
|
|
|
|
hostmodus(zuschauer)
|
|
assert zuschauer.get("/api/aufnahmen").json()["stuecke"] == stuecke
|
|
antwort = zuschauer.delete("/api/aufnahmen", params={"start": stuecke[0]["start"], "dauer": 60})
|
|
assert antwort.json() == {"segmente": 2}
|
|
assert geloescht == [(stuecke[0]["start"], 60.0)]
|