diff --git a/internal/docgov/engine_db_test.go b/internal/docgov/engine_db_test.go index 1fbf869f..9a5671c9 100644 --- a/internal/docgov/engine_db_test.go +++ b/internal/docgov/engine_db_test.go @@ -6,26 +6,67 @@ import ( "testing" "time" - "chat-agent-system/internal/config" + "chat-agent-system/internal/db" "chat-agent-system/internal/docgov" "github.com/jackc/pgx/v5/pgxpool" ) -func TestDocGov_DatabaseIntegration(t *testing.T) { - cfg, err := config.Load(".env") - if err != nil || cfg == nil || cfg.ContextDbHost == "" { - t.Skip("Skipping DB integration test: context DB not configured") - } - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) +// openIsolatedPool 仅连接显式提供的隔离测试库(TEST_DATABASE_URL), +// 禁止回退到共享环境库(agent_system_dev/test/main)。 +func openIsolatedPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := db.SkipIfNoTestDatabaseURL(t) + db.RefuseSharedEnvDSN(t, dsn) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() - - pool, err := pgxpool.New(ctx, cfg.ContextDSN()) - if err != nil || pool.Ping(ctx) != nil { - t.Skipf("Skipping DB integration test: cannot connect to %s: %v", cfg.ContextDSN(), err) + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Skipf("cannot connect to isolated test db: %v", err) } + if err := pool.Ping(ctx); err != nil { + pool.Close() + t.Skipf("isolated test db ping failed: %v", err) + } + if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil { + pool.Close() + t.Skipf("migration failed: %v", err) + } + return pool +} + +// seedTestProject 创建本测试专属项目行,满足 document_records.project_id 外键。 +func seedTestProject(t *testing.T, pool *pgxpool.Pool) int64 { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + stamp := time.Now().Format("20060102150405.000000") + var projectID int64 + err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status) + VALUES ('docgov-' || $1, 'DocGov Engine Test ' || $1, 'active', 2, 'active') + RETURNING id + `, stamp).Scan(&projectID) + if err != nil { + t.Fatalf("seed project: %v", err) + } + t.Cleanup(func() { + cleanupCtx := context.Background() + _, _ = pool.Exec(cleanupCtx, `DELETE FROM document_comments WHERE doc_id IN (SELECT doc_id FROM document_records WHERE project_id = $1)`, projectID) + _, _ = pool.Exec(cleanupCtx, `DELETE FROM document_change_proposals WHERE doc_id IN (SELECT doc_id FROM document_records WHERE project_id = $1)`, projectID) + _, _ = pool.Exec(cleanupCtx, `DELETE FROM document_records WHERE project_id = $1`, projectID) + _, _ = pool.Exec(cleanupCtx, `DELETE FROM projects WHERE id = $1`, projectID) + }) + return projectID +} + +func TestDocGov_DatabaseIntegration(t *testing.T) { + pool := openIsolatedPool(t) defer pool.Close() + projectID := seedTestProject(t, pool) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() engine := docgov.NewGovernanceEngine(pool) @@ -35,25 +76,25 @@ func TestDocGov_DatabaseIntegration(t *testing.T) { doc, err := engine.CreateOrGetDocument( ctx, testDocID, - 1, + projectID, "feasibility_reg", "DB集成测试文档", "V0.5_feasibility", "data/dev/workspaces/test_db", - 1, "张工", - 2, "李审核", - 3, "王总监", + 2, "张工", + 3, "李主管", + 1, "Seed Admin", ) if err != nil { t.Fatalf("CreateOrGetDocument in DB failed: %v", err) } if doc.DocID != testDocID { - t.Fatalf("unexpected docID: %s", doc.DocID) + t.Fatalf("unexpected docID: %s", testDocID) } // 2. 标记正文非模板并认领 _ = engine.MarkDocumentContentUpdated(ctx, testDocID, "db_test_sha256", false) - doc, err = engine.RecordDocumentViewed(ctx, testDocID, 1, docgov.RoleAuthor, "张工") + doc, err = engine.RecordDocumentViewed(ctx, testDocID, 2, docgov.RoleAuthor, "张工") if err != nil { t.Fatalf("RecordDocumentViewed failed: %v", err) } @@ -75,7 +116,7 @@ func TestDocGov_DatabaseIntegration(t *testing.T) { } // 4. 拍板同意提案 - prop, err = engine.DecideModificationProposal(ctx, prop.ProposalID, 1, true, "DB测试同意") + prop, err = engine.DecideModificationProposal(ctx, prop.ProposalID, 2, true, "DB测试同意") if err != nil { t.Fatalf("DecideModificationProposal in DB failed: %v", err) } @@ -84,14 +125,14 @@ func TestDocGov_DatabaseIntegration(t *testing.T) { } // 5. 批注与送审流程 - _, err = engine.AddComment(ctx, testDocID, 2, "李审核", docgov.RoleReviewer, "## 结论", docgov.CommentSuggestion, "请增加法规出处") + _, err = engine.AddComment(ctx, testDocID, 3, "李主管", docgov.RoleReviewer, "## 结论", docgov.CommentSuggestion, "请增加法规出处") if err != nil { t.Fatalf("AddComment in DB failed: %v", err) } - doc, err = engine.SubmitForReview(ctx, testDocID, 1, docgov.RoleAuthor, "张工") + doc, err = engine.SubmitForReview(ctx, testDocID, 2, docgov.RoleAuthor, "张工") if err != nil { - t.Fatalf("SubmitForReview in DB failed: %v", err) + t.Fatalf("SubmitForReview failed: %v", err) } if !doc.IsLocked { t.Fatalf("expected doc locked in reviewing") @@ -101,23 +142,15 @@ func TestDocGov_DatabaseIntegration(t *testing.T) { // TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK ensures reviewer/approver=0 // persist as SQL NULL (optional DutyMode) instead of FK-violating 0. func TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK(t *testing.T) { - cfg, err := config.Load(".env") - if err != nil || cfg == nil || cfg.ContextDbHost == "" { - t.Skip("Skipping DB integration test: context DB not configured") - } + pool := openIsolatedPool(t) + defer pool.Close() + projectID := seedTestProject(t, pool) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - pool, err := pgxpool.New(ctx, cfg.ContextDSN()) - if err != nil || pool.Ping(ctx) != nil { - t.Skipf("Skipping DB integration test: cannot connect to %s: %v", cfg.ContextDSN(), err) - } - defer pool.Close() - engine := docgov.NewGovernanceEngine(pool) testDocID := fmt.Sprintf("doc_opt_duty_%d", time.Now().UnixNano()) - projectID := time.Now().UnixNano()%900000 + 100000 doc, err := engine.CreateOrGetDocument( ctx, @@ -127,7 +160,7 @@ func TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK(t *testing.T) { "可选职责空审核人", "V0.5_feasibility", "", - 9, "系统管理员", + 2, "张工", 0, "", 0, "", ) @@ -135,7 +168,7 @@ func TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK(t *testing.T) { t.Fatalf("CreateOrGetDocument with reviewer/approver=0 failed: %v", err) } if doc.DocID != testDocID { - t.Fatalf("unexpected docID: %s", doc.DocID) + t.Fatalf("unexpected docID: %s", testDocID) } var reviewerNull, approverNull bool @@ -149,12 +182,10 @@ func TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK(t *testing.T) { if err != nil { t.Fatalf("query document_records: %v", err) } - if authorID != 9 { - t.Fatalf("author_id=%d want 9", authorID) + if authorID != 2 { + t.Fatalf("author_id=%d want 2", authorID) } if !reviewerNull || !approverNull { t.Fatalf("expected NULL reviewer/approver, got reviewerNull=%v approverNull=%v", reviewerNull, approverNull) } - - _, _ = pool.Exec(ctx, `DELETE FROM document_records WHERE doc_id = $1`, testDocID) } diff --git a/internal/notification/reminder_test.go b/internal/notification/reminder_test.go index 6fc75659..95f299f3 100644 --- a/internal/notification/reminder_test.go +++ b/internal/notification/reminder_test.go @@ -21,11 +21,11 @@ func TestPG_StageAdvanceReminder24h(t *testing.T) { ctx := context.Background() var projectID, instanceID, stageInstID int64 - code := "remind-" + time.Now().Format("150405.000") + code := "remind-" + time.Now().Format("150405.000000") if err := pool.QueryRow(ctx, ` INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status) - VALUES ($1, 'Reminder Test', 'active', 2, 'active') RETURNING id - `, code).Scan(&projectID); err != nil { + VALUES ($1, 'Reminder Test ' || $2, 'active', 2, 'active') RETURNING id + `, code, code).Scan(&projectID); err != nil { t.Fatalf("insert project: %v", err) } t.Cleanup(func() { diff --git a/internal/notification/service_test.go b/internal/notification/service_test.go index 44cbfe4f..1cdd6fcd 100644 --- a/internal/notification/service_test.go +++ b/internal/notification/service_test.go @@ -3,6 +3,7 @@ package notification_test import ( "context" "os" + "strings" "testing" "time" @@ -14,9 +15,14 @@ import ( func openPool(t *testing.T) *pgxpool.Pool { t.Helper() + // 隔离约定:仅提供专用测试库时运行,禁止默认回退到开发库。 + if os.Getenv("TEST_DATABASE_URL") == "" { + t.Skip("TEST_DATABASE_URL not set; skipping notification PG test") + return nil + } dsn := os.Getenv("TEST_DATABASE_URL") - if dsn == "" { - dsn = "postgres://app:dev_secret_db_pass@127.0.0.1:5433/agent_system_dev?sslmode=disable" + if strings.Contains(dsn, "agent_system_dev") || strings.Contains(dsn, "agent_system_test") || strings.Contains(dsn, "agent_system_main") { + t.Fatal("TEST_DATABASE_URL must not point at a shared env database") } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -48,11 +54,11 @@ func TestPG_EmitStageReadyDedupe(t *testing.T) { ctx := context.Background() var projectID, instanceID, stageInstID int64 - code := "notify-" + time.Now().Format("150405.000") + code := "notify-" + time.Now().Format("150405.000000") err := pool.QueryRow(ctx, ` INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status) - VALUES ($1, 'Notify Test', 'active', 2, 'active') RETURNING id - `, code).Scan(&projectID) + VALUES ($1, 'Notify Test ' || $2, 'active', 2, 'active') RETURNING id + `, code, code).Scan(&projectID) if err != nil { t.Fatalf("insert project: %v", err) } diff --git a/internal/projectinstance/pg_integration_test.go b/internal/projectinstance/pg_integration_test.go index 58ef79d6..f6e483a2 100644 --- a/internal/projectinstance/pg_integration_test.go +++ b/internal/projectinstance/pg_integration_test.go @@ -9,7 +9,6 @@ import ( "chat-agent-system/internal/db" "chat-agent-system/internal/docgov" - "chat-agent-system/internal/notification" "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/workflowdef" @@ -50,38 +49,31 @@ func TestPG_ActivationAndStageAdvance(t *testing.T) { defer pool.Close() ctx := context.Background() - wfSvc := workflowdef.NewService(pool, nil) - wfKey := "pg_p3_wf_" + time.Now().Format("150405") - _, ver, err := wfSvc.CreateDefinition(ctx, workflowdef.CreateDefinitionInput{ - Key: wfKey, Name: "PG P3 Test WF", - Definition: workflowdef.VersionDefinition{ - SchemaVersion: 1, - Stages: []workflowdef.StageDef{{ - Key: "biz_intake", Name: "商务摸底", Position: 0, - DocTasks: []workflowdef.DocTaskDef{{ - Key: "intake", Required: true, Blocking: true, DefaultEnabled: true, - OutputDocType: "intake", OutputDocTitle: "客户需求采集", - }}, - }}, - }, - }, 1) + // 治理链路仅接受官方工作流(BuildDefaultPlan official-only),seed 官方 main manifest。 + def, err := workflowdef.LoadManifestDefinition(workflowdef.V2ManifestFiles[workflowdef.OfficialKeyMain]) if err != nil { - t.Fatalf("create workflow def: %v", err) - } - if _, err := wfSvc.PublishVersion(ctx, ver.WorkflowID, ver.Version, 1); err != nil { - t.Fatalf("publish workflow: %v", err) + t.Fatalf("load manifest: %v", err) } wfStore := workflowdef.NewStore(pool) + if err := wfStore.UpsertPublishedSeed(ctx, workflowdef.CreateDefinitionInput{ + Key: workflowdef.OfficialKeyMain, + Name: workflowdef.OfficialWorkflowName(workflowdef.OfficialKeyMain), + Definition: def, + }, 1); err != nil { + t.Fatalf("seed official workflow: %v", err) + } + wfKey := workflowdef.OfficialKeyMain + pwStore := projectworkflow.NewStore(pool) instStore := projectinstance.NewStore(pool) docChecker := projectinstance.NewDocGovChecker(docgov.NewGovernanceEngine(pool)) svc := projectinstance.NewService(instStore, wfStore, pwStore, docChecker) var projectID int64 - code := "pg-p3-" + time.Now().Format("150405.000") + code := "pg-p3-" + time.Now().Format("150405.000000") err = pool.QueryRow(ctx, ` INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) - VALUES ($1, 'PG Phase3 Test', 'active', 2, 'planning', '', 1, 0) RETURNING id + VALUES ($1, 'PG Phase3 Test ' || $1, 'active', 2, 'planning', '', 1, 0) RETURNING id `, code).Scan(&projectID) if err != nil { t.Fatalf("insert project: %v", err) @@ -96,9 +88,9 @@ func TestPG_ActivationAndStageAdvance(t *testing.T) { } for i := range plan.Stages { for j := range plan.Stages[i].DocTasks { - if plan.Stages[i].DocTasks[j].TaskKey == "intake" || plan.Stages[i].DocTasks[j].TaskKey == "bd_intake" { + if plan.Stages[i].DocTasks[j].TaskKey == "bd_intake" { plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{ - AuthorUserID: 2, ReviewerUserID: 3, ApproverUserID: 4, + AuthorUserID: 2, ReviewerUserID: 3, ApproverUserID: 1, } } } @@ -125,19 +117,19 @@ func TestPG_ActivationAndStageAdvance(t *testing.T) { docID := "doc_pg_p3_intake_" + code engine := docgov.NewGovernanceEngine(pool) - _, err = engine.CreateOrGetDocument(ctx, docID, projectID, "intake", "客户需求采集", "V0.1", "", 2, "张", 3, "李", 4, "王") + _, err = engine.CreateOrGetDocument(ctx, docID, projectID, "bd_intake", "用户初始需求收集表", "V0.1", "", 2, "张", 3, "李", 1, "王") if err != nil { t.Fatalf("create doc: %v", err) } _ = engine.MarkDocumentContentUpdated(ctx, docID, "hash_pg_p3", false) _, _ = engine.SubmitForReview(ctx, docID, 2, "author", "张") _, _ = engine.ReviewApprove(ctx, docID, 3, "reviewer", "李", "ok") - _, err = engine.ApproverApprove(ctx, docID, 4, "approver", "王", "ok") + _, err = engine.ApproverApprove(ctx, docID, 1, "approver", "王", "ok") if err != nil { t.Fatalf("approve doc: %v", err) } - approvedResult, err := svc.HandleDocumentApproved(ctx, projectID, "intake") + approvedResult, err := svc.HandleDocumentApproved(ctx, projectID, "bd_intake") if err != nil { t.Fatalf("handle document approved: %v", err) } @@ -145,36 +137,22 @@ func TestPG_ActivationAndStageAdvance(t *testing.T) { t.Fatalf("expected stage ready after approval, got %+v", approvedResult) } - notifySvc := notification.NewService(pool) - if notifySvc.Available() { - created, emitErr := notifySvc.EmitStageReadyToAdvance(ctx, 2, notification.StageReadyPayload{ - ProjectID: projectID, - WorkflowInstanceID: approvedResult.InstanceID, - StageInstanceID: approvedResult.StageInstanceID, - StageKey: approvedResult.StageKey, - StageName: approvedResult.StageName, - ConfigRevision: approvedResult.ConfigRevision, - RequiredDocCount: approvedResult.RequiredDocCount, - ApprovedDocCount: approvedResult.ApprovedDocCount, - }) - if emitErr != nil { - t.Fatalf("emit stage ready notification: %v", emitErr) - } - if !created { - t.Fatal("expected stage ready notification to be created") - } - } - - readiness, err := svc.GetStageReadiness(ctx, activated.Instance.ID, "biz_intake", 2, true) + readiness, err := svc.GetStageReadiness(ctx, activated.Instance.ID, "V0.1_triage", 2, true) if err != nil { t.Fatalf("readiness: %v", err) } if !readiness.Ready { - t.Fatalf("expected ready stage, issues=%+v tasks=%+v", readiness.BlockingIssues, readiness.DocTasks) + otherTasks := []string{} + for _, dt := range readiness.DocTasks { + if !dt.Approved { + otherTasks = append(otherTasks, dt.TaskKey+":"+dt.DocState) + } + } + t.Fatalf("expected ready stage, pending=%v issues=%+v", otherTasks, readiness.BlockingIssues) } advanced, err := svc.AdvanceStage(ctx, activated.Instance.ID, 2, true, projectinstance.AdvanceStageInput{ - StageKey: "biz_intake", ExpectedConfigRevision: activated.Instance.ConfigRevision, + StageKey: "V0.1_triage", ExpectedConfigRevision: activated.Instance.ConfigRevision, }) if err != nil { t.Fatalf("advance: %v", err) @@ -183,3 +161,4 @@ func TestPG_ActivationAndStageAdvance(t *testing.T) { t.Fatalf("expected next stage or done, got %+v", advanced) } } + diff --git a/internal/server/chat_history_refresh_pg_test.go b/internal/server/chat_history_refresh_pg_test.go index 9cc3ce25..a7f94e16 100644 --- a/internal/server/chat_history_refresh_pg_test.go +++ b/internal/server/chat_history_refresh_pg_test.go @@ -40,7 +40,7 @@ func TestChatHistoryRefresh_PG_PreActivationAndTaskScope(t *testing.T) { code := "pg-refresh-" + time.Now().Format("150405.000") if err := pool.QueryRow(ctx, ` INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) - VALUES ($1, 'PG Refresh Test', 'active', $2, 'planning', '', 1, 0) RETURNING id + VALUES ($1, 'PG Refresh Test ' || $1, 'active', $2, 'planning', '', 1, 0) RETURNING id `, code, creatorID).Scan(&projectID); err != nil { t.Fatalf("insert project: %v", err) } @@ -70,6 +70,8 @@ func TestChatHistoryRefresh_PG_PreActivationAndTaskScope(t *testing.T) { _ = json.Unmarshal(recDefault.Body.Bytes(), &plan) reviewerID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "lizhuguan", "123456")) approverID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "wangzong", "123456")) + // 治理校验要求审核/批准负责人为在任项目成员(DUTY_ASSIGNEE_NOT_ACTIVE_MEMBER)。 + ensureProjectMembers(t, pool, projectID, map[int64]string{reviewerID: "reviewer", approverID: "approver"}) for i := range plan.Stages { for j := range plan.Stages[i].DocTasks { plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{ diff --git a/internal/server/e2e_workflow_governance_test.go b/internal/server/e2e_workflow_governance_test.go index b503fae6..9dc522a5 100644 --- a/internal/server/e2e_workflow_governance_test.go +++ b/internal/server/e2e_workflow_governance_test.go @@ -39,7 +39,7 @@ func TestE2E_WorkflowGovernance_CompleteAndReplan(t *testing.T) { code := "e2e-gov-" + time.Now().Format("150405.000") err := pool.QueryRow(ctx, ` INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) - VALUES ($1, 'E2E Governance', 'active', $2, 'planning', '', 1, 0) RETURNING id + VALUES ($1, 'E2E Governance ' || $1, 'active', $2, 'planning', '', 1, 0) RETURNING id `, code, creatorID).Scan(&projectID) if err != nil { t.Fatalf("insert project: %v", err) diff --git a/internal/server/phase5_acceptance_pg_test.go b/internal/server/phase5_acceptance_pg_test.go index 347bc1a1..ad368fb1 100644 --- a/internal/server/phase5_acceptance_pg_test.go +++ b/internal/server/phase5_acceptance_pg_test.go @@ -109,46 +109,35 @@ func TestPhase5_ActivateIdempotencyConflict_PG(t *testing.T) { ctx := context.Background() handler, _ := setupPGGovernanceServer(t, pool) + seedOfficialWorkflowManifest(t, pool, workflowdef.OfficialKeyMain) + creatorToken := loginTestUser(t, handler, "zhanggong", "123456") - - wfSvc := workflowdef.NewService(pool, nil) - wfKey := "pg_p5_idem_" + time.Now().Format("150405.000") - _, ver, err := wfSvc.CreateDefinition(ctx, workflowdef.CreateDefinitionInput{ - Key: wfKey, Name: "Idempotency WF", - Definition: workflowdef.VersionDefinition{ - SchemaVersion: 1, - Stages: []workflowdef.StageDef{{ - Key: "biz_intake", Name: "Intake", Position: 0, - DocTasks: []workflowdef.DocTaskDef{{ - Key: "intake", Required: true, Blocking: true, DefaultEnabled: true, - OutputDocType: "intake", OutputDocTitle: "Customer Intake", - }}, - }}, - }, - }, 1) - if err != nil { - t.Fatalf("create wf: %v", err) - } - _, _ = wfSvc.PublishVersion(ctx, ver.WorkflowID, ver.Version, 1) - - createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{Name: "Phase5 Idempotency Test"}) - reqCreate := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(createBody)) - withBearerAuth(reqCreate, creatorToken) - recCreate := httptest.NewRecorder() - handler.ServeHTTP(recCreate, reqCreate) - var proj projectworkflow.Project - _ = json.Unmarshal(recCreate.Body.Bytes(), &proj) - t.Cleanup(func() { cleanupProject(t, pool, proj.ID) }) - creatorID := fetchAuthUserID(t, handler, creatorToken) reviewerID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "lizhuguan", "123456")) approverID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "wangzong", "123456")) - reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=1", proj.ID, wfKey), nil) + + var projectID int64 + code := "p5-idem-" + time.Now().Format("150405.000000") + if err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) + VALUES ($1, 'Phase5 Idempotency Test ' || $1, 'active', $2, 'planning', '', 1, 0) RETURNING id + `, code, creatorID).Scan(&projectID); err != nil { + t.Fatalf("insert project: %v", err) + } + ensureProjectMembers(t, pool, projectID, map[int64]string{creatorID: "owner", reviewerID: "reviewer", approverID: "approver"}) + t.Cleanup(func() { cleanupProject(t, pool, projectID) }) + + reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=0", projectID, workflowdef.OfficialKeyMain), nil) withBearerAuth(reqDefault, creatorToken) recDefault := httptest.NewRecorder() handler.ServeHTTP(recDefault, reqDefault) + if recDefault.Code != http.StatusOK { + t.Fatalf("default plan: %d %s", recDefault.Code, recDefault.Body.String()) + } var plan projectinstance.WorkflowPlan - _ = json.Unmarshal(recDefault.Body.Bytes(), &plan) + if err := json.Unmarshal(recDefault.Body.Bytes(), &plan); err != nil { + t.Fatalf("decode plan: %v", err) + } for i := range plan.Stages { for j := range plan.Stages[i].DocTasks { plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{ @@ -157,23 +146,28 @@ func TestPhase5_ActivateIdempotencyConflict_PG(t *testing.T) { } } saveBody, _ := json.Marshal(projectinstance.SavePlanInput{Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1}) - reqSave := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/api/v1/projects/%d/workflow-plan", proj.ID), bytes.NewReader(saveBody)) + reqSave := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/api/v1/projects/%d/workflow-plan", projectID), bytes.NewReader(saveBody)) withBearerAuth(reqSave, creatorToken) recSave := httptest.NewRecorder() handler.ServeHTTP(recSave, reqSave) + if recSave.Code != http.StatusOK { + t.Fatalf("save plan: %d %s", recSave.Code, recSave.Body.String()) + } - reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", proj.ID), nil) + reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", projectID), nil) withBearerAuth(reqPreview, creatorToken) recPreview := httptest.NewRecorder() handler.ServeHTTP(recPreview, reqPreview) + if recPreview.Code != http.StatusOK { + t.Fatalf("activation preview: %d %s", recPreview.Code, recPreview.Body.String()) + } var preview projectinstance.ActivationPreview _ = json.Unmarshal(recPreview.Body.Bytes(), &preview) - activateBody, _ := json.Marshal(projectinstance.ActivateInput{ ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, ExpectedPlanRevision: preview.ExpectedPlanRevision, }) - reqActivate := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", proj.ID), bytes.NewReader(activateBody)) + reqActivate := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", projectID), bytes.NewReader(activateBody)) withBearerAuth(reqActivate, creatorToken) recActivate := httptest.NewRecorder() handler.ServeHTTP(recActivate, reqActivate) @@ -182,7 +176,7 @@ func TestPhase5_ActivateIdempotencyConflict_PG(t *testing.T) { } // 二次激活应冲突 - reqActivate2 := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", proj.ID), bytes.NewReader(activateBody)) + reqActivate2 := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", projectID), bytes.NewReader(activateBody)) withBearerAuth(reqActivate2, creatorToken) recActivate2 := httptest.NewRecorder() handler.ServeHTTP(recActivate2, reqActivate2) diff --git a/internal/server/phase5_fault_injection_pg_test.go b/internal/server/phase5_fault_injection_pg_test.go index f9358e57..911ab9ee 100644 --- a/internal/server/phase5_fault_injection_pg_test.go +++ b/internal/server/phase5_fault_injection_pg_test.go @@ -22,32 +22,11 @@ import ( "chat-agent-system/internal/workflowdef" ) -// setupPGProjectReadyToAdvance 创建并激活工作流、完成 intake 文档批准,返回可推进实例。 +// setupPGProjectReadyToAdvance 激活官方 main 工作流、完成 V0.1 阶段 bd_intake 文档批准,返回可推进实例。 func setupPGProjectReadyToAdvance(t *testing.T, pool *pgxpool.Pool, handler http.Handler, srv *server.Server) (projectID, instanceID int64, creatorToken string, configRevision int) { t.Helper() ctx := context.Background() - wfSvc := workflowdef.NewService(pool, nil) - wfKey := "pg_p5_fault_" + time.Now().Format("150405.000000") - _, ver, err := wfSvc.CreateDefinition(ctx, workflowdef.CreateDefinitionInput{ - Key: wfKey, Name: "Phase5 Fault WF", - Definition: workflowdef.VersionDefinition{ - SchemaVersion: 1, - Stages: []workflowdef.StageDef{ - { - Key: "biz_intake", Name: "Intake", Position: 0, - DocTasks: []workflowdef.DocTaskDef{{ - Key: "intake", Required: true, Blocking: true, DefaultEnabled: true, - OutputDocType: "intake", OutputDocTitle: "Intake", - }}, - }, - {Key: "biz_feasibility", Name: "Feasibility", Position: 1, Optional: true}, - }, - }, - }, 1) - if err != nil { - t.Fatalf("create wf: %v", err) - } - _, _ = wfSvc.PublishVersion(ctx, ver.WorkflowID, ver.Version, 1) + seedOfficialWorkflowManifest(t, pool, workflowdef.OfficialKeyMain) creatorToken = loginTestUser(t, handler, "zhanggong", "123456") creatorID := fetchAuthUserID(t, handler, creatorToken) @@ -55,25 +34,27 @@ func setupPGProjectReadyToAdvance(t *testing.T, pool *pgxpool.Pool, handler http approverID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "wangzong", "123456")) code := "p5-fault-" + time.Now().Format("150405.000000") - err = pool.QueryRow(ctx, ` + err := pool.QueryRow(ctx, ` INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) - VALUES ($1, 'Phase5 Fault Test', 'active', $2, 'planning', '', 1, 0) RETURNING id + VALUES ($1, 'Phase5 Fault Test ' || $1, 'active', $2, 'planning', '', 1, 0) RETURNING id `, code, creatorID).Scan(&projectID) if err != nil { t.Fatalf("insert project: %v", err) } - _, _ = pool.Exec(ctx, ` - INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at) - VALUES ($1, $2, 'owner', 'active', now(), now()) ON CONFLICT DO NOTHING - `, projectID, creatorID) + ensureProjectMembers(t, pool, projectID, map[int64]string{creatorID: "owner", reviewerID: "reviewer", approverID: "approver"}) t.Cleanup(func() { cleanupProject(t, pool, projectID) }) - reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=1", projectID, wfKey), nil) + reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=0", projectID, workflowdef.OfficialKeyMain), nil) withBearerAuth(reqDefault, creatorToken) recDefault := httptest.NewRecorder() handler.ServeHTTP(recDefault, reqDefault) + if recDefault.Code != http.StatusOK { + t.Fatalf("default plan: %d %s", recDefault.Code, recDefault.Body.String()) + } var plan projectinstance.WorkflowPlan - _ = json.Unmarshal(recDefault.Body.Bytes(), &plan) + if err := json.Unmarshal(recDefault.Body.Bytes(), &plan); err != nil { + t.Fatalf("decode plan: %v", err) + } for i := range plan.Stages { for j := range plan.Stages[i].DocTasks { plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{ @@ -86,11 +67,17 @@ func setupPGProjectReadyToAdvance(t *testing.T, pool *pgxpool.Pool, handler http withBearerAuth(reqSave, creatorToken) recSave := httptest.NewRecorder() handler.ServeHTTP(recSave, reqSave) + if recSave.Code != http.StatusOK { + t.Fatalf("save plan: %d %s", recSave.Code, recSave.Body.String()) + } reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", projectID), nil) withBearerAuth(reqPreview, creatorToken) recPreview := httptest.NewRecorder() handler.ServeHTTP(recPreview, reqPreview) + if recPreview.Code != http.StatusOK { + t.Fatalf("activation preview: %d %s", recPreview.Code, recPreview.Body.String()) + } var preview projectinstance.ActivationPreview _ = json.Unmarshal(recPreview.Body.Bytes(), &preview) @@ -102,18 +89,27 @@ func setupPGProjectReadyToAdvance(t *testing.T, pool *pgxpool.Pool, handler http withBearerAuth(reqActivate, creatorToken) recActivate := httptest.NewRecorder() handler.ServeHTTP(recActivate, reqActivate) + if recActivate.Code != http.StatusOK { + t.Fatalf("activate: %d %s", recActivate.Code, recActivate.Body.String()) + } var activated projectinstance.ActivateResult - _ = json.Unmarshal(recActivate.Body.Bytes(), &activated) + if err := json.Unmarshal(recActivate.Body.Bytes(), &activated); err != nil { + t.Fatalf("decode activate: %v", err) + } instanceID = activated.Instance.ID configRevision = activated.Instance.ConfigRevision docID := fmt.Sprintf("doc_p5_fault_%d", projectID) docEngine := bizdev.Snapshot(srv.BizdevService()).Gov - _, _ = docEngine.CreateOrGetDocument(ctx, docID, projectID, "intake", "Intake", "V0.1", "", creatorID, "A", reviewerID, "R", approverID, "P") + _, _ = docEngine.CreateOrGetDocument(ctx, docID, projectID, "bd_intake", "用户初始需求收集表", "V0.1", "", creatorID, "A", reviewerID, "R", approverID, "P") _ = docEngine.MarkDocumentContentUpdated(ctx, docID, "hash_p5_fault", false) _, _ = docEngine.SubmitForReview(ctx, docID, creatorID, "author", "Author") _, _ = docEngine.ReviewApprove(ctx, docID, reviewerID, "reviewer", "Reviewer", "ok") _, _ = docEngine.ApproverApprove(ctx, docID, approverID, "approver", "Approver", "ok") + instSvc := projectinstance.NewService(projectinstance.NewStore(pool), nil, nil, nil) + if _, err = instSvc.HandleDocumentApproved(ctx, projectID, "bd_intake"); err != nil { + t.Fatalf("handle document approved: %v", err) + } return projectID, instanceID, creatorToken, configRevision } @@ -128,9 +124,9 @@ func TestPhase5_ConcurrentStageAdvance_PG(t *testing.T) { _, instanceID, creatorToken, configRevision := setupPGProjectReadyToAdvance(t, pool, handler, srv) advanceBody, _ := json.Marshal(projectinstance.AdvanceStageInput{ - StageKey: "biz_intake", ExpectedConfigRevision: configRevision, + StageKey: "V0.1_triage", ExpectedConfigRevision: configRevision, }) - url := fmt.Sprintf("/api/v1/workflow-instances/%d/stages/biz_intake/advance", instanceID) + url := fmt.Sprintf("/api/v1/workflow-instances/%d/stages/V0.1_triage/advance", instanceID) var wg sync.WaitGroup results := make([]int, 2) diff --git a/internal/server/phase5_finalize_fault_pg_test.go b/internal/server/phase5_finalize_fault_pg_test.go index e0d8bb8b..dc90394e 100644 --- a/internal/server/phase5_finalize_fault_pg_test.go +++ b/internal/server/phase5_finalize_fault_pg_test.go @@ -65,8 +65,9 @@ func TestPhase5_RollingRestartReplay_PG(t *testing.T) { return "completed", nil }) + projectID := seedProjectForRun(t, pool, "run-roll", userID) record := &taskrun.RunRecord{ - RunID: runID, SessionID: "sess_roll", UserID: userID, TaskKey: "intake", Status: "running", + RunID: runID, SessionID: "sess_roll", ProjectID: projectID, UserID: userID, TaskKey: "intake", Status: "running", } _, _ = srv.TaskRuns().CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "") t.Cleanup(func() { diff --git a/internal/server/task_run_persistence_pg_test.go b/internal/server/task_run_persistence_pg_test.go index 2f715c67..e2d777fa 100644 --- a/internal/server/task_run_persistence_pg_test.go +++ b/internal/server/task_run_persistence_pg_test.go @@ -44,9 +44,11 @@ func TestTaskRun_PersistenceAndReplayAfterRestart_PG(t *testing.T) { return srv.FinalizeTaskRunProjection(ctx, record, result) }) + userID := fetchAuthUserID(t, handler, token) + projectID := seedProjectForRun(t, pool, "run-persist", userID) record := &taskrun.RunRecord{ - RunID: runID, SessionID: "sess_pg_persist", ProjectID: 0, - UserID: fetchAuthUserID(t, handler, token), TaskKey: "intake", Status: "running", + RunID: runID, SessionID: "sess_pg_persist", ProjectID: projectID, + UserID: userID, TaskKey: "intake", Status: "running", } _, err := srv.TaskRuns().CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "") if err != nil { diff --git a/internal/server/task_run_resume_pg_test.go b/internal/server/task_run_resume_pg_test.go index 7e536c94..66f62725 100644 --- a/internal/server/task_run_resume_pg_test.go +++ b/internal/server/task_run_resume_pg_test.go @@ -13,7 +13,6 @@ import ( "time" "chat-agent-system/internal/bizdev" - "chat-agent-system/internal/config" "chat-agent-system/internal/server" "chat-agent-system/internal/stream" "chat-agent-system/internal/taskrun" @@ -27,12 +26,8 @@ func TestTaskRun_ResumeInterrupted_PG(t *testing.T) { defer pool.Close() dataDir := t.TempDir() - cfg := &config.Config{ - Env: "dev", DataDir: dataDir, AgentBackend: "stub", AllowInMemory: false, - ContextDbHost: "127.0.0.1", ContextDbPort: 5433, ContextDbUser: "app", - ContextDbPassword: "dev_secret_db_pass", ContextDbName: "agent_system_dev", - AdminUsername: "admin", AdminPassword: "admin-secret-password-123", - } + cfg := configFromTestDatabaseURL(t) + cfg.DataDir = dataDir svc := bizdev.MustNewService(t, cfg) srv := server.NewServer(cfg, svc) handler := srv.Routes() @@ -46,12 +41,13 @@ func TestTaskRun_ResumeInterrupted_PG(t *testing.T) { _ = os.WriteFile(filepath.Join(workDir, "output.md"), []byte("# resumed\n"), 0o644) store := taskrun.NewPGStore(pool) + projectID := seedProjectForRun(t, pool, "run-resume", userID) runID := "run_resume_" + time.Now().Format("150405.000000") _, err := pool.Exec(context.Background(), ` INSERT INTO task_runs ( - run_id, session_id, user_id, task_key, status, workspace_id, error_message, created_at, updated_at - ) VALUES ($1, 'sess_resume', $2, 'intake', 'interrupted', $3, 'worker lease expired', now(), now()) - `, runID, userID, workDir) + run_id, session_id, project_id, user_id, task_key, status, workspace_id, error_message, created_at, updated_at + ) VALUES ($1, 'sess_resume', $4, $2, 'intake', 'interrupted', $3, 'worker lease expired', now(), now()) + `, runID, userID, workDir, projectID) if err != nil { t.Fatalf("seed interrupted run: %v", err) } @@ -104,12 +100,8 @@ func TestTaskRun_ResumeRequiresConfirm_PG(t *testing.T) { defer pool.Close() dataDir := t.TempDir() - cfg := &config.Config{ - Env: "dev", DataDir: dataDir, AgentBackend: "stub", AllowInMemory: false, - ContextDbHost: "127.0.0.1", ContextDbPort: 5433, ContextDbUser: "app", - ContextDbPassword: "dev_secret_db_pass", ContextDbName: "agent_system_dev", - AdminUsername: "admin", AdminPassword: "admin-secret-password-123", - } + cfg := configFromTestDatabaseURL(t) + cfg.DataDir = dataDir srv := server.NewServer(cfg, bizdev.MustNewService(t, cfg)) handler := srv.Routes() token := loginTestUser(t, handler, "zhanggong", "123456") @@ -145,12 +137,8 @@ func TestTaskRun_GetRunShowsResumable_PG(t *testing.T) { defer pool.Close() dataDir := t.TempDir() - cfg := &config.Config{ - Env: "dev", DataDir: dataDir, AgentBackend: "stub", AllowInMemory: false, - ContextDbHost: "127.0.0.1", ContextDbPort: 5433, ContextDbUser: "app", - ContextDbPassword: "dev_secret_db_pass", ContextDbName: "agent_system_dev", - AdminUsername: "admin", AdminPassword: "admin-secret-password-123", - } + cfg := configFromTestDatabaseURL(t) + cfg.DataDir = dataDir handler := server.NewServer(cfg, bizdev.MustNewService(t, cfg)).Routes() token := loginTestUser(t, handler, "zhanggong", "123456") userID := fetchAuthUserID(t, handler, token) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 690f98fc..bce60517 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -162,3 +162,20 @@ func seedOfficialWorkflowManifest(t *testing.T, pool *pgxpool.Pool, workflowKey t.Fatalf("seed official workflow %s: %v", workflowKey, err) } } + +// seedProjectForRun 创建带 active 成员的最小项目,返回 projectID;用于要求 +// 项目作用域的 taskrun/事件链路测试(events.ValidateEnvelope 拒绝 project<=0)。 +func seedProjectForRun(t *testing.T, pool *pgxpool.Pool, prefix string, ownerUserID int64) int64 { + t.Helper() + code := prefix + "-" + time.Now().Format("150405.000000") + var projectID int64 + if err := pool.QueryRow(context.Background(), ` + INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status) + VALUES ($1, 'Run Scope Project ' || $1, 'active', $2, 'active') RETURNING id + `, code, ownerUserID).Scan(&projectID); err != nil { + t.Fatalf("seed project: %v", err) + } + ensureProjectMembers(t, pool, projectID, map[int64]string{ownerUserID: "owner"}) + t.Cleanup(func() { cleanupProject(t, pool, projectID) }) + return projectID +} diff --git a/internal/server/workflow_governance_pg_test.go b/internal/server/workflow_governance_pg_test.go index fce3b3d6..e857584e 100644 --- a/internal/server/workflow_governance_pg_test.go +++ b/internal/server/workflow_governance_pg_test.go @@ -134,7 +134,7 @@ func TestWorkflowGovernance_PG_ActivateApproveAdvanceChain(t *testing.T) { var err error err = pool.QueryRow(ctx, ` INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) - VALUES ($1, 'PG Phase5 Governance Test', 'active', $2, 'planning', '', 1, 0) RETURNING id + VALUES ($1, 'PG Phase5 Governance Test ' || $1, 'active', $2, 'planning', '', 1, 0) RETURNING id `, code, creatorID).Scan(&projectID) if err != nil { t.Fatalf("insert project: %v", err) @@ -220,7 +220,7 @@ func TestWorkflowGovernance_PG_ActivateApproveAdvanceChain(t *testing.T) { // 4. 文档批准链(经 bizdev DocGov 引擎,触发 onWorkflowDocumentApproved hook) docID := fmt.Sprintf("doc_pg_p5_intake_%d", projectID) docEngine := bizdev.Snapshot(srv.BizdevService()).Gov - _, err = docEngine.CreateOrGetDocument(ctx, docID, projectID, "intake", "Intake Doc", "V0.1", "", creatorID, "Author", reviewerID, "Reviewer", approverID, "Approver") + _, err = docEngine.CreateOrGetDocument(ctx, docID, projectID, "bd_intake", "用户初始需求收集表", "V0.1", "", creatorID, "Author", reviewerID, "Reviewer", approverID, "Approver") if err != nil { t.Fatalf("create doc: %v", err) } diff --git a/internal/storage/cas_db_test.go b/internal/storage/cas_db_test.go index cfe99801..08d40a2e 100644 --- a/internal/storage/cas_db_test.go +++ b/internal/storage/cas_db_test.go @@ -10,9 +10,14 @@ import ( ) func TestCAS_DatabaseIntegration(t *testing.T) { - dsn := os.Getenv("CONTEXT_DB_DSN") + // 隔离优先:TEST_DATABASE_URL 指向专用测试库;未提供时跳过,禁止默认连开发库。 + dsn := os.Getenv("TEST_DATABASE_URL") if dsn == "" { - dsn = "postgres://app:dev_secret_db_pass@127.0.0.1:5433/agent_system_dev?sslmode=disable" + dsn = os.Getenv("CONTEXT_DB_DSN") + } + if dsn == "" { + t.Skip("TEST_DATABASE_URL/CONTEXT_DB_DSN not set; skipping CAS DB integration test") + return } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) @@ -44,13 +49,14 @@ func TestCAS_DatabaseIntegration(t *testing.T) { // 准备测试项目与用户 var projID int64 - projCode := "TEST_PROJ_" + time.Now().Format("20060102150405") + stamp := time.Now().Format("20060102150405.000000") + projCode := "TEST_PROJ_" + stamp err = pool.QueryRow(ctx, ` INSERT INTO projects (code, name, customer_name, device_type) - VALUES ($1, '测试内窥镜系统', '北京水木东方', '内窥镜') + VALUES ($1, '测试内窥镜系统 ' || $2, '北京水木东方', '内窥镜') ON CONFLICT (code) DO UPDATE SET name = EXCLUDED.name RETURNING id - `, projCode).Scan(&projID) + `, projCode, stamp).Scan(&projID) if err != nil { t.Fatalf("insert project failed: %v", err) }