"""Auftrags-System (Phase 2, 24.09.2026): Akten auf der Platte, Wiederaufnahme nach einem Neustart, benannte Nacharbeiten, Zeitlimit, Abbruch. Die Tests laufen als Kindprozess („prozess“); den systemd-Weg prüfen sie am gebauten Befehl.""" import json import os import sys import time import pytest from services import jobengine PY = sys.executable @pytest.fixture def motor(tmp_path, monkeypatch): ordner = tmp_path / "jobs" monkeypatch.setenv("MC_JOBS_DIR", str(ordner)) monkeypatch.setenv("MC_JOBS_ART", "prozess") monkeypatch.setattr(jobengine, "JOBS", {}) monkeypatch.setattr(jobengine, "_PROCS", {}) monkeypatch.setattr(jobengine, "_geladen", False) monkeypatch.setattr(jobengine, "_NACHARBEITEN", dict(jobengine._NACHARBEITEN)) return ordner def _warte(job_id: str, sekunden: float = 30) -> dict: ende = time.time() + sekunden while time.time() < ende: job = jobengine.JOBS[job_id] if job["state"] in ("done", "failed", "canceled"): return job time.sleep(0.05) raise AssertionError(f"Auftrag {job_id} wurde nicht fertig: {jobengine.JOBS[job_id]}") def _oeffentlich(job_id: str) -> dict: return next(j for j in jobengine.public_jobs() if j["id"] == job_id) def test_erfolg_mit_protokoll_und_akte(motor): code = "import sys; print('eins'); sys.stdout.write('10%\\r50%\\r100%\\n'); print('zwei')" job_id = jobengine.start_job([PY, "-c", code], "Probe") job = _warte(job_id) assert (job["state"], job["returncode"]) == ("done", 0) log = _oeffentlich(job_id)["log"] assert log[0].startswith("$ ") and log[1:] == ["eins", "100%", "zwei"] akte = json.loads((motor / f"{job_id}.json").read_text(encoding="utf-8")) assert akte["state"] == "done" and akte["label"] == "Probe" assert (motor / f"{job_id}.exit").read_text(encoding="utf-8").strip() == "0" def test_fehlschlag_mit_code(motor): job = _warte(jobengine.start_job([PY, "-c", "raise SystemExit(3)"], "Kaputt")) assert (job["state"], job["returncode"]) == ("failed", 3) def test_zeitlimit_beendet_den_auftrag(motor): job = _warte(jobengine.start_job([PY, "-c", "import time; time.sleep(30)"], "Hängt", zeitlimit_s=1)) assert job["state"] == "failed" and job["error"] == "Zeitlimit überschritten" assert any("Zeitlimit" in z for z in _oeffentlich(job["id"])["log"]) def test_abbrechen(motor): job_id = jobengine.start_job([PY, "-c", "import time; time.sleep(30)"], "Lang") time.sleep(0.3) assert jobengine.cancel_job(job_id) is True assert _warte(job_id)["state"] == "canceled" assert jobengine.cancel_job(job_id) is False def test_nacharbeit_nur_bei_erfolg_und_mit_daten(motor): aufrufe: list[dict] = [] jobengine.nacharbeit("probe:merken")(lambda **daten: aufrufe.append(daten)) _warte(jobengine.start_job([PY, "-c", "pass"], "Gut", nacharbeit="probe:merken", nacharbeit_daten={"model_id": "m", "role": "coder"})) _warte(jobengine.start_job([PY, "-c", "raise SystemExit(1)"], "Schlecht", nacharbeit="probe:merken", nacharbeit_daten={"x": 1})) assert aufrufe == [{"model_id": "m", "role": "coder"}] with pytest.raises(ValueError, match="Unbekannte Nacharbeit"): jobengine.start_job([PY, "-c", "pass"], "X", nacharbeit="gibt:es-nicht") def test_exklusiv_je_gruppe(motor): erster, _ = jobengine.start_job_exklusiv("maintenance", [PY, "-c", "import time; time.sleep(30)"], "Update A") zweiter, laeuft = jobengine.start_job_exklusiv("maintenance", [PY, "-c", "pass"], "Update B") assert erster and zweiter is None and laeuft["label"] == "Update A" assert jobengine.active_in_group("maintenance")["id"] == erster jobengine.cancel_job(erster) _warte(erster) assert jobengine.active_in_group("maintenance") is None def _akte(ordner, job_id: str, **felder) -> None: ordner.mkdir(parents=True, exist_ok=True) akte = {"id": job_id, "label": "Alt", "state": "running", "group": "download", "art": "systemd", "returncode": None, "started_at": time.time() - 60, "finished_at": None, "zeitlimit_s": 3600, "nacharbeit": None, "nacharbeit_daten": {}, **felder} (ordner / f"{job_id}.json").write_text(json.dumps(akte), encoding="utf-8") def test_wiederaufnahme_fertiger_einheit_mit_nacharbeit(motor): """MC2 startet neu, während ein Download lief; er ist inzwischen fertig — die Rolle wird gesetzt.""" aufrufe: list[dict] = [] jobengine.nacharbeit("probe:rolle")(lambda **daten: aufrufe.append(daten)) _akte(motor, "abc123", nacharbeit="probe:rolle", nacharbeit_daten={"role": "coder"}) (motor / "abc123.exit").write_text("0", encoding="utf-8") assert jobengine.wiederaufnehmen() == 1 assert _warte("abc123")["state"] == "done" assert aufrufe == [{"role": "coder"}] # Ein zweiter Start von MC2 wiederholt die Nacharbeit nicht. jobengine._geladen = False jobengine.JOBS.clear() assert jobengine.wiederaufnehmen() == 0 assert aufrufe == [{"role": "coder"}] def test_wiederaufnahme_verschwundene_einheit(motor, monkeypatch): monkeypatch.setattr(jobengine, "_einheit_lebt", lambda job_id: False) monkeypatch.setattr(jobengine, "_EINHEIT_PRUEFEN_S", 0.0) _akte(motor, "weg456") jobengine.wiederaufnehmen() job = _warte("weg456") assert job["state"] == "failed" and job["error"] == "Der Auftrag endete ohne Rückmeldung." def test_wiederaufnahme_laufender_einheit_bleibt_laufend(motor, monkeypatch): monkeypatch.setattr(jobengine, "_einheit_lebt", lambda job_id: True) monkeypatch.setattr(jobengine, "_EINHEIT_PRUEFEN_S", 0.0) _akte(motor, "lebt789", group="maintenance") jobengine.wiederaufnehmen() time.sleep(0.5) assert jobengine.JOBS["lebt789"]["state"] == "running" assert jobengine.active_in_group("maintenance")["id"] == "lebt789" (motor / "lebt789.exit").write_text("0", encoding="utf-8") assert _warte("lebt789")["state"] == "done" def test_wiederaufnahme_kindprozess_ist_verloren(motor): _akte(motor, "kind01", art="prozess") jobengine.wiederaufnehmen() job = jobengine.JOBS["kind01"] assert job["state"] == "failed" and "neu gestartet" in job["error"] def test_aufraeumen_loescht_alte_akten(motor, monkeypatch): monkeypatch.setattr(jobengine, "BEHALTEN_MAX", 1) ids = [_warte(jobengine.start_job([PY, "-c", "pass"], f"Nr {i}"))["id"] for i in range(3)] sichtbar = [j["id"] for j in jobengine.public_jobs()] assert sichtbar == [ids[-1]] assert not (motor / f"{ids[0]}.json").exists() and not (motor / f"{ids[0]}.log").exists() def test_systemd_befehl_ohne_geheimnis_in_der_befehlszeile(motor, monkeypatch): monkeypatch.setenv("MC_JOBS_ART", "systemd") monkeypatch.setenv("INVOCATION_ID", "von-mc2") befehle: list[list[str]] = [] class Ergebnis: returncode = 0 stdout = stderr = "" monkeypatch.setattr(jobengine.subprocess, "run", lambda befehl, **kw: befehle.append(befehl) or Ergebnis()) monkeypatch.setattr(jobengine, "_beobachten", lambda job_id: None) job_id = jobengine.start_job(["hf", "download", "org/modell"], "Download", env={"HF_TOKEN": "geheim-123"}, zeitlimit_s=600) befehl = befehle[0] assert befehl[0] == "systemd-run" and f"mc2-job-{job_id}" in befehl and "--collect" in befehl assert "RuntimeMaxSec=600" in befehl assert befehl[-3:] == ["hf", "download", "org/modell"] assert not any("geheim-123" in teil for teil in befehl) umgebung = (motor / f"{job_id}.env").read_text(encoding="utf-8") assert "export HF_TOKEN=geheim-123" in umgebung and "INVOCATION_ID" not in umgebung if os.name == "posix": assert (motor / f"{job_id}.env").stat().st_mode & 0o077 == 0 def test_systemd_start_scheitert_sauber(motor, monkeypatch): monkeypatch.setenv("MC_JOBS_ART", "systemd") class Ergebnis: returncode = 1 stdout = "" stderr = "Failed to connect to bus" monkeypatch.setattr(jobengine.subprocess, "run", lambda befehl, **kw: Ergebnis()) job_id = jobengine.start_job(["true"], "Probe") job = jobengine.JOBS[job_id] assert job["state"] == "failed" and "Failed to connect to bus" in job["error"] assert not (motor / f"{job_id}.env").exists() def test_zeilen_aus_ueberschreibt_mit_wagenruecklauf(): assert jobengine.zeilen_aus("a\n 1%\r 50%\r100%\nb\r\n") == ["a", "100%", "b"] assert jobengine.zeilen_aus("läuft\r 7%") == [" 7%"]