weichenandCursor
0f2057765a
Isolated-library coverage checks approved/candidate/impact envelopes, empty-env rollback, and zero FAV staging rows. Co-authored-by: Cursor <cursoragent@cursor.com>
86 lines
2.6 KiB
Go
86 lines
2.6 KiB
Go
package docflow_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/app/uow"
|
|
"chat-agent-system/internal/artifactvault"
|
|
"chat-agent-system/internal/db"
|
|
"chat-agent-system/internal/events"
|
|
|
|
"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 TestOutboxAndStagingSameTxRollback(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("same-tx outbox/staging is not part of -short")
|
|
}
|
|
dsn := db.SkipIfNoTestDatabaseURL(t)
|
|
if strings.Contains(dsn, "5433") || strings.Contains(dsn, "agent_system_dev") ||
|
|
(strings.Contains(dsn, "5434") && strings.Contains(dsn, "agent_system_test")) ||
|
|
strings.Contains(dsn, "5435") {
|
|
t.Fatal("TEST_DATABASE_URL points at a shared env port; refuse agent_system_dev/test/main")
|
|
}
|
|
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)
|
|
}
|
|
|
|
body := []byte(`{"doc_id":"doc_m2_tx","version_no":1}`)
|
|
env := events.Envelope{
|
|
EventID: "document:doc_m2_tx:v1", EventName: "document.agent_version_committed",
|
|
SchemaVersion: events.EnvelopeSchema, Environment: "dev", ProjectID: 9,
|
|
AggregateType: "document", AggregateID: "doc_m2_tx", AggregateVersion: 1,
|
|
ActorRef: "user:1", IdempotencyKey: "commit:doc_m2_tx:1", PayloadDigest: events.Digest(body),
|
|
Producer: "docflow", OccurredAt: time.Now().UTC(), Payload: body,
|
|
}
|
|
boom := errors.New("boom after staging")
|
|
unit := uow.NewPG(pool)
|
|
err = unit.Within(ctx, func(scope uow.TxScope) error {
|
|
ctx = uow.WithScope(ctx, scope)
|
|
if err := (events.PGSink{}).Append(ctx, env, []string{"projection", "notify"}); err != nil {
|
|
return err
|
|
}
|
|
ing := &artifactvault.Ingest{Env: "dev"}
|
|
if err := ing.StageAgentCommit(ctx, 9, "doc_m2_tx", env.EventID, 4, body); err != nil {
|
|
return err
|
|
}
|
|
return boom
|
|
})
|
|
if !errors.Is(err, boom) {
|
|
t.Fatalf("want boom, got %v", err)
|
|
}
|
|
var envelopes, artifacts int
|
|
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_envelopes WHERE event_id=$1`, env.EventID).Scan(&envelopes); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM final_artifacts WHERE source_event_id=$1`, env.EventID).Scan(&artifacts); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if envelopes != 0 || artifacts != 0 {
|
|
t.Fatalf("rollback leaked envelopes=%d artifacts=%d", envelopes, artifacts)
|
|
}
|
|
}
|