389 lines
16 KiB
Go
389 lines
16 KiB
Go
package main
|
||
|
||
import (
|
||
"database/sql"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"testing"
|
||
|
||
"agent-system-backend/internal/store"
|
||
"agent-system-backend/pkg/extension"
|
||
)
|
||
|
||
func TestMigrateTrustedContextManifestWithoutRawScope(t *testing.T) {
|
||
target, cleanup := pgSchemaDSN(t)
|
||
defer cleanup()
|
||
fixture := buildFixture(t)
|
||
src, err := store.OpenSQLite(fixture)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "model-attempt", InputSHA256: "request-hash", Delivery: "attempted",
|
||
Included: []extension.ProjectContextManifestEntry{{SourceID: "source-a", Revision: "r1", SourceHash: "ha", InputSHA256: "input-hash"}}}
|
||
if err := src.AppendContextManifestWithScope("alice", "r1", m, store.ContextManifestScope{Bound: true}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
m.Delivery = "responded"
|
||
if err := src.AppendContextManifestWithScope("alice", "r1", m, store.ContextManifestScope{Bound: true}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
task, err := src.CreateWorkTask("alice", "rdm", "p1", "c1", "document", "Prepare document")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
stepHash := strings.Repeat("a", 64)
|
||
if _, err := src.DB().Exec(`INSERT INTO workflow_step_sources(run_id,step_id,owner,project_id,step_attempt_id,output_sha256,attempt_id,authorization_epoch,publication_epoch,watermark,version,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||
"r1", "draft", "alice", "p1", "step-attempt", stepHash, "model-attempt", "auth-1", "pub-1", "water-1", "v1", "2026-10-14T00:00:00Z"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := src.Close(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
snapshot := filepath.Join(t.TempDir(), "snapshot.db")
|
||
if err := cmdBackup([]string{"--source", fixture, "--out", snapshot}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := cmdMigrate([]string{"--source", snapshot, "--target", target, "--migration-id", "context-copy"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
pg, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer pg.Close()
|
||
receipts, err := pg.ListContextManifests("alice", "r1")
|
||
if err != nil || len(receipts) != 2 || len(receipts[1].Included) != 1 || receipts[1].Included[0].SourceID != "source-a" {
|
||
t.Fatalf("content-free model receipts not copied: count=%d err=%v", len(receipts), err)
|
||
}
|
||
workScope, err := pg.WorkTaskScope("alice", "rdm", task.ID)
|
||
if err != nil || workScope.ProjectID != "p1" || workScope.ConversationID != "c1" {
|
||
t.Fatalf("trusted work task binding not copied: scope=%+v err=%v", workScope, err)
|
||
}
|
||
stepSource, err := pg.WorkflowStepSource("alice", "r1", "draft")
|
||
if err != nil || stepSource.StepAttemptID != "step-attempt" || stepSource.AttemptID != "model-attempt" || stepSource.ProjectID != "p1" {
|
||
t.Fatalf("workflow step source not copied: source=%+v err=%v", stepSource, err)
|
||
}
|
||
var rawColumns int
|
||
if err := pg.DB().QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='context_manifests' AND column_name IN ('resolve_goal','resolve_target_id')`).Scan(&rawColumns); err != nil || rawColumns != 0 {
|
||
t.Fatalf("migrated PG schema has raw prompt columns: count=%d err=%v", rawColumns, err)
|
||
}
|
||
if err := cmdMigrate([]string{"--source", snapshot, "--target", target, "--migration-id", "context-copy"}); err != nil {
|
||
t.Fatalf("idempotent context migration rerun: %v", err)
|
||
}
|
||
receipts, err = pg.ListContextManifests("alice", "r1")
|
||
if err != nil || len(receipts) != 2 {
|
||
t.Fatalf("rerun duplicated or lost model receipts: count=%d err=%v", len(receipts), err)
|
||
}
|
||
if err := pg.DB().QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='context_manifests' AND column_name IN ('resolve_goal','resolve_target_id')`).Scan(&rawColumns); err != nil || rawColumns != 0 {
|
||
t.Fatalf("rerun reintroduced raw prompt columns: count=%d err=%v", rawColumns, err)
|
||
}
|
||
}
|
||
|
||
// U4 集成（环境门控 AGENT_PG_TEST_DSN：CI test-pg job / 本机隔离实例）：
|
||
// 夹具 → 受管迁移 → 校验和验证 → 幂等重跑 → 断点续传 → 非空目标拒绝 →
|
||
// 未知 owner 阻塞。每用例独立 PG schema（与 store 测试同隔离约定）。
|
||
|
||
func pgSchemaDSN(t *testing.T) (string, func()) {
|
||
t.Helper()
|
||
dsn := os.Getenv("AGENT_PG_TEST_DSN")
|
||
if dsn == "" {
|
||
t.Skip("AGENT_PG_TEST_DSN 未设置：真库集成由 CI 的必跑 job 覆盖")
|
||
}
|
||
schema := "mig_" + strings.ToLower(store.NewID("t")) + "_" + fmt.Sprint(os.Getpid())
|
||
admin, err := sql.Open("pgx", dsn)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
|
||
admin.Close()
|
||
t.Fatalf("preinstall pg_trgm: %v", err)
|
||
}
|
||
if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
|
||
admin.Close()
|
||
t.Fatalf("create schema: %v", err)
|
||
}
|
||
cleanup := func() {
|
||
admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
|
||
admin.Close()
|
||
}
|
||
return store.AppendPGRuntimeParam(dsn, "search_path", schema+",public"), cleanup
|
||
}
|
||
|
||
func TestMigrateEndToEndAndIdempotent(t *testing.T) {
|
||
target, cleanup := pgSchemaDSN(t)
|
||
defer cleanup()
|
||
fixture := buildFixture(t)
|
||
src, err := sql.Open("sqlite", fixture)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer src.Close()
|
||
var sourcePublic, sourceStaff, sourcePrivate, sourceSecret string
|
||
var sourceKeyVersion int64
|
||
if err := src.QueryRow(`SELECT public_revision,staff_no_revision,private_revision,cursor_key_version,cursor_secret FROM account_directory_revision WHERE id=1`).Scan(
|
||
&sourcePublic, &sourceStaff, &sourcePrivate, &sourceKeyVersion, &sourceSecret,
|
||
); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-e2e"}); err != nil {
|
||
t.Fatalf("首次迁移：%v", err)
|
||
}
|
||
// 幂等重跑（同身份）：表级 done 对账，零重复行。
|
||
if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-e2e"}); err != nil {
|
||
t.Fatalf("幂等重跑：%v", err)
|
||
}
|
||
// 目标侧关键断言：布尔映射、NULL 保留、Unicode、BLOB、SERIAL 序列推进、
|
||
// 复合主键事件序。
|
||
st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer st.Close()
|
||
db := st.DB()
|
||
var targetPublic, targetStaff, targetPrivate, targetSecret string
|
||
var targetKeyVersion int64
|
||
if err := db.QueryRow(`SELECT public_revision,staff_no_revision,private_revision,cursor_key_version,cursor_secret FROM account_directory_revision WHERE id=1`).Scan(
|
||
&targetPublic, &targetStaff, &targetPrivate, &targetKeyVersion, &targetSecret,
|
||
); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if targetPublic != sourcePublic || targetStaff != sourceStaff || targetPrivate != sourcePrivate || targetKeyVersion != sourceKeyVersion || targetSecret != sourceSecret {
|
||
t.Fatal("迁移后的目录版本或游标密钥与源库不一致")
|
||
}
|
||
var pinned, archived bool
|
||
if err := db.QueryRow(`SELECT pinned, archived FROM conversations WHERE id='c1'`).Scan(&pinned, &archived); err != nil || !pinned || archived {
|
||
t.Fatalf("布尔映射：pinned=%v archived=%v err=%v", pinned, archived, err)
|
||
}
|
||
var title, note string
|
||
if err := db.QueryRow(`SELECT title, session_note FROM conversations WHERE id='c1'`).Scan(&title, ¬e); err != nil || !strings.Contains(note, "注") {
|
||
t.Fatalf("Unicode：title=%q note=%q err=%v", title, note, err)
|
||
}
|
||
var vec []byte
|
||
if err := db.QueryRow(`SELECT vec FROM knowledge_vectors WHERE doc_id='k1'`).Scan(&vec); err != nil || len(vec) != 12 {
|
||
t.Fatalf("BLOB 向量：len=%d err=%v", len(vec), err)
|
||
}
|
||
var ended sql.NullString
|
||
if err := db.QueryRow(`SELECT ended_at FROM runs WHERE id='r1'`).Scan(&ended); err != nil || ended.Valid {
|
||
t.Fatalf("NULL 保留：ended=%v err=%v", ended, err)
|
||
}
|
||
var seqMax int
|
||
if err := db.QueryRow(`SELECT MAX(seq) FROM events WHERE conversation_id='c1'`).Scan(&seqMax); err != nil || seqMax != 2 {
|
||
t.Fatalf("事件复合主键序：max=%d err=%v", seqMax, err)
|
||
}
|
||
// 复制的显式 id 之后，应用侧插入不撞序列（command_audit.id=7 已复制）。
|
||
if _, err := db.Exec(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner) VALUES ('2026-10-11T03:00:00Z','cmd-2','http','chat','done','op-2','c1','{}','alice')`); err != nil {
|
||
t.Fatalf("序列推进失败（复制后插入撞主键）：%v", err)
|
||
}
|
||
var linkedResource string
|
||
if err := db.QueryRow(`SELECT resource_id FROM project_audit_links WHERE host_audit_offset=7 AND source_id='commit-1'`).Scan(&linkedResource); err != nil || linkedResource != "doc-1" {
|
||
t.Fatalf("审计链接迁移丢失: resource=%q err=%v", linkedResource, err)
|
||
}
|
||
var newLinkID int64
|
||
if err := db.QueryRow(`INSERT INTO project_audit_links(host_audit_offset,project_id,owner,source_kind,source_id,operation_id,resource_kind,resource_id,run_id,disclosure_class,idempotency_key,created_at)
|
||
VALUES (7,'p1','alice','domain-commit','commit-2','op-1','document','doc-2','','metadata','outbox-2','2026-10-11T03:00:01Z') RETURNING id`).Scan(&newLinkID); err != nil || newLinkID <= 3 {
|
||
t.Fatalf("审计链接序列未推进: id=%d err=%v", newLinkID, err)
|
||
}
|
||
// 报告不含秘密值。
|
||
var report string
|
||
if err := db.QueryRow(`SELECT report FROM migration_control WHERE migration_id='mig-e2e'`).Scan(&report); err != nil || strings.Contains(report, "sk-secret-value") {
|
||
t.Fatalf("报告泄密：err=%v", err)
|
||
}
|
||
}
|
||
|
||
func TestCommandAuditExclusiveCopyRollsBackAndNeverRewindsSequence(t *testing.T) {
|
||
target, cleanup := pgSchemaDSN(t)
|
||
defer cleanup()
|
||
fixture := buildFixture(t)
|
||
src, err := openSourceRO(fixture)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer src.Close()
|
||
st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer st.Close()
|
||
db := st.DB()
|
||
if _, err := db.Exec(controlDDL); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cols, pk, err := targetColumns(db, "command_audit")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
bools, err := boolColumns(db, "command_audit", cols)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
count, checksum, err := sourceScan(src, "command_audit", cols, pk, bools)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
blocker, err := db.Begin()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var existing int
|
||
if err := blocker.QueryRow(`SELECT COUNT(*) FROM command_audit`).Scan(&existing); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
timeoutDB, err := sql.Open("pgx", store.AppendPGRuntimeParam(target, "lock_timeout", "100"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := migrateCommandAudit(timeoutDB, src, "retry-audit", cols, pk, bools, count, checksum); err == nil || !strings.Contains(err.Error(), "lock timeout") {
|
||
t.Fatalf("exclusive copy did not fail on a held reader: %v", err)
|
||
}
|
||
timeoutDB.Close()
|
||
if err := blocker.Rollback(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := migrateCommandAudit(db, src, "retry-audit", cols, pk, bools, count+1, checksum); err == nil {
|
||
t.Fatal("mismatched copy committed despite in-transaction target count check")
|
||
}
|
||
var rows, done int
|
||
if err := db.QueryRow(`SELECT COUNT(*) FROM command_audit`).Scan(&rows); err != nil || rows != 0 {
|
||
t.Fatalf("failed copy left rows: %d %v", rows, err)
|
||
}
|
||
if err := db.QueryRow(`SELECT COUNT(*) FROM migration_table_state WHERE table_name='command_audit'`).Scan(&done); err != nil || done != 0 {
|
||
t.Fatalf("failed copy recorded done: %d %v", done, err)
|
||
}
|
||
var enabled string
|
||
if err := db.QueryRow(`SELECT tgenabled FROM pg_trigger WHERE tgrelid='command_audit'::regclass AND tgname='project_audit_insert_guard_v1'`).Scan(&enabled); err != nil || enabled != "A" {
|
||
t.Fatalf("failed copy left guard disabled: %q %v", enabled, err)
|
||
}
|
||
var seq string
|
||
if err := db.QueryRow(`SELECT pg_get_serial_sequence('command_audit','id')`).Scan(&seq); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// setval is non-transactional; an interrupted offline attempt can leave a
|
||
// forward gap even though its rows and trigger-state changes roll back.
|
||
tx, err := db.Begin()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := tx.Exec(`LOCK TABLE command_audit IN ACCESS EXCLUSIVE MODE`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := tx.Exec(`ALTER TABLE command_audit DISABLE TRIGGER project_audit_insert_guard_v1`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := tx.Exec(`SELECT setval($1::regclass,50,true)`, seq); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := tx.Rollback(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := migrateCommandAudit(db, src, "retry-audit", cols, pk, bools, count, checksum); err != nil {
|
||
t.Fatalf("retry after rollback: %v", err)
|
||
}
|
||
var last int64
|
||
if err := db.QueryRow(`SELECT last_value FROM ` + seq).Scan(&last); err != nil || last < 50 {
|
||
t.Fatalf("copy rewound sequence: last=%d err=%v", last, err)
|
||
}
|
||
var nextID int64
|
||
if err := db.QueryRow(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload)
|
||
VALUES ('2026-10-14T00:00:00Z','after-copy','','','committed','','','alice','{}') RETURNING id`).Scan(&nextID); err != nil || nextID <= 50 {
|
||
t.Fatalf("post-copy ID did not advance: id=%d err=%v", nextID, err)
|
||
}
|
||
}
|
||
|
||
func TestMigrateResumeAfterInterrupt(t *testing.T) {
|
||
target, cleanup := pgSchemaDSN(t)
|
||
defer cleanup()
|
||
fixture := buildFixture(t)
|
||
|
||
if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-resume"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// 模拟断点：messages 已 done 后被部分回退（行删一半、状态回 pending），
|
||
// 重跑只重做该表，终态校验仍全绿。
|
||
db, err := sql.Open("pgx", target)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer db.Close()
|
||
if _, err := db.Exec(`DELETE FROM messages WHERE id='m3'`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := db.Exec(`UPDATE migration_table_state SET state='pending' WHERE migration_id='mig-resume' AND table_name='messages'`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-resume"}); err != nil {
|
||
t.Fatalf("断点续跑：%v", err)
|
||
}
|
||
var n int
|
||
if err := db.QueryRow(`SELECT COUNT(*) FROM messages`).Scan(&n); err != nil || n != 3 {
|
||
t.Fatalf("续跑后 messages=%d：重做该表应恢复 3 行", n)
|
||
}
|
||
}
|
||
|
||
func TestMigrateRefusesForeignDataAndWrongSnapshot(t *testing.T) {
|
||
target, cleanup := pgSchemaDSN(t)
|
||
defer cleanup()
|
||
fixture := buildFixture(t)
|
||
|
||
// 非空目标（人工数据、无迁移身份）：拒绝。
|
||
st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
st.Close()
|
||
db, err := sql.Open("pgx", target)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer db.Close()
|
||
if _, err := db.Exec(`INSERT INTO projects (id, name, settings, created_at, updated_at, owner) VALUES ('foreign', '外来', '{}', '2026-10-11T00:00:00Z', '2026-10-11T00:00:00Z', 'alice')`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
err = cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-x"})
|
||
if err == nil || !strings.Contains(err.Error(), "非空目标") {
|
||
t.Fatalf("非空目标应拒绝：%v", err)
|
||
}
|
||
if _, err := db.Exec(`DELETE FROM projects WHERE id='foreign'`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
// 同 migration_id 换快照：拒绝（仅同一冻结快照可续传）。
|
||
if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-snap"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
other := buildFixture(t) // 不同文件 ⇒ 不同 snapshot_id
|
||
err = cmdMigrate([]string{"--source", other, "--target", target, "--migration-id", "mig-snap"})
|
||
if err == nil || !strings.Contains(err.Error(), "快照不同") {
|
||
t.Fatalf("换快照应拒绝：%v", err)
|
||
}
|
||
|
||
// 其他 migration_id 已存在：拒绝。
|
||
err = cmdMigrate([]string{"--source", other, "--target", target, "--migration-id", "mig-other"})
|
||
if err == nil || !strings.Contains(err.Error(), "其他迁移记录") {
|
||
t.Fatalf("他身份应拒绝：%v", err)
|
||
}
|
||
}
|
||
|
||
func TestMigrateBlocksUnknownOwner(t *testing.T) {
|
||
target, cleanup := pgSchemaDSN(t)
|
||
defer cleanup()
|
||
fixture := buildFixture(t)
|
||
// 在夹具里造一个 users 之外的 owner 行（直接写源库）。
|
||
srcPath := fixture
|
||
st, err := store.OpenSQLite(srcPath) // 普通打开（夹具可写）
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := st.DB().Exec(`INSERT INTO knowledge (id, project_id, source, title, ref, chunk, created_at, owner) VALUES
|
||
('k_ghost', 'p1', 'upload', '幽灵', 'g.md', '…', '2026-10-11T01:00:00Z', 'ghost-user')`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
st.Close()
|
||
err = cmdMigrate([]string{"--source", srcPath, "--target", target, "--migration-id", "mig-ghost"})
|
||
if err == nil || !strings.Contains(err.Error(), "未知 owner") {
|
||
t.Fatalf("未知 owner 应阻塞：%v", err)
|
||
}
|
||
}
|