proxdeploy/tests/auftraege.test.ts

import { afterAll, beforeEach, describe, expect, it } from "vitest";
import { GET as ereignisseGet } from "@/app/api/v1/auftraege/[id]/ereignisse/route";
import { GET as auftragGet } from "@/app/api/v1/auftraege/[id]/route";
import { POST as abbrechenPost } from "@/app/api/v1/auftraege/[id]/abbrechen/route";
import { verschluesseln } from "@/lib/geheimnisse";
import { prisma } from "@/lib/db";
import { phasenAusfuehren } from "@/engine/ablauf/ablauf";
import { auftragAnlegen } from "@/server/services/auftraege";
import { vmidReservieren } from "@/server/services/vmid";
import { ablaufRegistrieren, auftragAusfuehren, verwaisteAuftraegeBeenden } from "@/worker/ausfuehren";
import { datenbankLeeren, datenbankVerfuegbar, testPrisma } from "./helpers/db";
import { angemeldet, anfrage, kontext } from "./helpers/anfragen";

async function serverAnlegen() {
  return await testPrisma.server.create({
    data: {
      name: "rz1",
      titel: "RZ 1",
      url: "https://127.0.0.1:8006",
      tokenId: "a@pve!b",
      tokenGeheimnis: verschluesseln("x"),
      fingerabdruck: "AA",
    },
  });
}

/** Liest einen SSE-Strom, bis `ende` kommt, und gibt die Ereignisse zurück. */
async function stromLesen(
  antwort: Response,
  nebenbei?: () => Promise<void>,
): Promise<Array<{ art: string; daten: unknown; id?: string }>> {
  const leser = antwort.body!.getReader();
  const dekodierer = new TextDecoder();
  let puffer = "";
  const ereignisse: Array<{ art: string; daten: unknown; id?: string }> = [];
  const parallel = nebenbei?.();
  for (;;) {
    const { value, done } = await leser.read();
    if (done) break;
    puffer += dekodierer.decode(value, { stream: true });
    let grenze: number;
    while ((grenze = puffer.indexOf("\n\n")) >= 0) {
      const block = puffer.slice(0, grenze);
      puffer = puffer.slice(grenze + 2);
      const art = /^event: (.+)$/m.exec(block)?.[1];
      const daten = /^data: (.+)$/m.exec(block)?.[1];
      const id = /^id: (.+)$/m.exec(block)?.[1];
      if (art && daten) ereignisse.push({ art, daten: JSON.parse(daten), id });
    }
  }
  await parallel;
  return ereignisse;
}

describe.runIf(datenbankVerfuegbar)("Aufträge, worker und Live-Strom", () => {
  afterAll(async () => {
    await prisma.$disconnect();
  });
  beforeEach(async () => {
    await datenbankLeeren();
  });

  it("VMID-Vergabe: zehn parallele Reservierungen bekommen zehn verschiedene IDs, Belegtes wird übersprungen", async () => {
    const server = await serverAnlegen();
    const ids = await Promise.all(
      Array.from({ length: 10 }, (_, i) =>
        vmidReservieren({
          serverId: server.id,
          auftragId: `a${i}`,
          belegt: new Set([100, 101, 103]),
          von: 100,
        }),
      ),
    );
    expect(new Set(ids).size).toBe(10);
    expect(ids).not.toContain(100);
    expect(ids).not.toContain(103);
    expect(Math.min(...ids)).toBe(102);
    await expect(
      vmidReservieren({ serverId: server.id, auftragId: "x", belegt: new Set(), von: 100, wunsch: 102 }),
    ).rejects.toThrow(/bereits belegt/);
  });

  it("worker führt einen Auftrag aus; SSE liefert Stand, Log und Ende; Reservierungen werden frei", async () => {
    const server = await serverAnlegen();
    const { benutzer, cookie } = await angemeldet("OPERATOR");
    ablaufRegistrieren("AUFRAEUMEN", async ({ auftrag, senke }) => {
      await vmidReservieren({
        serverId: auftrag.serverId,
        auftragId: auftrag.id,
        belegt: new Set(),
        von: 100,
      });
      const ergebnis = await phasenAusfuehren(
        [
          {
            schluessel: "eins",
            titel: "Eins",
            ausfuehren: async (k) => {
              await k.log("Zeile 1");
              return { detail: "fertig" };
            },
          },
          {
            schluessel: "zwei",
            titel: "Zwei",
            ausfuehren: async (k) => {
              await k.log("Zeile 2");
              return {};
            },
          },
        ],
        senke,
      );
      return { ...ergebnis, ergebnis: { ip: "10.0.0.5" } };
    });
    const auftrag = await auftragAnlegen(
      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
      { benutzer, quelle: "WEB" },
      null,
      async () => {},
    );
    expect(auftrag.kurzId).toMatch(/^c-[2-9a-z]{5}$/);

    const strom = await ereignisseGet(
      anfrage("GET", `/api/v1/auftraege/${auftrag.kurzId}/ereignisse`, cookie),
      kontext({ id: auftrag.kurzId }),
    );
    expect(strom.headers.get("content-type")).toContain("text/event-stream");
    const ereignisse = await stromLesen(strom, () => auftragAusfuehren(auftrag.id));

    const staende = ereignisse
      .filter((e) => e.art === "stand")
      .map((e) => (e.daten as { status: string }).status);
    // Je nach Takt sieht der erste Stand den Auftrag wartend oder schon laufend – nie fertig.
    expect(["WARTEND", "LAEUFT"]).toContain(staende[0]);
    expect(staende.at(-1)).toBe("ERFOLGREICH");
    const logs = ereignisse
      .filter((e) => e.art === "log")
      .flatMap((e) => (e.daten as Array<{ nachricht: string }>).map((z) => z.nachricht));
    expect(logs).toEqual(["Zeile 1", "Zeile 2"]);
    expect(ereignisse.at(-1)?.art).toBe("ende");

    const fertig = await (
      await auftragGet(
        anfrage("GET", `/api/v1/auftraege/${auftrag.kurzId}`, cookie),
        kontext({ id: auftrag.kurzId }),
      )
    ).json();
    expect(
      fertig.phasen.map((p: { schluessel: string; status: string }) => `${p.schluessel}:${p.status}`),
    ).toEqual(["eins:ERLEDIGT", "zwei:ERLEDIGT"]);
    expect(fertig.ergebnis).toEqual({ ip: "10.0.0.5" });
    expect(await testPrisma.vmidReservierung.count()).toBe(0);
  });

  it("Wiederverbinden mit Last-Event-ID liefert nur neue Log-Zeilen", async () => {
    const server = await serverAnlegen();
    const { benutzer, cookie } = await angemeldet("OPERATOR");
    ablaufRegistrieren("AUFRAEUMEN", async ({ senke }) =>
      phasenAusfuehren(
        [
          {
            schluessel: "a",
            titel: "A",
            ausfuehren: async (k) => {
              await k.log("alt");
              await k.log("neu");
              return {};
            },
          },
        ],
        senke,
      ),
    );
    const auftrag = await auftragAnlegen(
      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
      { benutzer, quelle: "WEB" },
      null,
      async () => {},
    );
    await auftragAusfuehren(auftrag.id);
    const erste = await testPrisma.logeintrag.findFirstOrThrow({
      where: { auftragId: auftrag.id, nachricht: "alt" },
    });

    const strom = await ereignisseGet(
      new Request(`http://localhost/api/v1/auftraege/${auftrag.kurzId}/ereignisse`, {
        headers: { cookie, "last-event-id": erste.id.toString() },
      }),
      kontext({ id: auftrag.kurzId }),
    );
    const ereignisse = await stromLesen(strom);
    const logs = ereignisse
      .filter((e) => e.art === "log")
      .flatMap((e) => (e.daten as Array<{ nachricht: string }>).map((z) => z.nachricht));
    expect(logs).toEqual(["neu"]);
  });

  it("Abbruch: Operator fordert an, der Ablauf endet vor der nächsten Phase als ABGEBROCHEN", async () => {
    const server = await serverAnlegen();
    const { benutzer, cookie } = await angemeldet("OPERATOR");
    let auftragKurz = "";
    ablaufRegistrieren("AUFRAEUMEN", async ({ senke, abbruchGewuenscht }) =>
      phasenAusfuehren(
        [
          {
            schluessel: "a",
            titel: "A",
            ausfuehren: async () => {
              await abbrechenPost(
                anfrage("POST", `/api/v1/auftraege/${auftragKurz}/abbrechen`, cookie),
                kontext({ id: auftragKurz }),
              );
              return {};
            },
          },
          { schluessel: "b", titel: "B", ausfuehren: async () => ({}) },
        ],
        senke,
        { abbruchGewuenscht },
      ),
    );
    const auftrag = await auftragAnlegen(
      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
      { benutzer, quelle: "WEB" },
      null,
      async () => {},
    );
    auftragKurz = auftrag.kurzId;
    await auftragAusfuehren(auftrag.id);
    const fertig = await testPrisma.auftrag.findUniqueOrThrow({
      where: { id: auftrag.id },
      include: { phasen: { orderBy: { reihenfolge: "asc" } } },
    });
    expect(fertig.status).toBe("ABGEBROCHEN");
    expect(fertig.phasen.map((p) => p.status)).toEqual(["ERLEDIGT", "FEHLER"]);
    expect(await testPrisma.protokoll.count({ where: { aktion: "Abbruch angefordert" } })).toBe(1);
  });

  it("ein doppelt gelieferter Job läuft nur einmal; verwaiste Aufträge werden beim Start beendet", async () => {
    const server = await serverAnlegen();
    const { benutzer } = await angemeldet("OPERATOR");
    let laeufe = 0;
    ablaufRegistrieren("AUFRAEUMEN", async () => {
      laeufe += 1;
      return { ok: true };
    });
    const auftrag = await auftragAnlegen(
      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
      { benutzer, quelle: "WEB" },
      null,
      async () => {},
    );
    await Promise.all([auftragAusfuehren(auftrag.id), auftragAusfuehren(auftrag.id)]);
    expect(laeufe).toBe(1);

    const haengt = await auftragAnlegen(
      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
      { benutzer, quelle: "WEB" },
      null,
      async () => {},
    );
    await testPrisma.auftrag.update({ where: { id: haengt.id }, data: { status: "LAEUFT" } });
    expect(await verwaisteAuftraegeBeenden()).toBe(1);
    expect((await testPrisma.auftrag.findUniqueOrThrow({ where: { id: haengt.id } })).status).toBe(
      "FEHLGESCHLAGEN",
    );
  });
});