mathiasandClaude Sonnet 5.5
c4233e3e05
PID%1000 collided with a concurrent job on the shared CI host (run on
e946709: process already listening on port 54000). freePort asks the OS;
Start is retried up to 3 times if the port is taken in between.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SZkhi49hfeSqBp5aZJNXsM
214 lines
6.6 KiB
Go
214 lines
6.6 KiB
Go
package store_test
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"os"
|
|
"path/filepath"
|
|
"testing"
|
|
|
|
embeddedpostgres "github.com/fergusstrange/embedded-postgres"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"git.d-ma.be/mathias/tapir/internal/adapters/store"
|
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
|
"git.d-ma.be/mathias/tapir/internal/ports"
|
|
)
|
|
|
|
// Static check: Store satisfies the Sink port.
|
|
var _ ports.Sink = (*store.Store)(nil)
|
|
|
|
// dsn points at the in-process Postgres started in TestMain. Tests run against
|
|
// real SQL (constraints, ON CONFLICT, jsonb) — not a mock — without docker or
|
|
// live-cluster credentials (embedded-postgres downloads its own PG binary).
|
|
var dsn string
|
|
|
|
func TestMain(m *testing.M) {
|
|
// Runtime/data dirs are per-process (PID-named) and the port is OS-assigned,
|
|
// so concurrent `go test` runs on a shared CI host don't collide on either.
|
|
// CachePath is shared so the PG archive is downloaded once, not per process.
|
|
pg, port, rt, err := startTestPostgres(fmt.Sprintf("tapir-epg-store-%d", os.Getpid()))
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "embedded-postgres start: %v\n", err)
|
|
os.Exit(1)
|
|
}
|
|
dsn = fmt.Sprintf("postgres://postgres:postgres@localhost:%d/postgres?sslmode=disable", port)
|
|
|
|
code := m.Run()
|
|
|
|
if err := pg.Stop(); err != nil {
|
|
fmt.Fprintf(os.Stderr, "embedded-postgres stop: %v\n", err)
|
|
}
|
|
_ = os.RemoveAll(rt)
|
|
os.Exit(code)
|
|
}
|
|
|
|
// freePort asks the OS for an unused TCP port. The port is released before
|
|
// return, so another process can grab it first; startTestPostgres retries.
|
|
func freePort() (uint32, error) {
|
|
l, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
return 0, fmt.Errorf("listen for free port: %w", err)
|
|
}
|
|
defer func() { _ = l.Close() }()
|
|
return uint32(l.Addr().(*net.TCPAddr).Port), nil
|
|
}
|
|
|
|
// startTestPostgres starts an embedded Postgres on an OS-assigned free port and
|
|
// returns it with that port and its runtime dir (caller removes it). A port
|
|
// taken between freePort and Start (shared CI host) is retried with a new one.
|
|
func startTestPostgres(name string) (*embeddedpostgres.EmbeddedPostgres, uint32, string, error) {
|
|
rt := filepath.Join(os.TempDir(), name)
|
|
var lastErr error
|
|
for range 3 {
|
|
port, err := freePort()
|
|
if err != nil {
|
|
return nil, 0, rt, err
|
|
}
|
|
pg := embeddedpostgres.NewDatabase(
|
|
embeddedpostgres.DefaultConfig().
|
|
Port(port).
|
|
RuntimePath(rt).
|
|
DataPath(filepath.Join(rt, "data")).
|
|
BinariesPath(filepath.Join(rt, "bin")).
|
|
CachePath(filepath.Join(os.TempDir(), "tapir-epg-cache")),
|
|
)
|
|
if lastErr = pg.Start(); lastErr == nil {
|
|
return pg, port, rt, nil
|
|
}
|
|
lastErr = fmt.Errorf("start on port %d: %w", port, lastErr)
|
|
}
|
|
return nil, 0, rt, lastErr
|
|
}
|
|
|
|
// uuids — fixed so tests are deterministic. user_id/video_id are UUID columns.
|
|
const (
|
|
userA = "11111111-1111-1111-1111-111111111111"
|
|
userB = "22222222-2222-2222-2222-222222222222"
|
|
videoX = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
|
videoY = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
|
|
)
|
|
|
|
func newStore(t *testing.T) *store.Store {
|
|
t.Helper()
|
|
s, err := store.New(context.Background(), dsn)
|
|
require.NoError(t, err, "migrations must apply clean and the pool must connect")
|
|
t.Cleanup(s.Close)
|
|
return s
|
|
}
|
|
|
|
// rawPool is a direct connection for test introspection (counts, truncation),
|
|
// kept out of the production Store API. Closed via t.Cleanup.
|
|
func rawPool(t *testing.T) *pgxpool.Pool {
|
|
t.Helper()
|
|
p, err := pgxpool.New(context.Background(), dsn)
|
|
require.NoError(t, err)
|
|
t.Cleanup(p.Close)
|
|
return p
|
|
}
|
|
|
|
// resetDB truncates between tests so each starts from a known state. Schema is
|
|
// shared across the run (migrations are idempotent via New).
|
|
func resetDB(t *testing.T, p *pgxpool.Pool) {
|
|
t.Helper()
|
|
_, err := p.Exec(context.Background(),
|
|
`TRUNCATE login_events, summary_actions, sink_deliveries, summaries, transcripts, videos, users CASCADE`)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func summary(userID, videoID, text string) domain.Summary {
|
|
return domain.Summary{
|
|
UserID: userID,
|
|
VideoID: videoID,
|
|
Summary: text,
|
|
Highlights: []string{"h1", "h2"},
|
|
Takeaways: []string{"t1"},
|
|
AIProvider: "local",
|
|
AIModel: "qwen",
|
|
}
|
|
}
|
|
|
|
func TestName(t *testing.T) {
|
|
require.Equal(t, "store", newStore(t).Name())
|
|
}
|
|
|
|
func TestDeliverInsertsSummary(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := newStore(t)
|
|
resetDB(t, rawPool(t))
|
|
|
|
require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "first")))
|
|
|
|
ok, err := s.HasSummary(ctx, userA, videoX)
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
}
|
|
|
|
func TestDeliverIsIdempotentOnUserVideo(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := newStore(t)
|
|
p := rawPool(t)
|
|
resetDB(t, p)
|
|
|
|
require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "first")))
|
|
// Re-deliver the same (user, video): must update in place, not duplicate or error.
|
|
require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "second")))
|
|
|
|
var count int
|
|
require.NoError(t, p.QueryRow(ctx,
|
|
`SELECT count(*) FROM summaries WHERE user_id = $1 AND video_id = $2`,
|
|
userA, videoX).Scan(&count))
|
|
require.Equal(t, 1, count, "second delivery must update, not duplicate")
|
|
|
|
var text string
|
|
require.NoError(t, p.QueryRow(ctx,
|
|
`SELECT summary FROM summaries WHERE user_id = $1 AND video_id = $2`,
|
|
userA, videoX).Scan(&text))
|
|
require.Equal(t, "second", text, "second delivery must overwrite the summary text")
|
|
|
|
// Exactly one store-delivery row for the summary (ON CONFLICT update).
|
|
var deliveries int
|
|
require.NoError(t, p.QueryRow(ctx,
|
|
`SELECT count(*) FROM sink_deliveries d
|
|
JOIN summaries m ON m.id = d.summary_id
|
|
WHERE m.user_id = $1 AND m.video_id = $2 AND d.sink = 'store'`,
|
|
userA, videoX).Scan(&deliveries))
|
|
require.Equal(t, 1, deliveries)
|
|
}
|
|
|
|
func TestSeenVideoIDsReturnsUsersSet(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := newStore(t)
|
|
resetDB(t, rawPool(t))
|
|
|
|
require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "x")))
|
|
require.NoError(t, s.Deliver(ctx, summary(userA, videoY, "y")))
|
|
|
|
seen, err := s.SeenVideoIDs(ctx, userA)
|
|
require.NoError(t, err)
|
|
require.Equal(t, map[string]bool{videoX: true, videoY: true}, seen)
|
|
}
|
|
|
|
func TestPerUserIsolation(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := newStore(t)
|
|
resetDB(t, rawPool(t))
|
|
|
|
require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "a-owns-this")))
|
|
|
|
// User B must not see user A's video, by either dedup read.
|
|
seenB, err := s.SeenVideoIDs(ctx, userB)
|
|
require.NoError(t, err)
|
|
require.Empty(t, seenB, "user B must not see user A's videos")
|
|
|
|
hasB, err := s.HasSummary(ctx, userB, videoX)
|
|
require.NoError(t, err)
|
|
require.False(t, hasB, "the same video_id under another user must be invisible")
|
|
|
|
hasA, err := s.HasSummary(ctx, userA, videoX)
|
|
require.NoError(t, err)
|
|
require.True(t, hasA)
|
|
}
|