119 lines
6.5 KiB
Python
119 lines
6.5 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*) printf 'PK' > "$PWD/Min Bog.epub"; echo "Wrote $PWD/Min Bog.epub (2 B)" >&2 ;;
|
|
ok-m4b) head -c 1000 /dev/zero > "$PWD/Lydbog æøå.m4b"; printf 'trin 2\rtrin 3\n' >&2; echo "Wrote $PWD/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}))
|
|
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")
|
|
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"]).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"] == "Lydbog æøå.m4b" and "trin 3" in sb["log"])
|
|
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"])
|
|
u = vent(c.post("/e-pub/start", data={"link": "ok-epub 2"}).json()["id"])
|
|
time.sleep(5)
|
|
tjek("ikke-hentet fil udløber efter TTL", c.get(f"/e-pub/status/{u['id']}").json()["tilstand"] == "udloebet"
|
|
and not (config.EPUB_SCRATCH / u["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)
|