You've already forked agentic-coding-workflow
test(acp): handler integration e2e (POST session + WS state event + DELETE)
This commit is contained in:
@@ -0,0 +1,270 @@
|
||||
//go:build integration
|
||||
|
||||
// handler_e2e_test.go drives the ACP handler through one full happy-path
|
||||
// session lifecycle:
|
||||
//
|
||||
// 1. POST /api/v1/acp/sessions → 201
|
||||
// 2. wait for supervisor handshake → SessionRunning
|
||||
// 3. WS /api/v1/acp/sessions/{id}/ws → expect "state" event
|
||||
// 4. DELETE /api/v1/acp/sessions/{id} → 204
|
||||
// 5. wait for monitor → SessionExited
|
||||
//
|
||||
// Build-tagged integration: spins up testcontainer PG + go build fake-acp-agent,
|
||||
// so it never runs under -short.
|
||||
package acp_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
ws "github.com/coder/websocket"
|
||||
"github.com/coder/websocket/wsjson"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/acp"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/audit"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/project"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/workspace"
|
||||
)
|
||||
|
||||
// stubWsService satisfies workspace.WorkspaceService for the e2e test.
|
||||
// Only Get is exercised by SessionService.Create; other methods panic.
|
||||
type stubWsService struct {
|
||||
wsRow *workspace.Workspace
|
||||
}
|
||||
|
||||
func (s *stubWsService) Get(_ context.Context, _ workspace.Caller, _ uuid.UUID) (*workspace.Workspace, error) {
|
||||
return s.wsRow, nil
|
||||
}
|
||||
func (s *stubWsService) Create(context.Context, workspace.Caller, string, workspace.CreateWorkspaceInput) (*workspace.Workspace, error) {
|
||||
panic("stubWsService.Create unused")
|
||||
}
|
||||
func (s *stubWsService) List(context.Context, workspace.Caller, string) ([]*workspace.Workspace, error) {
|
||||
panic("stubWsService.List unused")
|
||||
}
|
||||
func (s *stubWsService) Update(context.Context, workspace.Caller, uuid.UUID, workspace.UpdateWorkspaceInput) (*workspace.Workspace, error) {
|
||||
panic("stubWsService.Update unused")
|
||||
}
|
||||
func (s *stubWsService) Delete(context.Context, workspace.Caller, uuid.UUID) error {
|
||||
panic("stubWsService.Delete unused")
|
||||
}
|
||||
func (s *stubWsService) Sync(context.Context, workspace.Caller, uuid.UUID) (*workspace.Workspace, error) {
|
||||
panic("stubWsService.Sync unused")
|
||||
}
|
||||
func (s *stubWsService) UpsertCredential(context.Context, workspace.Caller, uuid.UUID, workspace.UpsertCredentialInput) (*workspace.Credential, error) {
|
||||
panic("stubWsService.UpsertCredential unused")
|
||||
}
|
||||
func (s *stubWsService) GetCredentialMeta(context.Context, workspace.Caller, uuid.UUID) (*workspace.Credential, error) {
|
||||
panic("stubWsService.GetCredentialMeta unused")
|
||||
}
|
||||
|
||||
// stubProjectAccess satisfies acp.ProjectAccess for the e2e test.
|
||||
// Only GetProjectByWorkspace is exercised; the rest panic.
|
||||
type stubProjectAccess struct {
|
||||
pj *project.Project
|
||||
}
|
||||
|
||||
func (s *stubProjectAccess) GetProjectByWorkspace(_ context.Context, _ uuid.UUID) (*project.Project, error) {
|
||||
return s.pj, nil
|
||||
}
|
||||
func (s *stubProjectAccess) GetIssueByNumber(context.Context, uuid.UUID, int) (*project.Issue, error) {
|
||||
panic("stubProjectAccess.GetIssueByNumber unused")
|
||||
}
|
||||
func (s *stubProjectAccess) GetRequirementByNumber(context.Context, uuid.UUID, int) (*project.Requirement, error) {
|
||||
panic("stubProjectAccess.GetRequirementByNumber unused")
|
||||
}
|
||||
|
||||
// TestHandler_Session_CreateAndWS_E2E covers the happy path from POST /sessions
|
||||
// through WS state event to DELETE — the full request → subprocess → relay →
|
||||
// shutdown loop, exercising Handler + SessionService + Supervisor end-to-end.
|
||||
func TestHandler_Session_CreateAndWS_E2E(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("e2e: spawns subprocess + testcontainer pg")
|
||||
}
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
|
||||
repo, pool := setupRepo(t)
|
||||
|
||||
uid := mustInsertUser(t, ctx, pool, "wsuser-e2e@local", false)
|
||||
pid, wsid := mustInsertProjectWorkspace(t, ctx, pool, uid)
|
||||
|
||||
// mustInsertProjectWorkspace hardcodes main_path='/tmp'. Override to a real
|
||||
// per-test directory so subprocess Spawn can chdir into it.
|
||||
cwd := t.TempDir()
|
||||
_, err := pool.Exec(ctx,
|
||||
`UPDATE workspaces SET main_path = $1 WHERE id = $2`, cwd, wsid)
|
||||
require.NoError(t, err)
|
||||
|
||||
// migrations/0003 sets default_branch DEFAULT 'main', so workspaces row
|
||||
// inserted by the helper already has DefaultBranch='main' — no UPDATE needed.
|
||||
|
||||
bin := fakeAgentBinary(t)
|
||||
enc := testEncryptor(t)
|
||||
|
||||
kind, err := repo.CreateAgentKind(ctx, &acp.AgentKind{
|
||||
ID: uuid.New(), Name: "ws-e2e-fake", DisplayName: "Fake",
|
||||
BinaryPath: bin, Args: []string{}, Enabled: true, CreatedBy: uid,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Real workspace row → seeded into the wsService stub so SessionService.Get
|
||||
// returns the same ProjectID + DefaultBranch the DB has.
|
||||
wsRow, err := workspace.NewPostgresRepository(pool).GetWorkspaceByID(ctx, wsid)
|
||||
require.NoError(t, err)
|
||||
|
||||
disp := notify.NewDispatcher(notifyRepoStub{}, slogDiscard())
|
||||
sup := acp.NewSupervisor(
|
||||
repo,
|
||||
audit.NewPostgresRecorder(pool),
|
||||
disp,
|
||||
nil, nil, // wtSvc/wsRepo: IsMainWorktree=true path → not called
|
||||
enc,
|
||||
acp.SupervisorConfig{
|
||||
SpawnTimeout: 5 * time.Second,
|
||||
KillGrace: 1 * time.Second,
|
||||
StderrBufferLines: 50,
|
||||
StderrTailBytes: 1000,
|
||||
EventMaxPayload: 65536,
|
||||
EventTruncateField: 4096,
|
||||
ShutdownGrace: 5 * time.Second,
|
||||
},
|
||||
slogDiscard(),
|
||||
)
|
||||
defer sup.ShutdownAll(context.Background())
|
||||
|
||||
wsStub := &stubWsService{wsRow: wsRow}
|
||||
paStub := &stubProjectAccess{pj: &project.Project{ID: pid, Slug: "p-test"}}
|
||||
|
||||
svc := acp.NewSessionService(
|
||||
repo, wsStub, nil, nil, paStub, sup,
|
||||
audit.NewPostgresRecorder(pool),
|
||||
acp.SessionServiceConfig{MaxActiveGlobal: 10, MaxActivePerUser: 5},
|
||||
)
|
||||
|
||||
h := acp.NewHandler(
|
||||
nil, // ak svc — not exercised by sessions endpoints
|
||||
svc, sup, repo,
|
||||
fakeResolver{uid: uid},
|
||||
fakeAdminLookup{admin: map[uuid.UUID]bool{uid: false}},
|
||||
enc,
|
||||
acp.WSConfig{
|
||||
PingInterval: 30 * time.Second,
|
||||
PongTimeout: 10 * time.Second,
|
||||
SendBuffer: 64,
|
||||
},
|
||||
)
|
||||
|
||||
r := chi.NewRouter()
|
||||
h.Mount(r)
|
||||
srv := httptest.NewServer(r)
|
||||
defer srv.Close()
|
||||
|
||||
// 1. POST /api/v1/acp/sessions
|
||||
branch := "main"
|
||||
body, err := json.Marshal(map[string]any{
|
||||
"workspace_id": wsid.String(),
|
||||
"agent_kind_id": kind.ID.String(),
|
||||
"branch": branch,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
req, err := http.NewRequest("POST", srv.URL+"/api/v1/acp/sessions", bytes.NewReader(body))
|
||||
require.NoError(t, err)
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer valid")
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
|
||||
var created map[string]any
|
||||
require.NoError(t, json.NewDecoder(resp.Body).Decode(&created))
|
||||
require.Equal(t, http.StatusCreated, resp.StatusCode, "body=%v", created)
|
||||
|
||||
sidStr, _ := created["id"].(string)
|
||||
require.NotEmpty(t, sidStr)
|
||||
sid := uuid.MustParse(sidStr)
|
||||
|
||||
// 2. Wait for handshake → SessionRunning (supervisor.Spawn runs in a goroutine
|
||||
// inside SessionService.Create, so the row starts as 'starting'). 10s budget
|
||||
// covers fake-agent stdin/stdout handshake on slow CI.
|
||||
deadline := time.After(10 * time.Second)
|
||||
for {
|
||||
s, _ := repo.GetSessionByID(ctx, sid)
|
||||
if s != nil && s.Status == acp.SessionRunning {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatalf("session never reached SessionRunning")
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
// 3. WS connect — Authorization header is honored by the Auth middleware.
|
||||
wsURL := strings.Replace(srv.URL, "http", "ws", 1) +
|
||||
"/api/v1/acp/sessions/" + sid.String() + "/ws"
|
||||
wsCtx, wsCancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer wsCancel()
|
||||
|
||||
wsConn, _, err := ws.Dial(wsCtx, wsURL, &ws.DialOptions{
|
||||
HTTPHeader: http.Header{"Authorization": []string{"Bearer valid"}},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
defer wsConn.Close(ws.StatusInternalError, "")
|
||||
|
||||
var stateEv map[string]any
|
||||
require.NoError(t, wsjson.Read(wsCtx, wsConn, &stateEv))
|
||||
assert.Equal(t, "state", stateEv["kind"])
|
||||
assert.Equal(t, sid.String(), stateEv["session_id"])
|
||||
assert.Equal(t, "running", stateEv["status"])
|
||||
assert.Equal(t, "main", stateEv["branch"])
|
||||
|
||||
// Close WS before DELETE so the subscriber drains and Kill doesn't race
|
||||
// the WS goroutine.
|
||||
_ = wsConn.Close(ws.StatusNormalClosure, "test done")
|
||||
|
||||
// 4. DELETE /api/v1/acp/sessions/{id}
|
||||
delReq, err := http.NewRequest("DELETE",
|
||||
srv.URL+"/api/v1/acp/sessions/"+sid.String(), nil)
|
||||
require.NoError(t, err)
|
||||
delReq.Header.Set("Authorization", "Bearer valid")
|
||||
|
||||
delResp, err := http.DefaultClient.Do(delReq)
|
||||
require.NoError(t, err)
|
||||
defer delResp.Body.Close()
|
||||
assert.Equal(t, http.StatusNoContent, delResp.StatusCode)
|
||||
|
||||
// 5. Wait for terminal status. Kill blocks on monitor.done internally,
|
||||
// so by the time DELETE returns the row should already be terminal —
|
||||
// poll briefly to guard against scheduler latency.
|
||||
deadline2 := time.After(5 * time.Second)
|
||||
for {
|
||||
s, _ := repo.GetSessionByID(ctx, sid)
|
||||
if s != nil && (s.Status == acp.SessionExited || s.Status == acp.SessionCrashed) {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-deadline2:
|
||||
s, _ := repo.GetSessionByID(ctx, sid)
|
||||
lastStatus := acp.SessionStatus("?")
|
||||
if s != nil {
|
||||
lastStatus = s.Status
|
||||
}
|
||||
t.Fatalf("session not terminated, last status=%s", lastStatus)
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user