test(docadvance): prove workflow advance shadow, uniqueness, and rollback
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
@@ -0,0 +1,260 @@
|
||||
package docadvance_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/app/docadvance"
|
||||
"chat-agent-system/internal/app/uow"
|
||||
"chat-agent-system/internal/db"
|
||||
"chat-agent-system/internal/docgov"
|
||||
"chat-agent-system/internal/events"
|
||||
"chat-agent-system/internal/notification"
|
||||
"chat-agent-system/internal/projectinstance"
|
||||
"chat-agent-system/internal/projectworkflow"
|
||||
"chat-agent-system/internal/workflowdef"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func init() {
|
||||
if _, err := os.Stat("deploy/postgres/goose"); err == nil {
|
||||
return
|
||||
}
|
||||
for _, cand := range []string{"../../..", "../../../.."} {
|
||||
if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
|
||||
_ = os.Chdir(cand)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestE1B_AdvanceShadowLiveRollback_PG(t *testing.T) {
|
||||
pool, ctx := e1bPool(t)
|
||||
fix := seedAdvanceFixture(t, ctx, pool, "one")
|
||||
|
||||
env := approvedEnvelope(fix.suffix+"-a", fix.docID, fix.projectID, 1)
|
||||
appendEnv(t, ctx, pool, env, []string{events.ConsumerWorkflowAdvance})
|
||||
|
||||
d := newAdvanceDispatcher(t, pool, "e1b-w")
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerWorkflowAdvance, events.ModeShadow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); err != nil || n < 1 {
|
||||
t.Fatalf("shadow n=%d err=%v", n, err)
|
||||
}
|
||||
var reqs int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM stage_advance_requests WHERE source_event_id=$1`, env.EventID).Scan(&reqs); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if reqs != 0 {
|
||||
t.Fatalf("shadow wrote stage_advance_requests=%d", reqs)
|
||||
}
|
||||
var recon int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, env.EventID).Scan(&recon); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if recon < 1 {
|
||||
t.Fatal("missing shadow recon")
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerWorkflowAdvance, events.ModeLive); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 2; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
_, _ = d.RunOnce(ctx, events.ConsumerWorkflowAdvance)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM stage_advance_requests WHERE workflow_instance_id=$1`, fix.instanceID).Scan(&reqs); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if reqs != 1 {
|
||||
t.Fatalf("concurrent/live unique stage_advance_requests=%d", reqs)
|
||||
}
|
||||
var notes int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1 AND type=$2`,
|
||||
env.EventID, notification.TypeStageReadyToAdvance).Scan(¬es); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if notes != 1 {
|
||||
t.Fatalf("visible stage-ready notifications=%d", notes)
|
||||
}
|
||||
|
||||
dup := approvedEnvelope(fix.suffix+"-dup", fix.docID, fix.projectID, 2)
|
||||
appendEnv(t, ctx, pool, dup, []string{events.ConsumerWorkflowAdvance})
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM stage_advance_requests WHERE workflow_instance_id=$1`, fix.instanceID).Scan(&reqs); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if reqs != 1 {
|
||||
t.Fatalf("second approved event created extra request=%d", reqs)
|
||||
}
|
||||
|
||||
stale := approvedEnvelope(fix.suffix+"-stale", "missing-doc-"+fix.suffix, fix.projectID, 3)
|
||||
appendEnv(t, ctx, pool, stale, []string{events.ConsumerWorkflowAdvance})
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var status string
|
||||
if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, stale.EventID, events.ConsumerWorkflowAdvance).Scan(&status); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if status != "quarantine" {
|
||||
t.Fatalf("stale/missing doc status=%s", status)
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerWorkflowAdvance, events.ModeOff); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance)
|
||||
if err != nil || n != 0 {
|
||||
t.Fatalf("rollback claimed=%d err=%v", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
func e1bPool(t *testing.T) (*pgxpool.Pool, context.Context) {
|
||||
t.Helper()
|
||||
if testing.Short() {
|
||||
t.Skip("e1b")
|
||||
}
|
||||
dsn := db.SkipIfNoTestDatabaseURL(t)
|
||||
db.RefuseSharedEnvDSN(t, dsn)
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(pool.Close)
|
||||
if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
return pool, ctx
|
||||
}
|
||||
|
||||
func newAdvanceDispatcher(t *testing.T, pool *pgxpool.Pool, owner string) *events.Dispatcher {
|
||||
t.Helper()
|
||||
gov := docgov.NewGovernanceEngine(pool)
|
||||
wfDef := workflowdef.NewService(pool, nil)
|
||||
pwSvc := projectworkflow.NewService(projectworkflow.NewStore(pool), nil)
|
||||
inst := projectinstance.NewService(projectinstance.NewStore(pool), wfDef.Store(), pwSvc.Store(), projectinstance.NewDocGovChecker(gov))
|
||||
d := events.NewDispatcher(pool, "dev", owner)
|
||||
if err := d.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{
|
||||
Inst: inst, Docs: gov, PW: pwSvc, Notify: notification.NewService(pool), Env: "dev",
|
||||
}).HandleLease); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
type advanceFix struct {
|
||||
suffix, docID string
|
||||
projectID int64
|
||||
instanceID int64
|
||||
}
|
||||
|
||||
func seedAdvanceFixture(t *testing.T, ctx context.Context, pool *pgxpool.Pool, tag string) advanceFix {
|
||||
t.Helper()
|
||||
suffix := fmt.Sprintf("%s-%d", tag, time.Now().UnixNano())
|
||||
var author, reviewer, approver int64
|
||||
if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'A','engineer') RETURNING id`, "e1b-a-"+suffix).Scan(&author); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'R','engineer') RETURNING id`, "e1b-r-"+suffix).Scan(&reviewer); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'P','engineer') RETURNING id`, "e1b-p-"+suffix).Scan(&approver); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := workflowdef.BootstrapSeeds(ctx, pool, nil, "templates/docs", author); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wfStore := workflowdef.NewStore(pool)
|
||||
pwStore := projectworkflow.NewStore(pool)
|
||||
instStore := projectinstance.NewStore(pool)
|
||||
gov := docgov.NewGovernanceEngine(pool)
|
||||
svc := projectinstance.NewService(instStore, wfStore, pwStore, projectinstance.NewDocGovChecker(gov))
|
||||
|
||||
var projectID int64
|
||||
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, 'E1B', 'active', $2, 'planning', '', 1, 0) RETURNING id
|
||||
`, "e1b-"+suffix, author).Scan(&projectID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
plan, err := svc.BuildDefaultPlan(ctx, workflowdef.OfficialKeyMain, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := range plan.Stages {
|
||||
for j := range plan.Stages[i].DocTasks {
|
||||
plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{
|
||||
AuthorUserID: author, ReviewerUserID: reviewer, ApproverUserID: approver,
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := svc.SaveWorkflowPlan(ctx, projectID, author, true, projectinstance.SavePlanInput{
|
||||
Plan: *plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
preview, err := svc.ActivationPreview(ctx, projectID, author, true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
activated, err := svc.Activate(ctx, projectID, author, true, projectinstance.ActivateInput{
|
||||
ConfirmationToken: preview.ConfirmationToken,
|
||||
ExpectedProjectVersion: preview.ExpectedProjectVersion,
|
||||
ExpectedPlanRevision: preview.ExpectedPlanRevision,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
docID := "doc_e1b_intake_" + suffix
|
||||
if _, err := gov.CreateOrGetDocument(ctx, docID, projectID, "intake", "客户需求采集", "V0.1", "", author, "A", reviewer, "R", approver, "P"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = gov.MarkDocumentContentUpdated(ctx, docID, "hash_e1b", false)
|
||||
if _, err := gov.SubmitForReview(ctx, docID, author, "author", "A"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := gov.ReviewApprove(ctx, docID, reviewer, "reviewer", "R", "ok"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := gov.ApproverApprove(ctx, docID, approver, "approver", "P", "ok"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return advanceFix{suffix: suffix, docID: docID, projectID: projectID, instanceID: activated.Instance.ID}
|
||||
}
|
||||
|
||||
func approvedEnvelope(id, docID string, projectID, ver int64) events.Envelope {
|
||||
body := []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d,"task_key":"intake"}`, docID, projectID))
|
||||
return events.Envelope{
|
||||
EventID: "approved-" + id, EventName: events.EventApproved, SchemaVersion: events.EnvelopeSchema,
|
||||
Environment: "dev", ProjectID: projectID, AggregateType: "document_approval", AggregateID: docID,
|
||||
AggregateVersion: ver, ActorRef: "user:4", IdempotencyKey: "approved-" + id,
|
||||
PayloadDigest: events.Digest(body), Producer: "docflow", OccurredAt: time.Now().UTC(), Payload: body,
|
||||
}
|
||||
}
|
||||
|
||||
func appendEnv(t *testing.T, ctx context.Context, pool *pgxpool.Pool, env events.Envelope, consumers []string) {
|
||||
t.Helper()
|
||||
if err := uow.NewPG(pool).Within(ctx, func(scope uow.TxScope) error {
|
||||
return (events.PGSink{}).Append(uow.WithScope(ctx, scope), env, consumers)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user