Resolve conflicts: both CURRENT claim tables (slot preserved), local multi-host coordination plan + origin phased companion, origin module-boundaries, combined PG isolation with remote product tests. FEATURE-MATRIX takes origin record_id layout; A-G done bump deferred pending evidence row apply.
224 lines
6.9 KiB
Go
224 lines
6.9 KiB
Go
package server_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"chat-agent-system/internal/auth"
|
|
"chat-agent-system/internal/config"
|
|
"chat-agent-system/internal/db"
|
|
"chat-agent-system/internal/workflowdef"
|
|
)
|
|
|
|
func loginTestAdmin(t *testing.T, handler http.Handler, cfg *config.Config) string {
|
|
t.Helper()
|
|
username := cfg.AdminUsername
|
|
password := cfg.AdminPassword
|
|
if username == "" {
|
|
username = "admin"
|
|
}
|
|
if password == "" {
|
|
password = "admin-secret-password-123"
|
|
}
|
|
body, _ := json.Marshal(map[string]string{
|
|
"username": username,
|
|
"password": password,
|
|
})
|
|
req := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", bytes.NewReader(body))
|
|
w := httptest.NewRecorder()
|
|
handler.ServeHTTP(w, req)
|
|
if w.Code != http.StatusOK {
|
|
t.Fatalf("admin login failed: %d %s", w.Code, w.Body.String())
|
|
}
|
|
var session auth.SessionInfo
|
|
if err := json.NewDecoder(w.Body).Decode(&session); err != nil {
|
|
t.Fatalf("decode login session: %v", err)
|
|
}
|
|
if session.Token == "" {
|
|
t.Fatal("expected non-empty session token")
|
|
}
|
|
return session.Token
|
|
}
|
|
|
|
func withBearerAuth(req *http.Request, token string) {
|
|
req.Header.Set("Authorization", "Bearer "+token)
|
|
}
|
|
|
|
func loginTestUser(t *testing.T, handler http.Handler, username, password string) string {
|
|
t.Helper()
|
|
body, _ := json.Marshal(map[string]string{
|
|
"username": username,
|
|
"password": password,
|
|
})
|
|
req := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", bytes.NewReader(body))
|
|
w := httptest.NewRecorder()
|
|
handler.ServeHTTP(w, req)
|
|
if w.Code != http.StatusOK {
|
|
t.Fatalf("login %s failed: %d %s", username, w.Code, w.Body.String())
|
|
}
|
|
var session auth.SessionInfo
|
|
if err := json.NewDecoder(w.Body).Decode(&session); err != nil {
|
|
t.Fatalf("decode login session: %v", err)
|
|
}
|
|
return session.Token
|
|
}
|
|
|
|
func withUserIDAuth(req *http.Request, token string) {
|
|
req.Header.Set("Authorization", "Bearer "+token)
|
|
}
|
|
|
|
func configFromTestDatabaseURL(t *testing.T) *config.Config {
|
|
t.Helper()
|
|
dsn := db.SkipIfNoTestDatabaseURL(t)
|
|
parsed, err := url.Parse(dsn)
|
|
if err != nil {
|
|
t.Fatalf("parse TEST_DATABASE_URL: %v", err)
|
|
}
|
|
host, portStr, splitErr := net.SplitHostPort(parsed.Host)
|
|
if splitErr != nil {
|
|
host = parsed.Host
|
|
portStr = "5432"
|
|
}
|
|
port, _ := strconv.Atoi(portStr)
|
|
user := parsed.User.Username()
|
|
password, _ := parsed.User.Password()
|
|
dbName := strings.TrimPrefix(parsed.Path, "/")
|
|
return &config.Config{
|
|
Env: "test",
|
|
DataDir: t.TempDir(),
|
|
AgentBackend: "stub",
|
|
AllowInMemory: false,
|
|
ContextDbHost: host,
|
|
ContextDbPort: port,
|
|
ContextDbUser: user,
|
|
ContextDbPassword: password,
|
|
ContextDbName: dbName,
|
|
AdminUsername: envOr("ADMIN_USERNAME", "admin"),
|
|
AdminPassword: envOr("ADMIN_PASSWORD", "admin-secret-password-123"),
|
|
}
|
|
}
|
|
|
|
func envOr(key, fallback string) string {
|
|
if v := strings.TrimSpace(os.Getenv(key)); v != "" {
|
|
return v
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
// resolvePiAgentSharedDataDir picks the host DATA_DIR that the pi-agent runner
|
|
// bind-mounts (compose: ./data/<env>/workspaces → /app/data/workspaces).
|
|
// Prefer DATA_DIR env, then repo-root data/dev|test via go.mod walk — never the
|
|
// package-local internal/server/data/* tree, which the runner cannot see.
|
|
func resolvePiAgentSharedDataDir(cwd, envDataDir string) string {
|
|
if d := strings.TrimSpace(envDataDir); d != "" {
|
|
return d
|
|
}
|
|
dir := filepath.Clean(cwd)
|
|
for i := 0; i < 8; i++ {
|
|
if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
|
|
for _, envName := range []string{"dev", "test"} {
|
|
candidate := filepath.Join(dir, "data", envName)
|
|
if st, err := os.Stat(filepath.Join(candidate, "workspaces")); err == nil && st.IsDir() {
|
|
return candidate
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
parent := filepath.Dir(dir)
|
|
if parent == dir {
|
|
break
|
|
}
|
|
dir = parent
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func piAgentSharedDataDir(t *testing.T) string {
|
|
t.Helper()
|
|
cwd, err := os.Getwd()
|
|
if err != nil {
|
|
t.Fatalf("getwd: %v", err)
|
|
}
|
|
if d := resolvePiAgentSharedDataDir(cwd, os.Getenv("DATA_DIR")); d != "" {
|
|
return d
|
|
}
|
|
t.Skip("shared pi-agent DATA_DIR not found (mount data/test/workspaces or data/dev/workspaces)")
|
|
return ""
|
|
}
|
|
|
|
func seedOfficialWorkflowManifest(t *testing.T, pool *pgxpool.Pool, workflowKey string) {
|
|
t.Helper()
|
|
filename, ok := workflowdef.V2ManifestFiles[workflowKey]
|
|
if !ok {
|
|
t.Fatalf("unknown official workflow key %q", workflowKey)
|
|
}
|
|
def, err := workflowdef.LoadManifestDefinition(filename)
|
|
if err != nil {
|
|
t.Fatalf("load manifest %s: %v", filename, err)
|
|
}
|
|
wfStore := workflowdef.NewStore(pool)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
if err := wfStore.UpsertPublishedSeed(ctx, workflowdef.CreateDefinitionInput{
|
|
Key: workflowKey,
|
|
Name: workflowdef.OfficialWorkflowName(workflowKey),
|
|
Definition: def,
|
|
}, 1); err != nil {
|
|
t.Fatalf("seed official workflow %s: %v", workflowKey, err)
|
|
}
|
|
}
|
|
|
|
// seedProjectForRun 创建带 active 成员的最小项目,返回 projectID;用于要求
|
|
// 项目作用域的 taskrun/事件链路测试(events.ValidateEnvelope 拒绝 project<=0)。
|
|
func seedProjectForRun(t *testing.T, pool *pgxpool.Pool, prefix string, ownerUserID int64) int64 {
|
|
t.Helper()
|
|
code := prefix + "-" + time.Now().Format("150405.000000")
|
|
var projectID int64
|
|
if err := pool.QueryRow(context.Background(), `
|
|
INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status)
|
|
VALUES ($1, 'Run Scope Project ' || $1, 'active', $2, 'active') RETURNING id
|
|
`, code, ownerUserID).Scan(&projectID); err != nil {
|
|
t.Fatalf("seed project: %v", err)
|
|
}
|
|
ensureProjectMembers(t, pool, projectID, map[int64]string{ownerUserID: "owner"})
|
|
t.Cleanup(func() { cleanupProject(t, pool, projectID) })
|
|
return projectID
|
|
}
|
|
|
|
// insertScopedTestProject creates a project row so task-run finalizers can emit
|
|
// outbox events (Environment + ProjectID are both required by events.Validate).
|
|
func insertScopedTestProject(t *testing.T, pool *pgxpool.Pool, creatorID int64, namePrefix string) int64 {
|
|
t.Helper()
|
|
if creatorID <= 0 {
|
|
creatorID = 1
|
|
}
|
|
code := namePrefix + "-" + time.Now().Format("150405.000000")
|
|
var projectID int64
|
|
err := pool.QueryRow(context.Background(), `
|
|
INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision)
|
|
VALUES ($1, $2, 'active', $3, 'planning', '', 1, 0) RETURNING id
|
|
`, code, namePrefix+" "+code, creatorID).Scan(&projectID)
|
|
if err != nil {
|
|
t.Fatalf("insert scoped project: %v", err)
|
|
}
|
|
_, _ = pool.Exec(context.Background(), `
|
|
INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at)
|
|
VALUES ($1, $2, 'owner', 'active', now(), now()) ON CONFLICT DO NOTHING
|
|
`, projectID, creatorID)
|
|
t.Cleanup(func() { cleanupProject(t, pool, projectID) })
|
|
return projectID
|
|
}
|