Alassane
f5a505c468
Co-authored-by: ayattara <alassane.yattara@savoirfairelinux.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
407 lines
14 KiB
Go
407 lines
14 KiB
Go
package github
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
|
|
"github.com/kandev/kandev/internal/testutil"
|
|
)
|
|
|
|
func TestIsPRWatchUniqueViolation_PostgresConstraint(t *testing.T) {
|
|
err := &pgconn.PgError{Code: "23505", ConstraintName: prWatchSearchingIndexName}
|
|
if !isPRWatchUniqueViolation(errors.Join(errors.New("wrapped"), err)) {
|
|
t.Fatal("expected PostgreSQL PR-watch unique violation to be recognized")
|
|
}
|
|
}
|
|
|
|
func TestIsPRWatchDiscoveredUniqueViolation_PostgresConstraint(t *testing.T) {
|
|
err := &pgconn.PgError{Code: "23505", ConstraintName: prWatchDiscoveredIndexName}
|
|
if !isPRWatchDiscoveredUniqueViolation(errors.Join(errors.New("wrapped"), err)) {
|
|
t.Fatal("expected PostgreSQL PR-watch discovered unique violation to be recognized")
|
|
}
|
|
}
|
|
|
|
func TestStore_UpdatePRWatchBranchIfSearching_PostgresDifferentSessionCollision(t *testing.T) {
|
|
db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t))
|
|
ctx := context.Background()
|
|
if _, err := db.ExecContext(ctx, `
|
|
CREATE TABLE tasks (id TEXT PRIMARY KEY, workspace_id TEXT, archived_at TIMESTAMP);
|
|
CREATE TABLE workspaces (id TEXT PRIMARY KEY);
|
|
INSERT INTO workspaces (id) VALUES ('ws-1');
|
|
INSERT INTO tasks (id, workspace_id) VALUES ('task-1', 'ws-1');`); err != nil {
|
|
t.Fatalf("create task fixtures: %v", err)
|
|
}
|
|
if _, err := db.ExecContext(ctx, `
|
|
CREATE TABLE github_pr_watches (
|
|
id TEXT PRIMARY KEY,
|
|
workspace_id TEXT NOT NULL DEFAULT '',
|
|
session_id TEXT NOT NULL DEFAULT '',
|
|
task_id TEXT NOT NULL,
|
|
repository_id TEXT NOT NULL DEFAULT '',
|
|
owner TEXT NOT NULL,
|
|
repo TEXT NOT NULL,
|
|
pr_number INTEGER NOT NULL,
|
|
branch TEXT NOT NULL,
|
|
last_checked_at TIMESTAMP,
|
|
last_comment_at TIMESTAMP,
|
|
last_check_status TEXT DEFAULT '',
|
|
last_review_state TEXT DEFAULT '',
|
|
created_at TIMESTAMP NOT NULL,
|
|
updated_at TIMESTAMP NOT NULL
|
|
);
|
|
CREATE UNIQUE INDEX idx_github_pr_watches_searching
|
|
ON github_pr_watches (task_id, repository_id, branch) WHERE pr_number = 0;`); err != nil {
|
|
t.Fatalf("create postgres PR-watch fixture: %v", err)
|
|
}
|
|
store := &Store{db: db, ro: db}
|
|
now := time.Now().UTC()
|
|
source := &PRWatch{
|
|
ID: "watch-source-pg", WorkspaceID: "ws-1", SessionID: "session-1", TaskID: "task-1",
|
|
RepositoryID: "repo-1", Owner: "owner", Repo: "repo", Branch: "feature/A",
|
|
CreatedAt: now, UpdatedAt: now,
|
|
}
|
|
sibling := &PRWatch{
|
|
ID: "watch-sibling-pg", WorkspaceID: "ws-1", SessionID: "session-2", TaskID: "task-1",
|
|
RepositoryID: "repo-1", Owner: "owner", Repo: "repo", Branch: "feature/B",
|
|
CreatedAt: now, UpdatedAt: now,
|
|
}
|
|
for _, watch := range []*PRWatch{source, sibling} {
|
|
if err := store.CreatePRWatch(ctx, watch); err != nil {
|
|
t.Fatalf("create watch %s: %v", watch.ID, err)
|
|
}
|
|
}
|
|
|
|
if err := store.UpdatePRWatchBranchIfSearching(ctx, source.ID, sibling.Branch); err != nil {
|
|
t.Fatalf("different-session task-owned collision must coalesce: %v", err)
|
|
}
|
|
watches, err := store.ListPRWatchesByTask(ctx, source.TaskID)
|
|
if err != nil {
|
|
t.Fatalf("list watches: %v", err)
|
|
}
|
|
if len(watches) != 1 || watches[0].ID != sibling.ID {
|
|
t.Fatalf("expected sibling to survive alone, got %+v", watches)
|
|
}
|
|
}
|
|
|
|
func TestStore_UpdatePRWatchPRNumber_PostgresDifferentSessionCollision(t *testing.T) {
|
|
db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t))
|
|
ctx := context.Background()
|
|
if _, err := db.ExecContext(ctx, `
|
|
CREATE TABLE tasks (id TEXT PRIMARY KEY, workspace_id TEXT, archived_at TIMESTAMP);
|
|
CREATE TABLE workspaces (id TEXT PRIMARY KEY);
|
|
INSERT INTO workspaces (id) VALUES ('ws-1');
|
|
INSERT INTO tasks (id, workspace_id) VALUES ('task-1', 'ws-1');`); err != nil {
|
|
t.Fatalf("create task fixtures: %v", err)
|
|
}
|
|
if _, err := db.ExecContext(ctx, `
|
|
CREATE TABLE github_pr_watches (
|
|
id TEXT PRIMARY KEY,
|
|
workspace_id TEXT NOT NULL DEFAULT '',
|
|
session_id TEXT NOT NULL DEFAULT '',
|
|
task_id TEXT NOT NULL,
|
|
repository_id TEXT NOT NULL DEFAULT '',
|
|
owner TEXT NOT NULL,
|
|
repo TEXT NOT NULL,
|
|
pr_number INTEGER NOT NULL,
|
|
branch TEXT NOT NULL,
|
|
last_checked_at TIMESTAMP,
|
|
last_comment_at TIMESTAMP,
|
|
last_check_status TEXT DEFAULT '',
|
|
last_review_state TEXT DEFAULT '',
|
|
created_at TIMESTAMP NOT NULL,
|
|
updated_at TIMESTAMP NOT NULL
|
|
);
|
|
CREATE UNIQUE INDEX idx_github_pr_watches_discovered
|
|
ON github_pr_watches (task_id, repository_id, pr_number) WHERE pr_number <> 0;`); err != nil {
|
|
t.Fatalf("create postgres PR-watch fixture: %v", err)
|
|
}
|
|
store := &Store{db: db, ro: db}
|
|
now := time.Now().UTC()
|
|
watches := []*PRWatch{
|
|
{ID: "watch-source-pg", WorkspaceID: "ws-1", SessionID: "session-1", TaskID: "task-1", RepositoryID: "repo-1", Owner: "owner", Repo: "repo", Branch: "feature/A", CreatedAt: now, UpdatedAt: now},
|
|
{ID: "watch-sibling-pg", WorkspaceID: "ws-1", SessionID: "session-2", TaskID: "task-1", RepositoryID: "repo-1", Owner: "owner", Repo: "repo", Branch: "feature/B", CreatedAt: now, UpdatedAt: now},
|
|
}
|
|
for _, watch := range watches {
|
|
if err := store.CreatePRWatch(ctx, watch); err != nil {
|
|
t.Fatalf("create watch %s: %v", watch.ID, err)
|
|
}
|
|
}
|
|
|
|
if err := store.UpdatePRWatchPRNumber(ctx, watches[1].ID, 99); err != nil {
|
|
t.Fatalf("promote sibling watch: %v", err)
|
|
}
|
|
if err := store.UpdatePRWatchPRNumber(ctx, watches[0].ID, 99); err != nil {
|
|
t.Fatalf("different-session discovered-watch collision must coalesce: %v", err)
|
|
}
|
|
|
|
got, err := store.GetPRWatchByTaskRepoPRNumber(ctx, "task-1", "repo-1", 99)
|
|
if err != nil {
|
|
t.Fatalf("get discovered watch: %v", err)
|
|
}
|
|
if got == nil || got.ID != watches[1].ID {
|
|
t.Fatalf("expected sibling to survive as discovered watch, got %+v", got)
|
|
}
|
|
searching, err := store.GetPRWatchByTaskRepoBranch(ctx, "task-1", "repo-1", "feature/A")
|
|
if err != nil {
|
|
t.Fatalf("get source searching watch: %v", err)
|
|
}
|
|
if searching != nil {
|
|
t.Fatalf("expected source watch to be dropped, got %+v", searching)
|
|
}
|
|
}
|
|
|
|
func TestStore_UpdatePRWatchPRNumber_PostgresUpdateCollisionKeepsTransactionUsable(t *testing.T) {
|
|
db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t))
|
|
ctx := context.Background()
|
|
if _, err := db.ExecContext(ctx, `
|
|
CREATE TABLE tasks (id TEXT PRIMARY KEY, workspace_id TEXT, archived_at TIMESTAMP);
|
|
CREATE TABLE workspaces (id TEXT PRIMARY KEY);
|
|
INSERT INTO workspaces (id) VALUES ('ws-1');
|
|
INSERT INTO tasks (id, workspace_id) VALUES ('task-1', 'ws-1');`); err != nil {
|
|
t.Fatalf("create task fixtures: %v", err)
|
|
}
|
|
if _, err := db.ExecContext(ctx, `
|
|
CREATE TABLE github_pr_watches (
|
|
id TEXT PRIMARY KEY,
|
|
workspace_id TEXT NOT NULL DEFAULT '',
|
|
session_id TEXT NOT NULL DEFAULT '',
|
|
task_id TEXT NOT NULL,
|
|
repository_id TEXT NOT NULL DEFAULT '',
|
|
owner TEXT NOT NULL,
|
|
repo TEXT NOT NULL,
|
|
pr_number INTEGER NOT NULL,
|
|
branch TEXT NOT NULL,
|
|
last_checked_at TIMESTAMP,
|
|
last_comment_at TIMESTAMP,
|
|
last_check_status TEXT DEFAULT '',
|
|
last_review_state TEXT DEFAULT '',
|
|
created_at TIMESTAMP NOT NULL,
|
|
updated_at TIMESTAMP NOT NULL
|
|
);
|
|
CREATE UNIQUE INDEX idx_github_pr_watches_discovered
|
|
ON github_pr_watches (task_id, repository_id, pr_number) WHERE pr_number <> 0;
|
|
CREATE OR REPLACE FUNCTION insert_colliding_pr_watch()
|
|
RETURNS trigger AS $$
|
|
BEGIN
|
|
IF NEW.id = 'watch-source-pg' AND NEW.pr_number = 99 THEN
|
|
INSERT INTO github_pr_watches (
|
|
id, workspace_id, session_id, task_id, repository_id, owner, repo,
|
|
pr_number, branch, created_at, updated_at
|
|
) VALUES (
|
|
'watch-sibling-pg', NEW.workspace_id, 'session-2', NEW.task_id,
|
|
NEW.repository_id, NEW.owner, NEW.repo, NEW.pr_number,
|
|
'feature/B', NEW.updated_at, NEW.updated_at
|
|
);
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
CREATE TRIGGER insert_colliding_pr_watch_before_update
|
|
BEFORE UPDATE OF pr_number ON github_pr_watches
|
|
FOR EACH ROW EXECUTE FUNCTION insert_colliding_pr_watch();`); err != nil {
|
|
t.Fatalf("create postgres PR-watch fixture: %v", err)
|
|
}
|
|
store := &Store{db: db, ro: db}
|
|
now := time.Now().UTC()
|
|
source := &PRWatch{
|
|
ID: "watch-source-pg", WorkspaceID: "ws-1", SessionID: "session-1", TaskID: "task-1",
|
|
RepositoryID: "repo-1", Owner: "owner", Repo: "repo", Branch: "feature/A",
|
|
CreatedAt: now, UpdatedAt: now,
|
|
}
|
|
if err := store.CreatePRWatch(ctx, source); err != nil {
|
|
t.Fatalf("create source watch: %v", err)
|
|
}
|
|
|
|
if err := store.UpdatePRWatchPRNumber(ctx, source.ID, 99); err != nil {
|
|
t.Fatalf("update-time discovered-watch collision must keep transaction usable: %v", err)
|
|
}
|
|
|
|
got, err := store.GetPRWatch(ctx, source.ID)
|
|
if err != nil {
|
|
t.Fatalf("get source watch: %v", err)
|
|
}
|
|
if got != nil {
|
|
t.Fatalf("expected source watch to be dropped after update-time collision, got %+v", got)
|
|
}
|
|
}
|
|
|
|
func TestStore_ResetPRWatch_DropsDiscoveredSourceOnSearchingCollision(t *testing.T) {
|
|
store := newTestStore(t)
|
|
ctx := context.Background()
|
|
now := time.Now().UTC()
|
|
source := &PRWatch{
|
|
ID: "watch-source", WorkspaceID: "ws-1", SessionID: "session-1", TaskID: "task-1",
|
|
RepositoryID: "repo-1", Owner: "owner", Repo: "repo", PRNumber: 7, Branch: "feature/A",
|
|
CreatedAt: now, UpdatedAt: now,
|
|
}
|
|
sibling := &PRWatch{
|
|
ID: "watch-sibling", WorkspaceID: "ws-1", SessionID: "session-2", TaskID: "task-1",
|
|
RepositoryID: "repo-1", Owner: "owner", Repo: "repo", PRNumber: 0, Branch: "feature/B",
|
|
CreatedAt: now, UpdatedAt: now,
|
|
}
|
|
for _, watch := range []*PRWatch{source, sibling} {
|
|
if err := store.CreatePRWatch(ctx, watch); err != nil {
|
|
t.Fatalf("create watch %s: %v", watch.ID, err)
|
|
}
|
|
}
|
|
|
|
if err := store.ResetPRWatch(ctx, source.ID, sibling.Branch); err != nil {
|
|
t.Fatalf("ResetPRWatch: %v", err)
|
|
}
|
|
if got, err := store.GetPRWatch(ctx, source.ID); err != nil {
|
|
t.Fatalf("get source watch: %v", err)
|
|
} else if got != nil {
|
|
t.Fatalf("source watch still exists after reset collision: %+v", got)
|
|
}
|
|
if got, err := store.GetPRWatch(ctx, sibling.ID); err != nil {
|
|
t.Fatalf("get sibling watch: %v", err)
|
|
} else if got == nil {
|
|
t.Fatal("sibling searching watch was removed; want it to survive")
|
|
}
|
|
}
|
|
|
|
// TestStore_ReviewWatch_ListTaskIDsAndReset pins the contract used by the
|
|
// review watch reset flow: every dedup row's task_id (including empty
|
|
// reservations) is enumerable, and ResetReviewWatchState wipes those rows
|
|
// atomically alongside last_polled_at.
|
|
func TestStore_ReviewWatch_ListTaskIDsAndReset(t *testing.T) {
|
|
store := newTestStore(t)
|
|
ctx := context.Background()
|
|
|
|
rw := &ReviewWatch{
|
|
WorkspaceID: "ws-1",
|
|
WorkflowID: "wf-1",
|
|
WorkflowStepID: "step-1",
|
|
AgentProfileID: "agent-1",
|
|
ExecutorProfileID: "exec-1",
|
|
ReviewScope: "user_and_teams",
|
|
Enabled: true,
|
|
}
|
|
if err := store.CreateReviewWatch(ctx, rw); err != nil {
|
|
t.Fatalf("create review watch: %v", err)
|
|
}
|
|
now := time.Now().UTC()
|
|
rw.LastPolledAt = &now
|
|
rw.UpdatedAt = now
|
|
if err := store.UpdateReviewWatch(ctx, rw); err != nil {
|
|
t.Fatalf("stamp last polled: %v", err)
|
|
}
|
|
|
|
tasks := []*ReviewPRTask{
|
|
{ReviewWatchID: rw.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: 1, PRURL: "https://github.com/acme/widget/pull/1", TaskID: "task-a", CreatedAt: now},
|
|
{ReviewWatchID: rw.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: 2, PRURL: "https://github.com/acme/widget/pull/2", TaskID: "task-b", CreatedAt: now},
|
|
{ReviewWatchID: rw.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: 3, PRURL: "https://github.com/acme/widget/pull/3", TaskID: "", CreatedAt: now},
|
|
}
|
|
for _, tp := range tasks {
|
|
if err := store.CreateReviewPRTask(ctx, tp); err != nil {
|
|
t.Fatalf("create review pr task: %v", err)
|
|
}
|
|
}
|
|
|
|
ids, err := store.ListReviewPRTaskIDsByWatch(ctx, rw.ID)
|
|
if err != nil {
|
|
t.Fatalf("list ids: %v", err)
|
|
}
|
|
if len(ids) != 3 {
|
|
t.Fatalf("ListReviewPRTaskIDsByWatch returned %d rows, want 3 (including empty reservation)", len(ids))
|
|
}
|
|
nonEmpty := 0
|
|
for _, id := range ids {
|
|
if id != "" {
|
|
nonEmpty++
|
|
}
|
|
}
|
|
if nonEmpty != 2 {
|
|
t.Errorf("expected 2 non-empty task IDs, got %d", nonEmpty)
|
|
}
|
|
|
|
if err := store.ResetReviewWatchState(ctx, rw.ID); err != nil {
|
|
t.Fatalf("reset: %v", err)
|
|
}
|
|
|
|
idsAfter, err := store.ListReviewPRTaskIDsByWatch(ctx, rw.ID)
|
|
if err != nil {
|
|
t.Fatalf("list ids after reset: %v", err)
|
|
}
|
|
if len(idsAfter) != 0 {
|
|
t.Errorf("expected 0 dedup rows after reset, got %d", len(idsAfter))
|
|
}
|
|
got, err := store.GetReviewWatch(ctx, rw.ID)
|
|
if err != nil {
|
|
t.Fatalf("get watch: %v", err)
|
|
}
|
|
if got.LastPolledAt != nil {
|
|
t.Errorf("expected LastPolledAt to be nil after reset, got %v", got.LastPolledAt)
|
|
}
|
|
}
|
|
|
|
// TestStore_IssueWatch_ListTaskIDsAndReset mirrors the review watch test for
|
|
// the issue watch reset path.
|
|
func TestStore_IssueWatch_ListTaskIDsAndReset(t *testing.T) {
|
|
store := newTestStore(t)
|
|
ctx := context.Background()
|
|
|
|
iw := &IssueWatch{
|
|
WorkspaceID: "ws-1",
|
|
WorkflowID: "wf-1",
|
|
WorkflowStepID: "step-1",
|
|
AgentProfileID: "agent-1",
|
|
ExecutorProfileID: "exec-1",
|
|
Enabled: true,
|
|
}
|
|
if err := store.CreateIssueWatch(ctx, iw); err != nil {
|
|
t.Fatalf("create issue watch: %v", err)
|
|
}
|
|
now := time.Now().UTC()
|
|
iw.LastPolledAt = &now
|
|
iw.UpdatedAt = now
|
|
if err := store.UpdateIssueWatch(ctx, iw); err != nil {
|
|
t.Fatalf("stamp last polled: %v", err)
|
|
}
|
|
|
|
// Three reservations, only two get task IDs assigned so the third
|
|
// exercises the empty-ID inclusion the reset flow depends on.
|
|
for _, n := range []int{1, 2, 3} {
|
|
if _, err := store.ReserveIssueWatchTask(ctx, iw.ID, "acme", "widget", n, "https://github.com/acme/widget/issues/0"); err != nil {
|
|
t.Fatalf("reserve %d: %v", n, err)
|
|
}
|
|
}
|
|
if err := store.AssignIssueWatchTaskID(ctx, iw.ID, "acme", "widget", 1, "task-a"); err != nil {
|
|
t.Fatalf("assign 1: %v", err)
|
|
}
|
|
if err := store.AssignIssueWatchTaskID(ctx, iw.ID, "acme", "widget", 2, "task-b"); err != nil {
|
|
t.Fatalf("assign 2: %v", err)
|
|
}
|
|
|
|
ids, err := store.ListIssueWatchTaskIDsByWatch(ctx, iw.ID)
|
|
if err != nil {
|
|
t.Fatalf("list ids: %v", err)
|
|
}
|
|
if len(ids) != 3 {
|
|
t.Fatalf("ListIssueWatchTaskIDsByWatch returned %d rows, want 3 (including empty reservation)", len(ids))
|
|
}
|
|
|
|
if err := store.ResetIssueWatchState(ctx, iw.ID); err != nil {
|
|
t.Fatalf("reset: %v", err)
|
|
}
|
|
|
|
idsAfter, err := store.ListIssueWatchTaskIDsByWatch(ctx, iw.ID)
|
|
if err != nil {
|
|
t.Fatalf("list ids after reset: %v", err)
|
|
}
|
|
if len(idsAfter) != 0 {
|
|
t.Errorf("expected 0 dedup rows after reset, got %d", len(idsAfter))
|
|
}
|
|
got, err := store.GetIssueWatch(ctx, iw.ID)
|
|
if err != nil {
|
|
t.Fatalf("get watch: %v", err)
|
|
}
|
|
if got.LastPolledAt != nil {
|
|
t.Errorf("expected LastPolledAt to be nil after reset, got %v", got.LastPolledAt)
|
|
}
|
|
}
|