fix(organize): preserve requests and verify identity recovery results
This commit is contained in:
root committed
1 parent
041d59405c
commit
d196235418
8 files changed
+483
-70
No files matched your search
@@ -0,0 +1,206 @@
|
||||
import { symlink, unlink, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { createTempTree, type TempTree } from "@umt/testkit";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
|
||||
import {
|
||||
retryOrganizeRecord,
|
||||
storedImportRequest,
|
||||
} from "../../src/modules/acquisition-organize/services/organize-retry.ts";
|
||||
import {
|
||||
assignOrganizeIdentity,
|
||||
organizeDecisionHook,
|
||||
} from "../../src/modules/acquisition-organize/services/takeover.ts";
|
||||
import { autoInvestigate, checkRecovery } from "../../src/modules/agent/investigate.ts";
|
||||
import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
|
||||
|
||||
const hasPg = await pgAvailable();
|
||||
const actor = { type: "user", id: "identity-recovery" };
|
||||
it("请求快照保留业务参数，不序列化运行身份与路径守卫", () => {
|
||||
expect(
|
||||
storedImportRequest({
|
||||
sourcePaths: ["/cache/movie.mkv"],
|
||||
namingMode: "keep",
|
||||
action: "move",
|
||||
trigger: "manual",
|
||||
actorType: "user",
|
||||
actorId: "secret",
|
||||
taskId: "old",
|
||||
idempotencyKey: "old",
|
||||
dryRun: true,
|
||||
pathGuard: async (p) => p,
|
||||
}),
|
||||
).toEqual({ sourcePaths: ["/cache/movie.mkv"], namingMode: "keep", action: "move" });
|
||||
});
|
||||
describe.skipIf(!hasPg)("按路径身份补齐后恢复原整理", () => {
|
||||
let h: Harness;
|
||||
let lib: TempTree;
|
||||
let cache: TempTree;
|
||||
let outside: TempTree;
|
||||
beforeAll(async () => {
|
||||
lib = await createTempTree({}, { sentinel: true });
|
||||
cache = await createTempTree({});
|
||||
outside = await createTempTree({});
|
||||
h = await createHarness("identity_recovery", {}, { withAgent: true });
|
||||
await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: "http://127.0.0.1:1" });
|
||||
}, 60_000);
|
||||
afterAll(async () => {
|
||||
await h?.close();
|
||||
await lib?.cleanup();
|
||||
await cache?.cleanup();
|
||||
await outside?.cleanup();
|
||||
});
|
||||
async function unknown(name: string, extra = {}) {
|
||||
const src = join(cache.root, `${name}.2022.1080p.mkv`);
|
||||
await writeFile(src, Buffer.alloc(300_000, 1));
|
||||
const plan = await planImport(h.d, {
|
||||
sourcePaths: [src],
|
||||
trigger: "manual",
|
||||
actorType: "user",
|
||||
actorId: actor.id,
|
||||
...extra,
|
||||
});
|
||||
expect(plan).toMatchObject({ status: "failed", reason: "identity_unknown" });
|
||||
const dec = await h.db
|
||||
.selectFrom("decision")
|
||||
.selectAll()
|
||||
.where("subject_id", "=", plan.recordId!)
|
||||
.executeTakeFirstOrThrow();
|
||||
return { src, plan, dec };
|
||||
}
|
||||
it("路径身份事项有文件证据和Agent入口，指定作品保留目录、命名与刮削参数", async () => {
|
||||
const targetDir = join(lib.root, "custom");
|
||||
const { src, plan, dec } = await unknown("Recovery.Path", {
|
||||
action: "copy",
|
||||
targetDir,
|
||||
namingMode: "keep",
|
||||
scrapeAfterImport: false,
|
||||
minSizeBytes: 100,
|
||||
});
|
||||
expect(dec.evidence).toMatchObject({
|
||||
sourcePaths: [src],
|
||||
mainFile: src,
|
||||
parsed: { title: "Recovery Path", year: 2022 },
|
||||
});
|
||||
expect((await h.api("GET", `/organization-records/${plan.recordId}`)).body.data.attention.agentIdentify).toBe(true);
|
||||
const entityId = await seedEntity(h.db, { kind: "movie", title: "Recovery Path", year: 2022 });
|
||||
const assigned = await assignOrganizeIdentity(h.d, dec.id, { mediaEntityId: entityId }, actor, "标题和年份一致");
|
||||
expect(assigned.status).toBe("running");
|
||||
const record = await h.db
|
||||
.selectFrom("organization_record")
|
||||
.selectAll()
|
||||
.where("id", "=", assigned.recordId!)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(record.plan).toMatchObject({
|
||||
retryOfRecordId: plan.recordId,
|
||||
request: { sourcePaths: [src], targetDir, namingMode: "keep", scrapeAfterImport: false, minSizeBytes: 100 },
|
||||
naming: { mode: "keep" },
|
||||
});
|
||||
const action = [{ capabilityId: "organize.assign_identity", input: { decisionId: dec.id }, status: "succeeded" }];
|
||||
expect((await checkRecovery(h.ctx, action)).state).toBe("pending");
|
||||
expect((await executeRecord(h.d, assigned.recordId!)).status).toBe("succeeded");
|
||||
expect((await checkRecovery(h.ctx, action)).state).toBe("recovered");
|
||||
const attention = await h.api("GET", "/organization-records?attention=true&organizeOnly=true");
|
||||
expect(attention.body.data.items.map((r: { id: string }) => r.id)).not.toContain(plan.recordId);
|
||||
});
|
||||
it("人工指定恢复原路径请求，不要求关联下载任务", async () => {
|
||||
const { plan, dec } = await unknown("Human.Path", { namingMode: "keep" });
|
||||
const entityId = await seedEntity(h.db, { kind: "movie", title: "Human Path", year: 2022 });
|
||||
const result = await organizeDecisionHook(h.d).execute(
|
||||
dec,
|
||||
{ optionKey: "select_entity", identity: { mediaEntityId: entityId } },
|
||||
actor,
|
||||
);
|
||||
expect(result).toMatchObject({ status: "running", mediaEntityId: entityId });
|
||||
expect(
|
||||
(
|
||||
await h.db
|
||||
.selectFrom("organization_record")
|
||||
.select("plan")
|
||||
.where("id", "=", result.recordId as string)
|
||||
.executeTakeFirstOrThrow()
|
||||
).plan,
|
||||
).toMatchObject({ retryOfRecordId: plan.recordId });
|
||||
});
|
||||
it("身份已找到但重新规划失败，保持原事项待处理", async () => {
|
||||
const { dec } = await unknown("Blocked.Path", { targetLibraryRootId: "missing-root" });
|
||||
const entityId = await seedEntity(h.db, { kind: "movie", title: "Blocked Path", year: 2022 });
|
||||
await expect(assignOrganizeIdentity(h.d, dec.id, { mediaEntityId: entityId }, actor, "标题一致")).rejects.toThrow(
|
||||
/重新整理没能执行/,
|
||||
);
|
||||
expect(
|
||||
(await h.db.selectFrom("decision").select("status").where("id", "=", dec.id).executeTakeFirstOrThrow()).status,
|
||||
).toBe("open");
|
||||
expect(
|
||||
(
|
||||
await checkRecovery(h.ctx, [
|
||||
{ capabilityId: "organize.assign_identity", input: { decisionId: dec.id }, status: "succeeded" },
|
||||
])
|
||||
).state,
|
||||
).toBe("failed");
|
||||
});
|
||||
it("恢复重新检查路径边界，拒绝后来被换成越界符号链接的源文件", async () => {
|
||||
const { src, dec } = await unknown("Guarded.Path");
|
||||
const entityId = await seedEntity(h.db, { kind: "movie", title: "Guarded Path", year: 2022 });
|
||||
const escaped = join(outside.root, "escaped.mkv");
|
||||
await writeFile(escaped, Buffer.alloc(300_000));
|
||||
await unlink(src);
|
||||
await symlink(escaped, src);
|
||||
await expect(assignOrganizeIdentity(h.d, dec.id, { mediaEntityId: entityId }, actor, "标题一致")).rejects.toThrow(
|
||||
/不在已登记/,
|
||||
);
|
||||
expect(
|
||||
(await h.db.selectFrom("decision").select("status").where("id", "=", dec.id).executeTakeFirstOrThrow()).status,
|
||||
).toBe("open");
|
||||
});
|
||||
it("Move 恢复完成后稳定幂等键重放沿用结果，不再读取已移动的源文件", async () => {
|
||||
const { plan } = await unknown("Replay.Path", { action: "move", namingMode: "keep" });
|
||||
const entityId = await seedEntity(h.db, { kind: "movie", title: "Replay Path", year: 2022 });
|
||||
const options = { mediaEntityId: entityId, idempotencyKey: "recovery:replay-path" };
|
||||
const first = await retryOrganizeRecord(h.d, plan.recordId!, actor, "test-replay", options);
|
||||
expect((await executeRecord(h.d, first.recordId!)).status).toBe("succeeded");
|
||||
const replay = await retryOrganizeRecord(h.d, plan.recordId!, actor, "test-replay", options);
|
||||
expect(replay).toMatchObject({ recordId: first.recordId, taskId: first.taskId, status: "succeeded" });
|
||||
});
|
||||
it("旧记录没有请求快照时拒绝猜测一次性参数", async () => {
|
||||
const { plan } = await unknown("Legacy.Path");
|
||||
await h.db.updateTable("organization_record").set({ plan: "{}" }).where("id", "=", plan.recordId!).execute();
|
||||
await expect(retryOrganizeRecord(h.d, plan.recordId!, actor, "test")).rejects.toThrow(/没有可恢复/);
|
||||
});
|
||||
it("整理文件季集事项默认调查，无写授权降级建议，资源站普通事项不开启全局恢复", async () => {
|
||||
await h.setSettings("agent", { autoInvestigate: false, autoIdentifyOrganize: true });
|
||||
const file = await h.db
|
||||
.insertInto("decision")
|
||||
.values({
|
||||
kind: "episode_mapping",
|
||||
title: "整理文件季集待确认",
|
||||
dedupe_key: "episode:test-files",
|
||||
evidence: JSON.stringify({ fileSamples: [{ recordId: "sample" }] }),
|
||||
})
|
||||
.returning("id")
|
||||
.executeTakeFirstOrThrow();
|
||||
const out = await autoInvestigate(h.ctx, file.id);
|
||||
expect(out).toMatchObject({ created: true, mode: "investigate" });
|
||||
const run = await h.db.selectFrom("agent_run").selectAll().where("id", "=", out.runId!).executeTakeFirstOrThrow();
|
||||
expect(run.mode).toBe("plan_only");
|
||||
expect((await autoInvestigate(h.ctx, file.id)).created).toBe(false);
|
||||
const ordinary = await h.db
|
||||
.insertInto("decision")
|
||||
.values({ kind: "episode_mapping", title: "资源站季集待确认", dedupe_key: "episode:test-ordinary" })
|
||||
.returning("id")
|
||||
.executeTakeFirstOrThrow();
|
||||
expect((await autoInvestigate(h.ctx, ordinary.id)).created).toBe(false);
|
||||
await h.setSettings("agent", { autoIdentifyOrganize: false });
|
||||
const disabled = await h.db
|
||||
.insertInto("decision")
|
||||
.values({
|
||||
kind: "episode_mapping",
|
||||
title: "整理关闭",
|
||||
dedupe_key: "episode:test-disabled",
|
||||
evidence: JSON.stringify({ fileSamples: [{}] }),
|
||||
})
|
||||
.returning("id")
|
||||
.executeTakeFirstOrThrow();
|
||||
expect((await autoInvestigate(h.ctx, disabled.id)).created).toBe(false);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user