251 lines
9.4 KiB
Python
251 lines
9.4 KiB
Python
"""Clips: Marke setzen, Fenster ziehen, schneiden."""
|
|
|
|
import httpx
|
|
|
|
from app import clips, db, 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 stuecke_mocken(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)
|
|
|
|
|
|
def test_aufnahmen_bekommen_ihren_film(client, zuschauer, hostmodus, film, monkeypatch):
|
|
"""Zuweisung schlägt den Abend am gleichen Datum; sonst kein Film."""
|
|
stuecke_mocken(monkeypatch)
|
|
client.post("/api/watched", json={"movie_id": film(1)})
|
|
with db.connect() as conn:
|
|
conn.execute("UPDATE watched SET watched_at = '2026-09-23T20:41:00+00:00'")
|
|
|
|
liste = client.get("/api/aufnahmen").json()["stuecke"]
|
|
assert [s["start"] for s in liste] == [
|
|
"2026-09-25T19:00:00+00:00",
|
|
"2026-09-24T00:30:00+00:00",
|
|
"2026-09-23T20:00:00+00:00",
|
|
]
|
|
# Der Abend vom 23. gilt auch für die Nacht danach
|
|
assert liste[2]["film"]["id"] == film(1) and liste[2]["zugewiesen"] is False
|
|
assert liste[1]["film"]["id"] == film(1)
|
|
assert liste[0]["film"] is None
|
|
|
|
# Zuweisen darf nur der Host
|
|
zuweisung = {"start": "2026-09-25T19:00:00+00:00", "movie_id": film(2)}
|
|
assert client.put("/api/aufnahmen/film", json=zuweisung).status_code == 403
|
|
hostmodus(zuschauer)
|
|
assert zuschauer.put("/api/aufnahmen/film", json=zuweisung).json() == {"zugewiesen": True}
|
|
liste = client.get("/api/aufnahmen").json()["stuecke"]
|
|
assert liste[0]["film"]["title"] == "Film 2" and liste[0]["zugewiesen"] is True
|
|
# Die Zuweisung schlägt den Abend
|
|
zuschauer.put("/api/aufnahmen/film", json={"start": liste[2]["start"], "movie_id": film(2)})
|
|
assert client.get("/api/aufnahmen").json()["stuecke"][2]["film"]["id"] == film(2)
|
|
# Aufheben
|
|
zuschauer.put("/api/aufnahmen/film", json={"start": zuweisung["start"], "movie_id": None})
|
|
assert client.get("/api/aufnahmen").json()["stuecke"][0]["film"] is None
|
|
|
|
|
|
def test_clip_aus_aufnahme(client, zuschauer, hostmodus, film):
|
|
hostmodus(zuschauer)
|
|
start = "2026-09-23T19:00:00+00:00"
|
|
zuschauer.put("/api/aufnahmen/film", json={"start": start, "movie_id": film(4)})
|
|
m = client.post("/api/clips/von-aufnahme", json={"start": start}).json()
|
|
assert m["movie_id"] == film(4) and m["title"] == "Film 4"
|
|
# Die Marke liegt kurz nach dem Anfang der Aufnahme
|
|
assert m["marke_at"] == "2026-09-23T19:01:00+00:00"
|
|
# Ohne Film gibt es die Marke trotzdem — unter Sonstige
|
|
m = client.post("/api/clips/von-aufnahme", json={"start": "2026-09-25T19:00:00+00:00"}).json()
|
|
assert m["movie_id"] is None
|
|
|
|
|
|
def test_taste_c_nimmt_den_film_der_laufenden_aufnahme(client, zuschauer, hostmodus, film, monkeypatch):
|
|
from datetime import timedelta
|
|
|
|
start = clips._iso(clips._jetzt() - timedelta(seconds=100))
|
|
|
|
async def segmente(von, bis):
|
|
return [{"start": start, "duration": 60}]
|
|
|
|
monkeypatch.setattr(clips, "_segmente", segmente)
|
|
# Es läuft eine Aufnahme, aber ohne Film: dann wie bisher der neueste Abend
|
|
client.post("/api/watched", json={"movie_id": film(1)})
|
|
with db.connect() as conn:
|
|
conn.execute("UPDATE watched SET watched_at = '2020-01-01T20:00:00+00:00'")
|
|
assert marke(client)["movie_id"] == film(1)
|
|
# Zugewiesen: der Film der Aufnahme
|
|
hostmodus(zuschauer)
|
|
zuschauer.put("/api/aufnahmen/film", json={"start": start, "movie_id": film(3)})
|
|
assert marke(client)["movie_id"] == film(3)
|
|
|
|
|
|
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_aufnahme_loescht_nur_der_host(client, zuschauer, hostmodus, film, monkeypatch):
|
|
geloescht = []
|
|
|
|
async def loeschen(start, dauer):
|
|
geloescht.append((start, dauer))
|
|
return 2
|
|
|
|
monkeypatch.setattr(clips, "aufnahme_loeschen", loeschen)
|
|
start = "2026-09-23T19:00:00+00:00"
|
|
assert client.delete("/api/aufnahmen", params={"start": start, "dauer": 60}).status_code == 403
|
|
|
|
hostmodus(zuschauer)
|
|
zuschauer.put("/api/aufnahmen/film", json={"start": start, "movie_id": film(1)})
|
|
antwort = zuschauer.delete("/api/aufnahmen", params={"start": start, "dauer": 60})
|
|
assert antwort.json() == {"segmente": 2}
|
|
assert geloescht == [(start, 60.0)]
|
|
# Mit der Aufnahme geht auch die Zuweisung
|
|
assert clips.film_der_aufnahme(start) is None
|