From c2d4e9a55c3f83fb389d4b8e9aa4024a65215cf2 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 22:16:44 +0000 Subject: [PATCH] test(docduty): add isolated PG heal, backfill, and concurrency fixtures Require TEST_DATABASE_URL and refuse agent_system_dev. Script creates a throwaway database, runs packages serially, then drops it. Co-authored-by: Cursor --- internal/authz/authorizer_pg_test.go | 129 ++++++++++++++++ internal/db/testdb.go | 22 +++ internal/docduty/store_pg_test.go | 171 +++++++++++++++++++-- scripts/run-isolated-duty-sync-pg-tests.sh | 78 ++++++++++ 4 files changed, 386 insertions(+), 14 deletions(-) create mode 100644 internal/authz/authorizer_pg_test.go create mode 100755 scripts/run-isolated-duty-sync-pg-tests.sh diff --git a/internal/authz/authorizer_pg_test.go b/internal/authz/authorizer_pg_test.go new file mode 100644 index 00000000..0a6add27 --- /dev/null +++ b/internal/authz/authorizer_pg_test.go @@ -0,0 +1,129 @@ +package authz + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/db" + "chat-agent-system/internal/docduty" +) + +func TestUpsertProjectMemberDuty_ConcurrentLastWriter_PG(t *testing.T) { + if testing.Short() { + t.Skip("skipping PostgreSQL duty concurrency in -short") + } + dsn := db.SkipIfNoTestDatabaseURL(t) + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("postgres unavailable: %v", err) + } + defer pool.Close() + if err := pool.Ping(ctx); err != nil { + t.Fatalf("postgres ping failed: %v", err) + } + if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil { + t.Fatalf("migration failed: %v", err) + } + + suffix := time.Now().UnixNano() + ownerID := insertAuthzUser(t, ctx, pool, suffix, "owner") + reviewerA := insertAuthzUser(t, ctx, pool, suffix, "ra") + reviewerB := insertAuthzUser(t, ctx, pool, suffix, "rb") + + var projectID int64 + if err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id) + VALUES ($1, $2, 'active', $3) + RETURNING id + `, fmt.Sprintf("duty-race-%d", suffix), "concurrent reviewer", ownerID).Scan(&projectID); err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3)`, ownerID, reviewerA, reviewerB) + }) + + a := NewAuthorizer(pool) + dutyAuthor := docduty.DutyAuthor + if err := a.UpsertProjectMemberSpec(ctx, projectID, ownerID, MemberUpsert{ + ProjectRole: RoleOwner, DocumentDuty: &dutyAuthor, + }); err != nil { + t.Fatal(err) + } + if err := a.UpsertProjectMember(ctx, projectID, reviewerA, RoleMember); err != nil { + t.Fatal(err) + } + if err := a.UpsertProjectMember(ctx, projectID, reviewerB, RoleMember); err != nil { + t.Fatal(err) + } + + duty := docduty.DutyReviewer + var wg sync.WaitGroup + errCh := make(chan error, 16) + targets := []int64{reviewerA, reviewerB} + for i := 0; i < 12; i++ { + target := targets[i%2] + wg.Add(1) + go func(uid int64) { + defer wg.Done() + if err := a.UpsertProjectMemberSpec(ctx, projectID, uid, MemberUpsert{ + ProjectRole: RoleMember, DocumentDuty: &duty, + }); err != nil { + errCh <- err + } + }(target) + } + wg.Wait() + close(errCh) + for upsertErr := range errCh { + t.Fatalf("concurrent upsert failed after retry: %v", upsertErr) + } + + rows, err := pool.Query(ctx, ` + SELECT user_id FROM project_members + WHERE project_id = $1 AND status = 'active' AND document_duty = 'reviewer' + ORDER BY user_id + `, projectID) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var holders []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + t.Fatal(err) + } + holders = append(holders, id) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if len(holders) != 1 { + t.Fatalf("want exactly one active reviewer, got %v", holders) + } + if holders[0] != reviewerA && holders[0] != reviewerB { + t.Fatalf("unexpected reviewer %d", holders[0]) + } +} + +func insertAuthzUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix int64, kind string) int64 { + t.Helper() + var id int64 + if err := pool.QueryRow(ctx, ` + INSERT INTO users (account_id, name, role) + VALUES ($1, $2, 'engineer') + RETURNING id + `, fmt.Sprintf("authz_%s_%d", kind, suffix), "用户"+kind).Scan(&id); err != nil { + t.Fatalf("insert user %s: %v", kind, err) + } + return id +} diff --git a/internal/db/testdb.go b/internal/db/testdb.go index b71f192f..4abbb032 100644 --- a/internal/db/testdb.go +++ b/internal/db/testdb.go @@ -1,11 +1,33 @@ package db import ( + "context" "os" "strings" + "sync" "testing" + + "github.com/jackc/pgx/v5/pgxpool" ) +var ( + migrateOnceMu sync.Mutex + migratedDSNs = map[string]error{} +) + +// RunMigrationsOnceForTest 同一 DSN 只迁一次，避免并行包抢同一隔离库。 +func RunMigrationsOnceForTest(ctx context.Context, pool *pgxpool.Pool, dsn string) error { + key := strings.TrimSpace(dsn) + migrateOnceMu.Lock() + defer migrateOnceMu.Unlock() + if err, ok := migratedDSNs[key]; ok { + return err + } + err := RunMigrations(ctx, pool, ResolveMigrationsDir()) + migratedDSNs[key] = err + return err +} + const testDatabaseEnv = "TEST_DATABASE_URL" // RequireTestDatabaseURL 集成测试必须显式提供隔离测试库，禁止默认连开发库。 diff --git a/internal/docduty/store_pg_test.go b/internal/docduty/store_pg_test.go index a92e5f5d..0b8cb796 100644 --- a/internal/docduty/store_pg_test.go +++ b/internal/docduty/store_pg_test.go @@ -11,30 +11,45 @@ import ( "chat-agent-system/internal/db" ) -func TestProjectReviewerDutySync_PG(t *testing.T) { +func openDutyTestPool(t *testing.T) (context.Context, context.CancelFunc, *pgxpool.Pool) { + t.Helper() if testing.Short() { t.Skip("skipping PostgreSQL duty sync integration in -short") } dsn := db.SkipIfNoTestDatabaseURL(t) ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) - defer cancel() pool, err := pgxpool.New(ctx, dsn) if err != nil { - t.Skipf("postgres unavailable: %v", err) + cancel() + t.Fatalf("postgres unavailable: %v", err) } - defer pool.Close() if err := pool.Ping(ctx); err != nil { - t.Skipf("postgres ping failed: %v", err) + pool.Close() + cancel() + t.Fatalf("postgres ping failed: %v", err) } - if err := db.RunMigrations(ctx, pool, db.ResolveMigrationsDir()); err != nil { - t.Skipf("migration failed: %v", err) + if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil { + pool.Close() + cancel() + t.Fatalf("migration failed: %v", err) } + t.Cleanup(func() { + pool.Close() + cancel() + }) + return ctx, cancel, pool +} + +func TestProjectReviewerDutySync_PG(t *testing.T) { + ctx, _, pool := openDutyTestPool(t) suffix := time.Now().UnixNano() authorID := insertDutyUser(t, ctx, pool, suffix, "author") reviewerID := insertDutyUser(t, ctx, pool, suffix, "reviewer") outsiderID := insertDutyUser(t, ctx, pool, suffix, "none") placeholderID := insertDutyUser(t, ctx, pool, suffix, "placeholder") + wrongApproverID := insertDutyUser(t, ctx, pool, suffix, "wrong_appr") + assignedApproverID := insertDutyUser(t, ctx, pool, suffix, "assigned_appr") var projectID int64 if err := pool.QueryRow(ctx, ` @@ -48,14 +63,16 @@ func TestProjectReviewerDutySync_PG(t *testing.T) { _, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id = $1`, projectID) _, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID) _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) - _, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3,$4)`, authorID, reviewerID, outsiderID, placeholderID) + _, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3,$4,$5,$6)`, authorID, reviewerID, outsiderID, placeholderID, wrongApproverID, assignedApproverID) }) if _, err := pool.Exec(ctx, ` INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at) VALUES ($1, $2, 'owner', 'author', 'active', now(), now()), - ($1, $3, 'member', 'reviewer', 'active', now(), now()) - `, projectID, authorID, reviewerID); err != nil { + ($1, $3, 'member', 'reviewer', 'active', now(), now()), + ($1, $4, 'member', '', 'active', now(), now()), + ($1, $5, 'member', 'approver', 'active', now(), now()) + `, projectID, authorID, reviewerID, wrongApproverID, assignedApproverID); err != nil { t.Fatalf("insert members: %v", err) } @@ -72,8 +89,8 @@ func TestProjectReviewerDutySync_PG(t *testing.T) { intakeID := fmt.Sprintf("doc_proj%d_intake", projectID) secondID := fmt.Sprintf("doc_proj%d_prelim_design", projectID) - insertDutyDoc(t, ctx, pool, projectID, intakeID, "intake", authorID, placeholderID, 0) - insertDutyDoc(t, ctx, pool, projectID, secondID, "prelim_design", authorID, placeholderID, 0) + insertDutyDoc(t, ctx, pool, projectID, intakeID, "intake", authorID, placeholderID, wrongApproverID) + insertDutyDoc(t, ctx, pool, projectID, secondID, "prelim_design", authorID, placeholderID, wrongApproverID) assigned, n, err := HealProjectDocuments(ctx, pool, projectID) if err != nil { @@ -87,6 +104,9 @@ func TestProjectReviewerDutySync_PG(t *testing.T) { } assertDocReviewer(t, ctx, pool, intakeID, reviewerID) assertDocReviewer(t, ctx, pool, secondID, reviewerID) + assertDocApprover(t, ctx, pool, intakeID, assignedApproverID) + assertDocApprover(t, ctx, pool, secondID, assignedApproverID) + assertDocAuthorName(t, ctx, pool, intakeID, "用户author") _, n2, err := HealProjectDocuments(ctx, pool, projectID) if err != nil { @@ -96,6 +116,94 @@ func TestProjectReviewerDutySync_PG(t *testing.T) { t.Fatalf("heal must be idempotent, second pass updated %d rows", n2) } assertDocReviewer(t, ctx, pool, secondID, reviewerID) + assertDocApprover(t, ctx, pool, secondID, assignedApproverID) +} + +func TestHealBinding_EmptyAssignedKeepsMember_PG(t *testing.T) { + ctx, _, pool := openDutyTestPool(t) + suffix := time.Now().UnixNano() + authorID := insertDutyUser(t, ctx, pool, suffix, "keep_author") + memberID := insertDutyUser(t, ctx, pool, suffix, "keep_member") + + var projectID int64 + if err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id) + VALUES ($1, $2, 'active', $3) + RETURNING id + `, fmt.Sprintf("duty-keep-%d", suffix), "empty assigned keep", authorID).Scan(&projectID); err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2)`, authorID, memberID) + }) + if _, err := pool.Exec(ctx, ` + INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at) + VALUES ($1, $2, 'owner', 'author', 'active', now(), now()), + ($1, $3, 'member', '', 'active', now(), now()) + `, projectID, authorID, memberID); err != nil { + t.Fatalf("insert members: %v", err) + } + docID := fmt.Sprintf("doc_proj%d_intake", projectID) + insertDutyDoc(t, ctx, pool, projectID, docID, "intake", authorID, memberID, memberID) + if _, err := pool.Exec(ctx, `UPDATE document_records SET author_name = $2 WHERE doc_id = $1`, docID, "用户keep_author"); err != nil { + t.Fatal(err) + } + + _, n, err := HealProjectDocuments(ctx, pool, projectID) + if err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("empty assigned reviewer/approver must keep existing member, updated %d", n) + } + assertDocReviewer(t, ctx, pool, docID, memberID) + assertDocApprover(t, ctx, pool, docID, memberID) +} + +func TestOwnerAuthorBackfill_MultiOwnerDeterministic_PG(t *testing.T) { + ctx, _, pool := openDutyTestPool(t) + suffix := time.Now().UnixNano() + lowOwner := insertDutyUser(t, ctx, pool, suffix, "low_owner") + creatorOwner := insertDutyUser(t, ctx, pool, suffix, "creator_owner") + if lowOwner >= creatorOwner { + t.Fatalf("expected inserted low owner id < creator, got %d >= %d", lowOwner, creatorOwner) + } + + var projectID int64 + if err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id) + VALUES ($1, $2, 'active', $3) + RETURNING id + `, fmt.Sprintf("duty-owners-%d", suffix), "multi owner backfill", creatorOwner).Scan(&projectID); err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2)`, lowOwner, creatorOwner) + }) + if _, err := pool.Exec(ctx, ` + INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at) + VALUES ($1, $2, 'owner', '', 'active', now(), now()), + ($1, $3, 'owner', '', 'active', now(), now()) + `, projectID, lowOwner, creatorOwner); err != nil { + t.Fatalf("insert owners: %v", err) + } + + if _, err := pool.Exec(ctx, OwnerAuthorBackfillOneProjectSQL, projectID); err != nil { + t.Fatalf("owner author backfill: %v", err) + } + assertMemberDuty(t, ctx, pool, projectID, creatorOwner, DutyAuthor) + assertMemberDuty(t, ctx, pool, projectID, lowOwner, DutyNone) + + if _, err := pool.Exec(ctx, OwnerAuthorBackfillOneProjectSQL, projectID); err != nil { + t.Fatalf("second backfill: %v", err) + } + assertMemberDuty(t, ctx, pool, projectID, creatorOwner, DutyAuthor) + assertMemberDuty(t, ctx, pool, projectID, lowOwner, DutyNone) } func insertDutyUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix int64, kind string) int64 { @@ -134,12 +242,47 @@ func insertDutyDoc(t *testing.T, ctx context.Context, pool *pgxpool.Pool, projec } func assertDocReviewer(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, want int64) { + t.Helper() + assertDocSlot(t, ctx, pool, docID, "reviewer_id", want) +} + +func assertDocApprover(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, want int64) { + t.Helper() + assertDocSlot(t, ctx, pool, docID, "approver_id", want) +} + +func assertDocSlot(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, column string, want int64) { t.Helper() var got int64 - if err := pool.QueryRow(ctx, `SELECT COALESCE(reviewer_id, 0) FROM document_records WHERE doc_id = $1`, docID).Scan(&got); err != nil { + q := fmt.Sprintf(`SELECT COALESCE(%s, 0) FROM document_records WHERE doc_id = $1`, column) + if err := pool.QueryRow(ctx, q, docID).Scan(&got); err != nil { t.Fatal(err) } if got != want { - t.Fatalf("%s reviewer_id=%d want %d", docID, got, want) + t.Fatalf("%s %s=%d want %d", docID, column, got, want) + } +} + +func assertDocAuthorName(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, want string) { + t.Helper() + var got string + if err := pool.QueryRow(ctx, `SELECT COALESCE(author_name, '') FROM document_records WHERE doc_id = $1`, docID).Scan(&got); err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("%s author_name=%q want %q", docID, got, want) + } +} + +func assertMemberDuty(t *testing.T, ctx context.Context, pool *pgxpool.Pool, projectID, userID int64, want string) { + t.Helper() + var got string + if err := pool.QueryRow(ctx, ` + SELECT COALESCE(document_duty, '') FROM project_members WHERE project_id = $1 AND user_id = $2 + `, projectID, userID).Scan(&got); err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("user %d document_duty=%q want %q", userID, got, want) } } diff --git a/scripts/run-isolated-duty-sync-pg-tests.sh b/scripts/run-isolated-duty-sync-pg-tests.sh new file mode 100755 index 00000000..0061449b --- /dev/null +++ b/scripts/run-isolated-duty-sync-pg-tests.sh @@ -0,0 +1,78 @@ +#!/usr/bin/env bash +# 隔离库跑 document_duty 相关 PostgreSQL 集成测试。 +# 不启停/重建 compose，不连 agent_system_dev / 共享对话库。 +# +# 用法 A（已有隔离库）： +# TEST_DATABASE_URL='postgres://app:pass@127.0.0.1:55432/agent_duty_sync_itest?sslmode=disable' \ +# ./scripts/run-isolated-duty-sync-pg-tests.sh +# +# 用法 B（Codex/本机用管理连接建临时库，测完删除）： +# PG_ADMIN_URL='postgres://postgres:pass@127.0.0.1:55432/postgres?sslmode=disable' \ +# ./scripts/run-isolated-duty-sync-pg-tests.sh +# +# 禁止：把 TEST_DATABASE_URL 指到 agent_system_dev、5433 默认开发库。 +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "${ROOT_DIR}" + +refuse_shared_dsn() { + local dsn="$1" + if [[ "${dsn}" == *"agent_system_dev"* || "${dsn}" == *"/agent_system_test"* || "${dsn}" == *"/agent_system_main"* ]]; then + echo "[错误] 拒绝共享环境库名（agent_system_dev/test/main）。请 CREATE DATABASE 临时库。" >&2 + exit 2 + fi +} + +rewrite_db_name() { + local src="$1" + local dbname="$2" + local prefix query + if [[ "${src}" == *"?"* ]]; then + prefix="${src%%\?*}" + query="${src#*\?}" + echo "${prefix%/*}/${dbname}?${query}" + else + echo "${src%/*}/${dbname}" + fi +} + +CREATED_DB="" +ADMIN_URL="${PG_ADMIN_URL:-}" +TEST_URL="${TEST_DATABASE_URL:-}" + +if [[ -n "${ADMIN_URL}" ]]; then + refuse_shared_dsn "${ADMIN_URL}" + if ! command -v psql >/dev/null 2>&1; then + echo "[错误] 需要 psql 才能用 PG_ADMIN_URL 建临时库" >&2 + exit 1 + fi + CREATED_DB="agent_duty_sync_itest_$$" + echo "[info] creating isolated database ${CREATED_DB}" + psql "${ADMIN_URL}" -v ON_ERROR_STOP=1 -c "CREATE DATABASE ${CREATED_DB};" + cleanup() { + if [[ -n "${CREATED_DB}" ]]; then + echo "[info] dropping isolated database ${CREATED_DB}" + psql "${ADMIN_URL}" -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS ${CREATED_DB};" || true + fi + } + trap cleanup EXIT + TEST_URL="$(rewrite_db_name "${ADMIN_URL}" "${CREATED_DB}")" +elif [[ -n "${TEST_URL}" ]]; then + refuse_shared_dsn "${TEST_URL}" +else + echo "[错误] 请设置 TEST_DATABASE_URL（隔离库）或 PG_ADMIN_URL（自动建临时库）" >&2 + echo "示例：" >&2 + echo " createdb agent_duty_sync_itest" >&2 + echo " TEST_DATABASE_URL='postgres://app:pass@127.0.0.1:55432/agent_duty_sync_itest?sslmode=disable' \\" >&2 + echo " go test ./internal/docduty ./internal/authz ./internal/server -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|ConcurrentLastWriter|OwnerAuthor'" >&2 + exit 1 +fi + +export TEST_DATABASE_URL="${TEST_URL}" +echo "[info] running isolated duty-sync PostgreSQL tests" + +# 包串行，避免并行 goose 抢同一临时库。 +go test ./internal/docduty -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|OwnerAuthor|HealBinding' +go test ./internal/authz -count=1 -run 'ConcurrentLastWriter|DocumentDuty' +go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole'