test(pg): isolate PG integration tests from shared dev database
修复 go test -short 在设置 TEST_DATABASE_URL 时的既有欠债: - storage/notification/docgov:移除硬编码 agent_system_dev 回退 DSN,改为 无 TEST_DATABASE_URL 即 skip;固定项目名改为 name||code 唯一化。 - docgov engine_db:改为隔离库 + 自建项目行,替换随机 projectID 撞外键的旧写法。 - server/*_pg、projectinstance/pg_integration:迁移到官方 main 工作流 (BuildDefaultPlan official-only 治理后的正确前提),补齐在任成员 (DUTY_ASSIGNEE_NOT_ACTIVE_MEMBER)、任务键 bd_intake、阶段键 V0.1_triage、 项目作用域 run record(events.ValidateEnvelope 拒绝 project<=0)、 task_run_resume 去除内嵌 dev 凭证的 config 硬编码。 - 新增 seedProjectForRun 测试辅助。 验证:fresh 隔离库(55432)上 go test -short -p 1 ./... 除 archtest (另行注册表分支修复)外全部通过;包内并行迁移竞争需 -p 1。
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user