From ce53c54fde40a2ee87c90fc6624019590073b293 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Thu, 7 May 2026 18:18:08 +0800 Subject: [PATCH] test(acp): integration matrix scaffold + 3 implemented scenarios (CreateSuccess/Quota/BranchConflict) --- internal/app/acp_integration_test.go | 498 +++++++++++++++++++++++++++ 1 file changed, 498 insertions(+) create mode 100644 internal/app/acp_integration_test.go diff --git a/internal/app/acp_integration_test.go b/internal/app/acp_integration_test.go new file mode 100644 index 0000000..1194807 --- /dev/null +++ b/internal/app/acp_integration_test.go @@ -0,0 +1,498 @@ +//go:build integration + +// acp_integration_test.go — through-stack integration matrix for the ACP module +// (spec §11.3, 14 scenarios). +// +// SCOPE: This file lays out the matrix of 14 scenarios as a scaffold + sub-tests. +// Most scenarios are already covered at lower test layers; through-stack coverage +// here ensures the wiring (HTTP → SessionService → Supervisor → Relay → fake-agent) +// composes correctly under real testcontainers/Postgres + a real fake-agent +// subprocess + a real chi router. +// +// Coverage map (lower-layer test ↔ this file): +// +// Scenario Lower-layer coverage This file +// ----------------------------- -------------------------------- ----------------- +// CreateSessionSuccess E3 TestSupervisor_HappyPath ✓ scaffold + test +// WSStreamLiveEvents G4 TestHandler_Session_E2E deferred (G4 ≈) +// PromptAndCancel D8 TestRelay_Call_RoundTrip* deferred +// SpawnFailedBadBinary E3 TestSupervisor_HangAgent deferred +// HandshakeTimeout E3 TestSupervisor_HangAgent deferred +// ChildCrashed E3 TestSupervisor_AgentCrashes deferred +// StartupReaper D1 ResetStuckSessionsOnRestart deferred +// WSReconnectWithSince D8 TestRelay_FanoutToSubscribers deferred +// FsOutsideWorktreeRejected D4 fs/read_text_file unit deferred +// PerUserQuotaExceeded F3 SessionService.Create unit ✓ scaffold + test +// BranchConflict F3 SessionService.Create unit ✓ scaffold + test +// MainWorktreeMutex D1 AcquireMainWorktree unit deferred +// DeleteAgentKindInUse A11 repo TestPgRepo_DeleteInUse deferred +// DisableAgentKindOK A* AgentKindService unit deferred +// +// DEFERRED scenarios remain as t.Skip stubs with the lower-layer test referenced. +// Future work: fill them in using the same suite + helpers below. +// +// Build/run: +// go test -count=1 -tags=integration ./internal/app/... -run TestACPIntegration + +package app_test + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "testing" + "time" + + "github.com/go-chi/chi/v5" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + tcpg "github.com/testcontainers/testcontainers-go/modules/postgres" + + "github.com/yan1h/agent-coding-workflow/internal/acp" + "github.com/yan1h/agent-coding-workflow/internal/acp/handlers" + "github.com/yan1h/agent-coding-workflow/internal/audit" + "github.com/yan1h/agent-coding-workflow/internal/infra/crypto" + "github.com/yan1h/agent-coding-workflow/internal/infra/db" + "github.com/yan1h/agent-coding-workflow/internal/infra/logger" + "github.com/yan1h/agent-coding-workflow/internal/infra/notify" + "github.com/yan1h/agent-coding-workflow/internal/project" + "github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware" + "github.com/yan1h/agent-coding-workflow/internal/user" + "github.com/yan1h/agent-coding-workflow/internal/workspace" +) + +func TestACPIntegration_Matrix(t *testing.T) { + if testing.Short() { + t.Skip("integration; needs docker + go build") + } + suite := newACPIntegrationSuite(t) + t.Cleanup(suite.cleanup) + + // 14 scenarios per spec §11.3. Implemented = full through-stack assertion. + // Deferred = covered at lower test layers; sub-test t.Skip's with the reference. + t.Run("CreateSessionSuccess", suite.testCreateSessionSuccess) + t.Run("WSStreamLiveEvents", deferredAt("G4 TestHandler_Session_CreateAndWS_E2E")) + t.Run("PromptAndCancel", deferredAt("D8 TestRelay_*")) + t.Run("SpawnFailedBadBinary", deferredAt("E3 TestSupervisor_HangAgent")) + t.Run("HandshakeTimeout", deferredAt("E3 TestSupervisor_HangAgent_HandshakeTimeout")) + t.Run("ChildCrashed", deferredAt("E3 TestSupervisor_AgentCrashes")) + t.Run("StartupReaper", deferredAt("D1 unit + F4 reaper wiring inspection")) + t.Run("WSReconnectWithSince", deferredAt("D8 TestRelay_FanoutToSubscribers")) + t.Run("FsOutsideWorktreeRejected", deferredAt("D4 fs/read_text_file unit")) + t.Run("PerUserQuotaExceeded", suite.testPerUserQuotaExceeded) + t.Run("BranchConflict", suite.testBranchConflict) + t.Run("MainWorktreeMutex", deferredAt("D1 AcquireMainWorktree unit")) + t.Run("DeleteAgentKindInUse", deferredAt("A11 TestPgRepo_AgentKind_DeleteInUse")) + t.Run("DisableAgentKindOK", deferredAt("A* AgentKindService unit")) +} + +// deferredAt returns a sub-test that t.Skip's with a pointer to the lower-layer +// test that already covers the scenario. +func deferredAt(ref string) func(t *testing.T) { + return func(t *testing.T) { + t.Skipf("covered at lower layer: %s", ref) + } +} + +// acpIntegrationSuite holds shared state across the matrix sub-tests. +type acpIntegrationSuite struct { + srv *httptest.Server + pool *pgxpool.Pool + adminToken string + userToken string + wsID uuid.UUID + fakeBin string + cleanup func() +} + +func newACPIntegrationSuite(t *testing.T) *acpIntegrationSuite { + t.Helper() + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not available") + } + + ctx := context.Background() + + container, err := tcpg.Run(ctx, "postgres:16-alpine", + tcpg.WithDatabase("acw"), + tcpg.WithUsername("acw"), + tcpg.WithPassword("acw"), + ) + require.NoError(t, err) + t.Cleanup(func() { _ = container.Terminate(ctx) }) + + dsn, err := container.ConnectionString(ctx, "sslmode=disable") + require.NoError(t, err) + require.NoError(t, db.Migrate(ctx, dsn, "file://../../migrations")) + + pool, err := db.NewPool(ctx, dsn) + require.NoError(t, err) + t.Cleanup(pool.Close) + + log := logger.New(logger.Options{Format: logger.FormatJSON, Level: logger.LevelInfo}) + + auditRec := audit.NewPostgresRecorder(pool) + encKey := make([]byte, 32) + for i := range encKey { + encKey[i] = byte(i + 1) + } + enc, err := crypto.NewEncryptor(encKey) + require.NoError(t, err) + + userRepo := user.NewPostgresRepository(pool) + userSvc := user.NewService(userRepo, auditRec) + + notifyRepo := notify.NewPostgresRepository(pool) + disp := notify.NewDispatcher(notifyRepo, log) + + adminUser, err := userSvc.Bootstrap(ctx, "admin@local", "password123") + require.NoError(t, err) + require.NotNil(t, adminUser) + + // Build the ACP supervisor + service against this DB. + acpRepo := acp.NewPostgresRepository(pool) + + akSvc := acp.NewAgentKindService(acpRepo, enc, auditRec) + + supCfg := acp.SupervisorConfig{ + SpawnTimeout: 5 * time.Second, + KillGrace: 1 * time.Second, + StderrBufferLines: 50, + StderrTailBytes: 1000, + EventMaxPayload: 65536, + EventTruncateField: 4096, + ShutdownGrace: 5 * time.Second, + } + sup := acp.NewSupervisor(acpRepo, auditRec, disp, nil, nil, enc, supCfg, log) + + // Minimal stubs for SessionService dependencies — sufficient for branch=main + // path (no wtSvc / wsRepo calls) and tests that exercise quota/conflict logic. + wsStub := &acpStubWsSvc{} + paStub := &acpStubProjectAccess{} + + sessSvc := acp.NewSessionService( + acpRepo, wsStub, nil, nil, paStub, sup, auditRec, + acp.SessionServiceConfig{MaxActiveGlobal: 20, MaxActivePerUser: 3}, + ) + + r := chi.NewRouter() + r.Use(middleware.RequestID) + user.NewHandler(userSvc).Mount(r) + acp.NewHandler( + akSvc, sessSvc, sup, acpRepo, + userSvc, + acpUserAdminAdapter{svc: userSvc}, + enc, + acp.WSConfig{PingInterval: 30 * time.Second, PongTimeout: 10 * time.Second, SendBuffer: 64}, + ).Mount(r) + + srv := httptest.NewServer(r) + t.Cleanup(srv.Close) + + // Login admin → adminToken + adminToken := login(t, srv.URL, "admin@local", "password123") + + // Provision a regular user via direct SQL + Login (user.Service has no public + // Create method; Bootstrap is idempotent admin-only). Hash matches user.NewService's + // expectation: bcrypt of "password123" produced by user package's hashing helper. + regularUserID := uuid.New() + regularEmail := "regular@local" + // Use Bootstrap on a fresh email to get a properly hashed user, then flip is_admin off. + regularUser, err := userSvc.Bootstrap(ctx, regularEmail, "password123") + require.NoError(t, err) + require.NotNil(t, regularUser) + regularUserID = regularUser.ID + _, err = pool.Exec(ctx, `UPDATE users SET is_admin = false WHERE id = $1`, regularUserID) + require.NoError(t, err) + userToken := login(t, srv.URL, regularEmail, "password123") + + // Project + workspace rows. Direct SQL to avoid needing the full project handler + // here — the ACP path uses workspace.WorkspaceService.Get (stubbed) so the rows + // only need to exist for the ProjectAccess lookups SessionService performs. + pid := uuid.New() + wsID := uuid.New() + cwd := t.TempDir() + _, err = pool.Exec(ctx, `INSERT INTO projects (id, slug, name, owner_id) VALUES ($1, $2, 'p', $3)`, + pid, "p-acp-int", regularUser.ID) + require.NoError(t, err) + _, err = pool.Exec(ctx, `INSERT INTO workspaces + (id, project_id, slug, name, git_remote_url, main_path, default_branch) + VALUES ($1, $2, $3, 'w', 'git@x:y.git', $4, 'main')`, + wsID, pid, "ws-acp-int", cwd) + require.NoError(t, err) + + // Acquire the workspace row for the stub to return. + wsRepoLocal := workspace.NewPostgresRepository(pool) + wsRow, err := wsRepoLocal.GetWorkspaceByID(ctx, wsID) + require.NoError(t, err) + wsStub.ws = wsRow + paStub.pid = pid + + suite := &acpIntegrationSuite{ + srv: srv, + pool: pool, + adminToken: adminToken, + userToken: userToken, + wsID: wsID, + fakeBin: buildFakeAgent(t), + } + suite.cleanup = func() { + sup.ShutdownAll(context.Background()) + } + return suite +} + +// === Implemented sub-tests === + +func (s *acpIntegrationSuite) testCreateSessionSuccess(t *testing.T) { + akID := s.createAgentKind(t, "fake-success", s.fakeBin, nil) + + body, _ := json.Marshal(map[string]any{ + "workspace_id": s.wsID.String(), "agent_kind_id": akID, "branch": "main", + }) + resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, body) + require.Equal(t, http.StatusCreated, resp.StatusCode) + + var dto struct { + ID string `json:"id"` + Status string `json:"status"` + } + require.NoError(t, json.NewDecoder(resp.Body).Decode(&dto)) + _ = resp.Body.Close() + assert.Equal(t, "starting", dto.Status) + + s.waitFor(t, 10*time.Second, func() bool { + return s.getSession(t, dto.ID).Status == "running" + }) +} + +func (s *acpIntegrationSuite) testPerUserQuotaExceeded(t *testing.T) { + akID := s.createAgentKind(t, "fake-quota", s.fakeBin, nil) + + // MaxActivePerUser=3 (set in suite). Create 3 sessions on distinct branches. + for i := 0; i < 3; i++ { + body, _ := json.Marshal(map[string]any{ + "workspace_id": s.wsID.String(), + "agent_kind_id": akID, + "branch": "feat-q-" + strconv.Itoa(i), + }) + resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, body) + require.Equal(t, http.StatusCreated, resp.StatusCode, "creating session %d", i) + _ = resp.Body.Close() + } + + // 4th must 429. + body, _ := json.Marshal(map[string]any{ + "workspace_id": s.wsID.String(), + "agent_kind_id": akID, + "branch": "feat-q-overflow", + }) + resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, body) + defer resp.Body.Close() + assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode) +} + +func (s *acpIntegrationSuite) testBranchConflict(t *testing.T) { + akID := s.createAgentKind(t, "fake-bc", s.fakeBin, nil) + + first, _ := json.Marshal(map[string]any{ + "workspace_id": s.wsID.String(), + "agent_kind_id": akID, + "branch": "main", + }) + resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, first) + require.Equal(t, http.StatusCreated, resp.StatusCode) + _ = resp.Body.Close() + + // Second request on same main branch: main worktree mutex blocks → 409. + resp2 := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, first) + defer resp2.Body.Close() + assert.Equal(t, http.StatusConflict, resp2.StatusCode) +} + +// === Helper methods === + +func (s *acpIntegrationSuite) do(t *testing.T, method, path, token string, body []byte) *http.Response { + t.Helper() + var rd io.Reader + if body != nil { + rd = bytes.NewReader(body) + } + req, _ := http.NewRequest(method, s.srv.URL+path, rd) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + return resp +} + +func (s *acpIntegrationSuite) createAgentKind(t *testing.T, name, bin string, env map[string]string) string { + t.Helper() + body := map[string]any{ + "name": name, "display_name": name, "binary_path": bin, + "args": []string{}, "env": env, "enabled": true, + } + bjson, _ := json.Marshal(body) + resp := s.do(t, http.MethodPost, "/api/v1/acp/agent-kinds", s.adminToken, bjson) + defer resp.Body.Close() + require.Equal(t, http.StatusCreated, resp.StatusCode) + var dto struct { + ID string `json:"id"` + } + require.NoError(t, json.NewDecoder(resp.Body).Decode(&dto)) + return dto.ID +} + +type acpSessionDTO struct { + ID string `json:"id"` + Status string `json:"status"` +} + +func (s *acpIntegrationSuite) getSession(t *testing.T, sid string) acpSessionDTO { + t.Helper() + resp := s.do(t, http.MethodGet, "/api/v1/acp/sessions/"+sid, s.userToken, nil) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + var dto acpSessionDTO + require.NoError(t, json.NewDecoder(resp.Body).Decode(&dto)) + return dto +} + +func (s *acpIntegrationSuite) waitFor(t *testing.T, d time.Duration, pred func() bool) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if pred() { + return + } + time.Sleep(100 * time.Millisecond) + } + t.Fatal("waitFor: timeout") +} + +// === Stubs / adapters === + +type acpUserAdminAdapter struct{ svc user.Service } + +func (a acpUserAdminAdapter) IsAdmin(ctx context.Context, id uuid.UUID) (bool, error) { + u, err := a.svc.Get(ctx, id) + if err != nil { + return false, err + } + return u.IsAdmin, nil +} + +// acpStubWsSvc is the minimal workspace.WorkspaceService stub needed for the +// branch=main happy path (only Get is called). +type acpStubWsSvc struct { + ws *workspace.Workspace +} + +func (s *acpStubWsSvc) Get(_ context.Context, _ workspace.Caller, _ uuid.UUID) (*workspace.Workspace, error) { + return s.ws, nil +} + +func (s *acpStubWsSvc) Create(context.Context, workspace.Caller, string, workspace.CreateWorkspaceInput) (*workspace.Workspace, error) { + return nil, fmt.Errorf("acpStubWsSvc.Create unused") +} + +func (s *acpStubWsSvc) List(context.Context, workspace.Caller, string) ([]*workspace.Workspace, error) { + return nil, fmt.Errorf("acpStubWsSvc.List unused") +} + +func (s *acpStubWsSvc) Update(context.Context, workspace.Caller, uuid.UUID, workspace.UpdateWorkspaceInput) (*workspace.Workspace, error) { + return nil, fmt.Errorf("acpStubWsSvc.Update unused") +} + +func (s *acpStubWsSvc) Delete(context.Context, workspace.Caller, uuid.UUID) error { + return fmt.Errorf("acpStubWsSvc.Delete unused") +} + +func (s *acpStubWsSvc) Sync(context.Context, workspace.Caller, uuid.UUID) (*workspace.Workspace, error) { + return nil, fmt.Errorf("acpStubWsSvc.Sync unused") +} + +func (s *acpStubWsSvc) UpsertCredential(context.Context, workspace.Caller, uuid.UUID, workspace.UpsertCredentialInput) (*workspace.Credential, error) { + return nil, fmt.Errorf("acpStubWsSvc.UpsertCredential unused") +} + +func (s *acpStubWsSvc) GetCredentialMeta(context.Context, workspace.Caller, uuid.UUID) (*workspace.Credential, error) { + return nil, fmt.Errorf("acpStubWsSvc.GetCredentialMeta unused") +} + +// acpStubProjectAccess satisfies acp.ProjectAccess; only GetProjectByWorkspace is +// exercised by the branch=main path. +type acpStubProjectAccess struct { + pid uuid.UUID +} + +func (s *acpStubProjectAccess) GetProjectByWorkspace(_ context.Context, _ uuid.UUID) (*project.Project, error) { + return &project.Project{ID: s.pid, Slug: "p-acp-int"}, nil +} + +func (s *acpStubProjectAccess) GetIssueByNumber(context.Context, uuid.UUID, int) (*project.Issue, error) { + return nil, fmt.Errorf("acpStubProjectAccess.GetIssueByNumber unused") +} + +func (s *acpStubProjectAccess) GetRequirementByNumber(context.Context, uuid.UUID, int) (*project.Requirement, error) { + return nil, fmt.Errorf("acpStubProjectAccess.GetRequirementByNumber unused") +} + +// === Helpers === + +// login posts /api/v1/auth/login and returns the bearer token. +func login(t *testing.T, baseURL, email, password string) string { + t.Helper() + body, _ := json.Marshal(map[string]string{"email": email, "password": password}) + resp, err := http.Post(baseURL+"/api/v1/auth/login", "application/json", bytes.NewReader(body)) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + var lr struct { + Token string `json:"token"` + } + require.NoError(t, json.NewDecoder(resp.Body).Decode(&lr)) + require.NotEmpty(t, lr.Token) + return lr.Token +} + +// buildFakeAgent compiles cmd/fake-acp-agent into t.TempDir and returns the path. +func buildFakeAgent(t *testing.T) string { + t.Helper() + bin := filepath.Join(t.TempDir(), "fake-acp-agent") + if runtime.GOOS == "windows" { + bin += ".exe" + } + wd, _ := os.Getwd() + for { + if _, err := os.Stat(filepath.Join(wd, "go.mod")); err == nil { + break + } + parent := filepath.Dir(wd) + if parent == wd { + t.Fatal("repo root not found") + } + wd = parent + } + cmd := exec.Command("go", "build", "-o", bin, "./cmd/fake-acp-agent") + cmd.Dir = wd + out, err := cmd.CombinedOutput() + require.NoError(t, err, "build fake-acp-agent: %s", out) + return bin +} + +// suppress unused import warnings while scaffold is partial: +var _ = handlers.NewFsHandler +var _ = strings.HasPrefix