151 lines
8.6 KiB
Python
151 lines
8.6 KiB
Python
# Integrationstest af /e-pub under en RIGTIG uvicorn (baggrundsjobs + subprocess), med et
|
|
# stub-værktøj i stedet for det rigtige og current_user patchet (ingen OIDC). Kør fra repo-roden:
|
|
# python tests/epub_uvicorn.py
|
|
import json, os, pathlib, sys, tempfile, threading, time
|
|
|
|
ROD = pathlib.Path(tempfile.mkdtemp(prefix="epub-test-"))
|
|
os.environ.update(PORTAL_DATA=str(ROD / "data"), AUTHENTIK_BASE="https://x.invalid", OIDC_CLIENT_ID="x",
|
|
OIDC_CLIENT_SECRET="x", AUTHENTIK_API_TOKEN="x", SESSION_SECRET="x", RP_API_KEY="hemmelig-noegle")
|
|
VAERKTOEJ = ROD / "tools"
|
|
VAERKTOEJ.mkdir()
|
|
for f in ("run.sh", "run-audiobook.sh", "lib.sh"):
|
|
(VAERKTOEJ / f).write_text("#!/bin/bash\n")
|
|
(VAERKTOEJ / "fetch.sh").write_text(r'''#!/bin/bash
|
|
echo "trin 1: $1" >&2
|
|
case "$1" in
|
|
ok-epub*) sleep 1; printf 'PK' > "$PWD/Min Bog.epub"; echo "Wrote $PWD/Min Bog.epub (2 B)" >&2 ;;
|
|
ok-m4b) head -c 1000 /dev/zero > "$PWD/Ane Forfatter - Lydbog æøå.m4b"; printf 'trin 2\rtrin 3\n' >&2; echo "Wrote $PWD/Ane Forfatter - Lydbog æøå.m4b (1000 B)" >&2 ;;
|
|
env) env | sort >&2; [ -d "$TMP" ] && [ -z "$(ls -A)" ] && touch "$PWD/env.epub" ;;
|
|
bad) exit 2 ;;
|
|
gone) exit 3 ;;
|
|
proxy) exit 4 ;;
|
|
slow) sleep 31.7 & wait ;;
|
|
*) echo "noget andet: $RP_API_KEY" >&2; exit 7 ;;
|
|
esac
|
|
''')
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
|
|
os.chdir(pathlib.Path(__file__).resolve().parents[1])
|
|
import config # noqa: E402
|
|
config.EPUB_SETTINGS_PATH.write_text(json.dumps({"tools_dir": str(VAERKTOEJ), "job_timeout_seconds": 3,
|
|
"output_ttl_seconds": 4, "max_concurrent_jobs": 1,
|
|
"cwa_ingest_dir": str(ROD / "cwa"), "abs_dir": str(ROD / "abs"), "auto_send": False}))
|
|
(ROD / "cwa").mkdir()
|
|
(ROD / "abs").mkdir()
|
|
import main # noqa: E402
|
|
import uvicorn # noqa: E402
|
|
import httpx # noqa: E402
|
|
|
|
BRUGER = {"name": "Test", "username": "test", "groups": []}
|
|
main.current_user = lambda request: BRUGER
|
|
srv = uvicorn.Server(uvicorn.Config(main.app, port=8019, log_level="warning"))
|
|
threading.Thread(target=srv.run, daemon=True).start()
|
|
while not srv.started:
|
|
time.sleep(0.1)
|
|
c = httpx.Client(base_url="http://127.0.0.1:8019", headers={"Accept": "application/json"}, timeout=30)
|
|
|
|
|
|
def tjek(navn, ok):
|
|
print(("OK " if ok else "FEJL ") + navn)
|
|
if not ok:
|
|
global fejl
|
|
fejl += 1
|
|
|
|
|
|
def vent(jid, slut=("faerdig", "fejl")):
|
|
for _ in range(100):
|
|
st = c.get(f"/e-pub/status/{jid}").json()
|
|
if st["tilstand"] in slut:
|
|
return st
|
|
time.sleep(0.2)
|
|
return st
|
|
|
|
|
|
fejl = 0
|
|
tjek("ingen adgang uden gruppe", c.get("/e-pub").status_code == 403)
|
|
BRUGER["groups"] = ["authentik Admins"]
|
|
tjek("admin har adgang", c.get("/e-pub").status_code == 200)
|
|
tjek("admin-side", c.get("/admin/epub").status_code == 200)
|
|
tjek("tom input afvist", c.post("/e-pub/start", data={"link": " "}).status_code == 400)
|
|
tjek("styretegn afvist", c.post("/e-pub/start", data={"link": "a\nb"}).status_code == 400)
|
|
tjek("for stor body afvist", c.post("/e-pub/start", data={"link": "x" * 20000}).status_code == 413)
|
|
|
|
a = c.post("/e-pub/start", data={"link": "ok-epub 1"}).json()
|
|
b = c.post("/e-pub/start", data={"link": "ok-m4b"}).json()
|
|
tjek("semafor: andet job venter i kø", c.get(f"/e-pub/status/{b['id']}").json()["tilstand"] == "koe")
|
|
sa = vent(a["id"])
|
|
tjek("epub færdig", sa["tilstand"] == "faerdig" and sa["filnavn"] == "Min Bog.epub")
|
|
tjek("epub kan sendes til CWA", sa["kan_sende"] == "cwa")
|
|
r = c.post(f"/e-pub/send/{a['id']}")
|
|
tjek("sendt til CWA", r.status_code == 200 and r.json()["sendt"] == "CWA"
|
|
and (ROD / "cwa" / "Min Bog.epub").read_bytes() == b"PK" and not list((ROD / "cwa").glob(".*")))
|
|
tjek("ikke to gange", c.post(f"/e-pub/send/{a['id']}").status_code == 400)
|
|
r = c.get(f"/e-pub/hent/{a['id']}")
|
|
tjek("epub content-type", r.headers["content-type"] == "application/epub+zip" and r.content == b"PK")
|
|
tjek("filnavn i Content-Disposition", "Min%20Bog.epub" in r.headers["content-disposition"] or "Min Bog.epub" in r.headers["content-disposition"])
|
|
time.sleep(0.3)
|
|
tjek("slettet efter hentning", c.get(f"/e-pub/status/{a['id']}").json()["tilstand"] == "hentet"
|
|
and not (config.EPUB_SCRATCH / a["id"] / "ud").exists())
|
|
tjek("anden hentning = 410", c.get(f"/e-pub/hent/{a['id']}").status_code == 410)
|
|
sb = vent(b["id"])
|
|
tjek("m4b færdig, \\r-linjer splittet", sb["filnavn"] == "Ane Forfatter - Lydbog æøå.m4b" and "trin 3" in sb["log"])
|
|
r = c.post(f"/e-pub/send/{b['id']}")
|
|
tjek("m4b til Audiobookshelf som Forfatter/Titel", r.json().get("sendt") == "Audiobookshelf"
|
|
and (ROD / "abs" / "Ane Forfatter" / "Lydbog æøå" / "Ane Forfatter - Lydbog æøå.m4b").stat().st_size == 1000)
|
|
tjek("m4b content-type", c.get(f"/e-pub/hent/{b['id']}").headers["content-type"] == "audio/mp4")
|
|
|
|
e = vent(c.post("/e-pub/start", data={"link": "env"}).json()["id"])
|
|
navne = {l.split("=", 1)[0] for l in e["log"] if "=" in l and not l.startswith("trin")}
|
|
tjek("rent env (kun PATH/TMP/RP_* + bash' egne)", navne <= {"PATH", "TMP", "RP_API_KEY", "RP_BASE_URL", "RP_REQUIRED", "PWD", "SHLVL", "_"})
|
|
tjek("RP_REQUIRED=1, nøgle maskeret, frisk mappe", "RP_REQUIRED=1" in e["log"] and "RP_API_KEY=***" in e["log"] and e["tilstand"] == "faerdig")
|
|
|
|
for inp, forventet in (("bad", "Kunne ikke læse linket."), ("gone", "Ikke tilgængelig (udløbet?)."), ("proxy", "Serverfejl: proxy")):
|
|
s = vent(c.post("/e-pub/start", data={"link": inp}).json()["id"])
|
|
tjek(f"exit for {inp}", s["tilstand"] == "fejl" and s["besked"].startswith(forventet))
|
|
s = vent(c.post("/e-pub/start", data={"link": "andet"}).json()["id"])
|
|
tjek("generisk fejl m. sidste linjer, ingen nøgle", s["besked"].startswith("Noget gik galt.") and "hemmelig" not in s["besked"])
|
|
t0 = time.time()
|
|
s = vent(c.post("/e-pub/start", data={"link": "slow"}).json()["id"])
|
|
tjek("timeout dræber jobbet", s["tilstand"] == "fejl" and "lang tid" in s["besked"] and time.time() - t0 < 8)
|
|
tjek("underprocesser dræbt med", os.system("pgrep -f '^sleep 31.7$' >/dev/null") != 0)
|
|
|
|
BRUGER.update(username="anden", groups=["Bogklub"])
|
|
tjek("andre brugeres job = 403 uden gruppe", c.get(f"/e-pub/status/{b['id']}").status_code == 403)
|
|
config.EPUB_SETTINGS_PATH.write_text(json.dumps({**json.loads(config.EPUB_SETTINGS_PATH.read_text()), "allowed_group": "Bogklub"}))
|
|
tjek("gruppe giver adgang", c.get("/e-pub").status_code == 200)
|
|
tjek("men ikke til andres job", c.get(f"/e-pub/status/{b['id']}").status_code == 404)
|
|
tjek("og ikke admin", c.get("/admin/epub").status_code == 403)
|
|
|
|
BRUGER.update(username="test", groups=["authentik Admins"])
|
|
import epub # noqa: E402
|
|
indst = json.loads(config.EPUB_SETTINGS_PATH.read_text())
|
|
config.EPUB_SETTINGS_PATH.write_text(json.dumps({**indst, "auto_send": True, "output_ttl_seconds": 30}))
|
|
u = vent(c.post("/e-pub/start", data={"link": "ok-epub 3"}).json()["id"])
|
|
time.sleep(0.5)
|
|
u = c.get(f"/e-pub/status/{u['id']}").json()
|
|
tjek("auto-send til CWA", u["sendt"] == "CWA" and (ROD / "cwa" / "Min Bog (2).epub").exists() and u["tilstand"] == "faerdig")
|
|
|
|
# Genstart: en færdig fil overlever, et job der var i gang bliver "afbrudt".
|
|
k = vent(c.post("/e-pub/start", data={"link": "slow"}).json()["id"], slut=("koerer",))
|
|
epub._jobs.clear()
|
|
epub._indlaes()
|
|
tjek("færdig fil overlever genstart", c.get(f"/e-pub/status/{u['id']}").json()["tilstand"] == "faerdig"
|
|
and c.get(f"/e-pub/hent/{u['id']}").content == b"PK")
|
|
k2 = c.get(f"/e-pub/status/{k['id']}").json()
|
|
tjek("igangværende job → afbrudt efter genstart", k2["tilstand"] == "fejl" and "genstartet" in k2["besked"])
|
|
time.sleep(3.5) # det gamle job når at time ud; må ikke overskrive noget
|
|
|
|
config.EPUB_SETTINGS_PATH.write_text(json.dumps({**indst, "auto_send": False, "output_ttl_seconds": 4}))
|
|
v = vent(c.post("/e-pub/start", data={"link": "ok-epub 4"}).json()["id"])
|
|
time.sleep(1)
|
|
tjek("ikke-hentet fil findes stadig inden TTL", c.get(f"/e-pub/status/{v['id']}").json()["tilstand"] == "faerdig")
|
|
time.sleep(5)
|
|
tjek("job + fil væk efter TTL", c.get(f"/e-pub/status/{v['id']}").status_code == 404
|
|
and not (config.EPUB_SCRATCH / v["id"]).exists())
|
|
tjek("admin gem validerer", c.post("/admin/epub", data={"tools_dir": "relativ"}, follow_redirects=False).headers["location"].startswith("/admin/epub?saved=Afvist"))
|
|
c.post("/admin/epub", data={"tools_dir": "/x", "max_concurrent_jobs": "99", "job_timeout_seconds": "abc", "rp_required": "on"})
|
|
s = json.loads(config.EPUB_SETTINGS_PATH.read_text())
|
|
tjek("ints klemmes, ugyldige → standard", s["max_concurrent_jobs"] == 8 and s["job_timeout_seconds"] == 1200 and s["rp_required"] is True)
|
|
print("ALT OK" if not fejl else f"{fejl} FEJL")
|
|
srv.should_exit = True
|
|
sys.exit(1 if fejl else 0)
|