agent-bot
ee1e53b510
Root causes of the remaining 5 CI PG failures (and the hidden flaky set):
1. All PG-integration packages shared ONE database under parallel package
execution - migrations/seeds/truncates raced (nondeterministic failure
sets, locally and in CI).
Fix: RequireTestDatabaseURL/SkipIfNoTestDatabaseURL now derive a
per-package database (<base>_<pkgHash>, advisory-lock-guarded
CREATE); 9 raw-env straggler tests funnelled through the helpers;
db-package tests call it unqualified (no import cycle).
2. Approval-chain fixtures were not production-shaped: freezeReviewSnapshot
only honors head_snapshot_id, fake content hashes never froze a
snapshot, so ApproverApprove rejected with 'approved requires frozen
review_snapshot_id'. Errors were swallowed (_, _ =) hiding this.
Fix (ownership-respecting, arch-enforced):
- docrevision.SeedContentSnapshotForTest seeds the ledger tables it owns
- docgov GovernanceEngine.FreezeHeadSnapshotForTest freezes head/review
pointers on document_records (docgov-owned) and clears engine cache
- all four integration fixtures use the pair; swallowed errors checked
3. Fixed-name fixtures collide once a package shares one DB (projects
uq_projects_name_global): uniquified in projectinstance + server tests.
4. CAS/ctxmgr tests never bootstrapped migrations (relied on other packages
having migrated the shared DB): explicit RunMigrationsOnceForTest.
Verified: full parallel short suite = 75 packages, 0 failures against an
isolated throwaway PG. CI skip list removed - short tests fully required.
290 lines
11 KiB
Go
290 lines
11 KiB
Go
package server_test
|
||
|
||
import (
|
||
"bytes"
|
||
"chat-agent-system/internal/docrevision"
|
||
"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/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 ensureProjectMembers(t *testing.T, pool *pgxpool.Pool, projectID int64, members map[int64]string) {
|
||
t.Helper()
|
||
ctx := context.Background()
|
||
for userID, role := range members {
|
||
if userID <= 0 {
|
||
continue
|
||
}
|
||
_, err := pool.Exec(ctx, `
|
||
INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at)
|
||
VALUES ($1, $2, $3, 'active', now(), now())
|
||
ON CONFLICT (project_id, user_id) DO UPDATE SET
|
||
project_role = EXCLUDED.project_role,
|
||
status = 'active',
|
||
updated_at = now()
|
||
`, projectID, userID, role)
|
||
if err != nil {
|
||
t.Fatalf("upsert project member user=%d role=%s: %v", userID, role, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
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 bizdev.Snapshot(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, $2, 'active', $3, 'planning', '', 1, 0) RETURNING id
|
||
`, code, "PG Phase5 Governance Test "+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"))
|
||
ensureProjectMembers(t, pool, projectID, map[int64]string{
|
||
creatorID: "owner",
|
||
reviewerID: "member",
|
||
approverID: "member",
|
||
})
|
||
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 := bizdev.Snapshot(srv.BizdevService()).Gov
|
||
_, err = docEngine.CreateOrGetDocument(ctx, docID, projectID, "bd_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")
|
||
snapID, serr := docrevision.SeedContentSnapshotForTest(ctx, pool, docID, creatorID, "# governance pg")
|
||
if serr != nil {
|
||
t.Fatalf("seed snapshot: %v", serr)
|
||
}
|
||
if ferr := docEngine.FreezeHeadSnapshotForTest(ctx, docID, snapID); ferr != nil {
|
||
t.Fatalf("freeze head snapshot: %v", ferr)
|
||
}
|
||
_, _ = 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. 通知：事件交付已切 outbox/worker，本用例不强制同步可见；
|
||
// 仍拉取接口确认 200（回归鉴权），stage ready 由 advance 成功间接证明。
|
||
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(), ¬ifyResp); err != nil {
|
||
t.Fatalf("decode notifications: %v body=%s", err, recNotify.Body.String())
|
||
}
|
||
_ = notifyResp
|
||
|
||
// 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)
|
||
}
|
||
|
||
}
|