Files
Agent-System/internal/server/workflow_governance_pg_test.go
weichen d6f5e0e28a feat(projectcontext): bizdev 注入项目对话并完成 PG 治理 E2E
新增 ctxmgr 项目对话采集并接入 PM/TE BuildSnapshot;PG 测试统一 TEST_DATABASE_URL 并使用三正式工作流 key。
2026-09-13 02:50:23 +00:00

265 lines
9.7 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package server_test
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"chat-agent-system/internal/auth"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/db"
"chat-agent-system/internal/notification"
"chat-agent-system/internal/projectinstance"
"chat-agent-system/internal/server"
"chat-agent-system/internal/workflowdef"
)
func openPGPoolWithMigrations(t *testing.T) *pgxpool.Pool {
t.Helper()
dsn := db.SkipIfNoTestDatabaseURL(t)
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
t.Skipf("postgres unavailable: %v", err)
return nil
}
if err := pool.Ping(ctx); err != nil {
pool.Close()
t.Skipf("postgres ping failed: %v", err)
return nil
}
migCtx, migCancel := context.WithTimeout(context.Background(), 60*time.Second)
defer migCancel()
if err := db.RunMigrations(migCtx, pool, db.ResolveMigrationsDir()); err != nil {
pool.Close()
t.Skipf("migration failed: %v", err)
return nil
}
if !projectinstance.TablesReady(ctx, pool) || !workflowdef.TablesReady(ctx, pool) {
pool.Close()
t.Skip("workflow/project instance tables not ready")
}
if auth.IAMTablesReady(ctx, pool) {
testCfg := configFromTestDatabaseURL(t)
authStore, err := auth.NewPgStore(pool, testCfg)
if err == nil {
_ = auth.ReactivateDevSeedUsers(ctx, authStore, testCfg)
}
}
return pool
}
func fetchAuthUserID(t *testing.T, handler http.Handler, token string) int64 {
t.Helper()
req := httptest.NewRequest(http.MethodGet, "/api/v1/auth/me", nil)
withBearerAuth(req, token)
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("auth/me: %d %s", rec.Code, rec.Body.String())
}
var me struct {
User struct {
ID int64 `json:"id"`
} `json:"user"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &me); err != nil {
t.Fatalf("decode auth/me: %v", err)
}
if me.User.ID <= 0 {
t.Fatal("expected positive user id from auth/me")
}
return me.User.ID
}
func setupPGGovernanceServer(t *testing.T, pool *pgxpool.Pool) (http.Handler, *server.Server) {
t.Helper()
cfg := configFromTestDatabaseURL(t)
svc := bizdev.MustNewService(t, cfg)
if svc.Pool() == nil {
t.Skip("bizdev service not connected to postgres")
}
srv := server.NewServer(cfg, svc)
return srv.Routes(), srv
}
func TestWorkflowGovernance_PG_ActivateApproveAdvanceChain(t *testing.T) {
pool := openPGPoolWithMigrations(t)
if pool == nil {
return
}
defer pool.Close()
ctx := context.Background()
handler, srv := setupPGGovernanceServer(t, pool)
wfKey := workflowdef.OfficialKeyMain
seedOfficialWorkflowManifest(t, pool, wfKey)
creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
guestToken := loginTestUser(t, handler, "guest", "123456")
creatorID := fetchAuthUserID(t, handler, creatorToken)
// 1. 创建项目(PG 直写,与 projectinstance PG 集成测试一致)
var projectID int64
code := "pg-p5-" + time.Now().Format("150405.000")
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
`, code, creatorID).Scan(&projectID)
if err != nil {
t.Fatalf("insert project: %v", err)
}
_, 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 (project_id, user_id) DO NOTHING
`, projectID, creatorID)
if err != nil {
t.Fatalf("insert project member: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
})
// 2. 获取默认规划并保存三权
reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=0", projectID, wfKey), 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
if err := json.Unmarshal(recDefault.Body.Bytes(), &plan); err != nil {
t.Fatalf("decode plan: %v", err)
}
reviewerID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "lizhuguan", "123456"))
approverID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "wangzong", "123456"))
for i := range plan.Stages {
for j := range plan.Stages[i].DocTasks {
plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{
AuthorUserID: creatorID, ReviewerUserID: reviewerID, ApproverUserID: approverID,
}
}
}
saveBody, _ := json.Marshal(projectinstance.SavePlanInput{
Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1,
})
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())
}
// 3. 激活预览 + 确认激活
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
if err := json.Unmarshal(recPreview.Body.Bytes(), &preview); err != nil {
t.Fatalf("decode preview: %v", err)
}
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", projectID), bytes.NewReader(activateBody))
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
if err := json.Unmarshal(recActivate.Body.Bytes(), &activated); err != nil {
t.Fatalf("decode activate: %v", err)
}
instanceID := activated.Instance.ID
// 4. 文档批准链(经 bizdev DocGov 引擎,触发 onWorkflowDocumentApproved hook)
docID := fmt.Sprintf("doc_pg_p5_intake_%d", projectID)
docEngine := srv.BizdevService().DocGov()
_, err = docEngine.CreateOrGetDocument(ctx, docID, projectID, "intake", "Intake Doc", "V0.1", "", creatorID, "Author", reviewerID, "Reviewer", approverID, "Approver")
if err != nil {
t.Fatalf("create doc: %v", err)
}
_ = docEngine.MarkDocumentContentUpdated(ctx, docID, "hash_pg_p5", false)
_, _ = docEngine.SubmitForReview(ctx, docID, creatorID, "author", "Author")
_, _ = docEngine.ReviewApprove(ctx, docID, reviewerID, "reviewer", "Reviewer", "ok")
_, err = docEngine.ApproverApprove(ctx, docID, approverID, "approver", "Approver", "ok")
if err != nil {
t.Fatalf("approve doc: %v", err)
}
// 5. 创建者应收到 stage ready 通知
reqNotify := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/notifications", projectID), nil)
withBearerAuth(reqNotify, creatorToken)
recNotify := httptest.NewRecorder()
handler.ServeHTTP(recNotify, reqNotify)
if recNotify.Code != http.StatusOK {
t.Fatalf("list notifications: %d %s", recNotify.Code, recNotify.Body.String())
}
var notifyResp struct {
Items []map[string]any `json:"items"`
}
if err := json.Unmarshal(recNotify.Body.Bytes(), &notifyResp); err != nil {
t.Fatalf("decode notifications: %v body=%s", err, recNotify.Body.String())
}
foundReady := false
for _, n := range notifyResp.Items {
if n["type"] == notification.TypeStageReadyToAdvance {
foundReady = true
break
}
}
if !foundReady {
t.Fatalf("expected stage_ready_to_advance notification, got %+v", notifyResp.Items)
}
// 6. ABAC:访客无权读取项目上下文
reqCtxGuest := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/context/project?project_id=%d", projectID), nil)
withBearerAuth(reqCtxGuest, guestToken)
recCtxGuest := httptest.NewRecorder()
handler.ServeHTTP(recCtxGuest, reqCtxGuest)
if recCtxGuest.Code != http.StatusForbidden {
t.Fatalf("guest project context expected 403, got %d %s", recCtxGuest.Code, recCtxGuest.Body.String())
}
// 7. 创建者推进阶段
advanceBody, _ := json.Marshal(projectinstance.AdvanceStageInput{
StageKey: "V0.1_triage",
ExpectedConfigRevision: activated.Instance.ConfigRevision,
})
reqAdvance := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/workflow-instances/%d/stages/V0.1_triage/advance", instanceID), bytes.NewReader(advanceBody))
withBearerAuth(reqAdvance, creatorToken)
recAdvance := httptest.NewRecorder()
handler.ServeHTTP(recAdvance, reqAdvance)
if recAdvance.Code != http.StatusOK {
t.Fatalf("advance stage: %d %s", recAdvance.Code, recAdvance.Body.String())
}
var advanceResult projectinstance.AdvanceStageResult
if err := json.Unmarshal(recAdvance.Body.Bytes(), &advanceResult); err != nil {
t.Fatalf("decode advance: %v", err)
}
if advanceResult.NextStageKey != "V0.5_feasibility" && !advanceResult.InstanceDone {
t.Fatalf("unexpected advance result: %+v", advanceResult)
}
}