package projectinstance_test import ( "context" "testing" "time" "github.com/jackc/pgx/v5/pgxpool" "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" ) func openPGPool(t *testing.T) *pgxpool.Pool { t.Helper() dsn := db.SkipIfNoTestDatabaseURL(t) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() pool, err := pgxpool.New(ctx, dsn) if err != nil { t.Skip(err) } if err := pool.Ping(ctx); err != nil { pool.Close() t.Skip(err) } migCtx, migCancel := context.WithTimeout(context.Background(), 30*time.Second) defer migCancel() if err := db.RunMigrations(migCtx, pool, db.ResolveMigrationsDir()); err != nil { pool.Close() t.Skipf("migration failed: %v", err) } if !projectinstance.TablesReady(ctx, pool) { pool.Close() t.Skip("project_workflow_instances not migrated") } if !workflowdef.TablesReady(ctx, pool) { pool.Close() t.Skip("workflow_definitions not migrated") } return pool } func TestPG_ActivationAndStageAdvance(t *testing.T) { pool := openPGPool(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) 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) } wfStore := workflowdef.NewStore(pool) 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") 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 `, code).Scan(&projectID) if err != nil { t.Fatalf("insert project: %v", err) } t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) }) plan, err := svc.BuildDefaultPlan(ctx, wfKey, 0) if err != nil { t.Fatalf("default plan: %v", err) } 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].Duties = projectinstance.DocDuty{ AuthorUserID: 2, ReviewerUserID: 3, ApproverUserID: 4, } } } } _, err = svc.SaveWorkflowPlan(ctx, projectID, 2, true, projectinstance.SavePlanInput{ Plan: *plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, }) if err != nil { t.Fatalf("save plan: %v", err) } preview, err := svc.ActivationPreview(ctx, projectID, 2, true) if err != nil { t.Fatalf("preview: %v", err) } activated, err := svc.Activate(ctx, projectID, 2, true, projectinstance.ActivateInput{ ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, ExpectedPlanRevision: preview.ExpectedPlanRevision, }) if err != nil { t.Fatalf("activate: %v", err) } docID := "doc_pg_p3_intake_" + code engine := docgov.NewGovernanceEngine(pool) _, err = engine.CreateOrGetDocument(ctx, docID, projectID, "intake", "客户需求采集", "V0.1", "", 2, "张", 3, "李", 4, "王") 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") if err != nil { t.Fatalf("approve doc: %v", err) } approvedResult, err := svc.HandleDocumentApproved(ctx, projectID, "intake") if err != nil { t.Fatalf("handle document approved: %v", err) } if approvedResult == nil || !approvedResult.StageReady { 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) if err != nil { t.Fatalf("readiness: %v", err) } if !readiness.Ready { t.Fatalf("expected ready stage, issues=%+v tasks=%+v", readiness.BlockingIssues, readiness.DocTasks) } advanced, err := svc.AdvanceStage(ctx, activated.Instance.ID, 2, true, projectinstance.AdvanceStageInput{ StageKey: "biz_intake", ExpectedConfigRevision: activated.Instance.ConfigRevision, }) if err != nil { t.Fatalf("advance: %v", err) } if advanced.NextStageKey == "" && !advanced.InstanceDone { t.Fatalf("expected next stage or done, got %+v", advanced) } }