angel/inventory-plus-plus
1
0
Fork 0
Code Issues 34 Pull Requests Actions Packages Projects 4 Releases Wiki Activity

tests: migrate assertions to testify's assert/require

Browse Source 
Replaces raw t.Error/t.Errorf/t.Fatal/t.Fatalf across every test file
that has any (domains/accounts, domains/authentication,
domains/raw_events, domains/amazon, domains/reports x2) with testify's
assert (non-halting) / require (halting) equivalents. The three
Example-based tests (server/ui/svg, server/ui/charts) have no
*testing.T at all - nothing to convert there.

require.Eventually replaces several hand-rolled polling loops in
domains/amazon/mock_test.go. Its condition function runs on a separate
goroutine (confirmed in testify's source), so calling require.* from
inside one - which two of the new Eventually calls initially did, via
the isProcessed helper - is unsafe per Go's testing rules (t.FailNow
must only be called from the test's own goroutine). Fixed by splitting
a *testing.T-free queryIsProcessed(ctx, pool, shopID, eventID) out of
isProcessed for use inside those closures specifically.

github.com/stretchr/testify promoted from an indirect to a direct
dependency (go.mod only - it was already present transitively, so
go.sum is unchanged).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XEDaCB7C2NEBgyvqEtZuxY
This commit is contained in:
angel
2026-08-20 00:24:04 -06:00
co-authored by Claude Sonnet 5
parent e0735854ce
commit 3a1d05857e
8 changed files with 230 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files
domains/amazon/mock_test.go
+99 -152
View File
@@ -8,6 +8,8 @@ import (
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"ruben/inventory2/domains/raw_events"
"ruben/inventory2/internal/testdb"
@@ -70,9 +72,15 @@ func (s *notifySpy) ackAt(t *testing.T, i int) {
s.mu.Lock()
ack := s.acks[i]
s.mu.Unlock()
if err := ack(context.Background()); err != nil {
t.Fatalf("ack() error = %v", err)
}
require.NoError(t, ack(context.Background()), "ack()")
}
// completedCountSnapshot is safe to call from another goroutine (e.g. from
// inside require.Eventually's condition), unlike helpers that call t.Fatal.
func (s *notifySpy) completedCountSnapshot() int {
s.mu.Lock()
defer s.mu.Unlock()
return s.completedCount
}
// waitForCount blocks until at least n Notify calls have fully completed
@@ -80,20 +88,9 @@ func (s *notifySpy) ackAt(t *testing.T, i int) {
// more than once with increasing n.
func (s *notifySpy) waitForCount(t *testing.T, n int, timeout time.Duration) {
t.Helper()
deadline := time.Now().Add(timeout)
for {
s.mu.Lock()
got := s.completedCount
s.mu.Unlock()
if got >= n {
return
}
if time.Now().After(deadline) {
t.Fatalf("timed out after %v waiting for %d total completed notifications (got %d so far)", timeout, n, got)
}
time.Sleep(5 * time.Millisecond)
}
require.Eventually(t, func() bool {
return s.completedCountSnapshot() >= n
}, timeout, 5*time.Millisecond, "timed out waiting for %d total completed notifications (got %d)", n, s.completedCountSnapshot())
}
// insertRawAmazonEvent inserts directly into mock.raw_shop_events, the same
@@ -109,9 +106,7 @@ func insertRawAmazonEvent(t *testing.T, pool *pgxpool.Pool, shopID, eventID stri
INSERT INTO mock.raw_shop_events (platform, shop_id, event_timestamp, event_id, raw_payload)
VALUES ('amazon', $1, NOW(), $2, '{}'::jsonb)
`, shopID, eventID)
if err != nil {
t.Fatalf("failed to insert raw amazon event: %v", err)
}
require.NoError(t, err, "insert raw amazon event")
}
// insertRawAmazonEvents bulk-inserts n events for shopID in one statement,
@@ -123,22 +118,27 @@ func insertRawAmazonEvents(t *testing.T, pool *pgxpool.Pool, shopID string, n in
SELECT 'amazon', $1, NOW() + (s || ' milliseconds')::interval, 'evt-' || s, '{}'::jsonb
FROM generate_series(1, $2) AS s
`, shopID, n)
if err != nil {
t.Fatalf("failed to insert %d raw amazon events: %v", n, err)
}
require.NoError(t, err, "insert %d raw amazon events", n)
}
func isProcessed(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) bool {
t.Helper()
// queryIsProcessed has no *testing.T dependency, so it's safe to call from
// require.Eventually's condition function (which testify runs on a
// separate goroutine - t.Fatal/require must only be called from the main
// test goroutine). isProcessed wraps it for direct, main-goroutine use.
func queryIsProcessed(ctx context.Context, pool *pgxpool.Pool, shopID, eventID string) (bool, error) {
var processed bool
err := pool.QueryRow(context.Background(), `
err := pool.QueryRow(ctx, `
SELECT processed_at IS NOT NULL
FROM mock.shop_amazon_events
WHERE shop_id = $1 AND event_id = $2
`, shopID, eventID).Scan(&processed)
if err != nil {
t.Fatalf("failed to check processed state: %v", err)
}
return processed, err
}
func isProcessed(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) bool {
t.Helper()
processed, err := queryIsProcessed(context.Background(), pool, shopID, eventID)
require.NoError(t, err, "check processed state")
return processed
}
@@ -152,9 +152,7 @@ func isNotified(t *testing.T, pool *pgxpool.Pool, shopID, eventID string) bool {
FROM mock.shop_amazon_events
WHERE shop_id = $1 AND event_id = $2
`, shopID, eventID).Scan(&notified)
if err != nil {
t.Fatalf("failed to check notified state: %v", err)
}
require.NoError(t, err, "check notified state")
return notified
}
@@ -164,9 +162,7 @@ func countUnprocessed(t *testing.T, pool *pgxpool.Pool, shopID string) int {
err := pool.QueryRow(context.Background(), `
SELECT count(*) FROM mock.shop_amazon_events WHERE shop_id = $1 AND processed_at IS NULL
`, shopID).Scan(&n)
if err != nil {
t.Fatalf("failed to count unprocessed events: %v", err)
}
require.NoError(t, err, "count unprocessed events")
return n
}
@@ -197,12 +193,33 @@ func terminateListenConnection(t *testing.T, pool *pgxpool.Pool) {
ORDER BY backend_start DESC
LIMIT 1
`, eventChannelName).Scan(&pid)
if err != nil {
t.Fatalf("failed to find the LISTEN connection's backend pid: %v", err)
}
require.NoError(t, err, "find the LISTEN connection's backend pid")
if _, err := pool.Exec(ctx, `SELECT pg_terminate_backend($1)`, pid); err != nil {
t.Fatalf("failed to terminate backend %d: %v", pid, err)
_, err = pool.Exec(ctx, `SELECT pg_terminate_backend($1)`, pid)
require.NoError(t, err, "terminate backend %d", pid)
}
// waitReady blocks until m signals it's actively listening, failing the
// test if that doesn't happen within timeout.
func waitReady(t *testing.T, m *Mocks, timeout time.Duration) {
t.Helper()
select {
case <-m.Ready():
case <-time.After(timeout):
require.Fail(t, "ProcessEvents() did not become ready (LISTEN registered)", "within %v", timeout)
}
}
// waitStopped blocks until ProcessEvents returns on errCh, asserting it
// returns a nil error, failing the test if that doesn't happen within
// timeout.
func waitStopped(t *testing.T, errCh <-chan error, timeout time.Duration) {
t.Helper()
select {
case err := <-errCh:
require.NoError(t, err, "ProcessEvents() after context cancellation")
case <-time.After(timeout):
require.Fail(t, "ProcessEvents() did not return", "within %v of context cancellation", timeout)
}
}
@@ -218,15 +235,12 @@ func TestProcessUnprocessedEvents_ProcessesAllEventsAcrossBatches(t *testing.T)
const n = 150 // exceeds the 100-row LIMIT per batch inside processUnprocessedEvents
insertRawAmazonEvents(t, pool, shopID, n)
if err := m.processUnprocessedEvents(ctx); err != nil {
t.Fatalf("processUnprocessedEvents() error = %v", err)
}
require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents()")
spy.waitForCount(t, n, 5*time.Second)
if got := countUnprocessed(t, pool, shopID); got != 0 {
t.Errorf("countUnprocessed() = %d, want 0 (all %d events should be processed across multiple 100-row batches)", got, n)
}
assert.Zero(t, countUnprocessed(t, pool, shopID),
"countUnprocessed() should be 0 (all %d events should be processed across multiple 100-row batches)", n)
}
func TestProcessUnprocessedEvents_NoListenerConfigured(t *testing.T) {
@@ -239,18 +253,14 @@ func TestProcessUnprocessedEvents_NoListenerConfigured(t *testing.T) {
insertRawAmazonEvent(t, pool, shopID, "evt-1")
if err := m.processUnprocessedEvents(ctx); err != nil {
t.Fatalf("processUnprocessedEvents() error = %v", err)
}
require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents()")
// with no listener, nothing ever acks - the event is expected to stay
// "notified" forever, not silently marked processed.
if isProcessed(t, pool, shopID, "evt-1") {
t.Error("event was marked processed despite no listener being configured to ack it")
}
if !isNotified(t, pool, shopID, "evt-1") {
t.Error("event should be in the notified state after being handed off with no listener to ack it")
}
assert.False(t, isProcessed(t, pool, shopID, "evt-1"),
"event was marked processed despite no listener being configured to ack it")
assert.True(t, isNotified(t, pool, shopID, "evt-1"),
"event should be in the notified state after being handed off with no listener to ack it")
}
// TestProcessUnprocessedEvents_RetriesUnackedNotification is the core of
@@ -271,47 +281,29 @@ func TestProcessUnprocessedEvents_RetriesUnackedNotification(t *testing.T) {
insertRawAmazonEvent(t, pool, shopID, "evt-1")
if err := m.processUnprocessedEvents(ctx); err != nil {
t.Fatalf("processUnprocessedEvents() (1st pass) error = %v", err)
}
require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (1st pass)")
spy.waitForCount(t, 1, 2*time.Second)
if isProcessed(t, pool, shopID, "evt-1") {
t.Fatal("event was marked processed despite the listener never acking")
}
if !isNotified(t, pool, shopID, "evt-1") {
t.Fatal("event should be in the notified state after the first dispatch")
}
require.False(t, isProcessed(t, pool, shopID, "evt-1"), "event was marked processed despite the listener never acking")
require.True(t, isNotified(t, pool, shopID, "evt-1"), "event should be in the notified state after the first dispatch")
// still within notifyRetryAfter: shouldn't be re-notified yet.
if err := m.processUnprocessedEvents(ctx); err != nil {
t.Fatalf("processUnprocessedEvents() (immediate 2nd pass) error = %v", err)
}
if got := len(spy.eventsSnapshot()); got != 1 {
t.Fatalf("listener was notified %d times before notifyRetryAfter elapsed, want 1", got)
}
require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (immediate 2nd pass)")
require.Len(t, spy.eventsSnapshot(), 1, "listener should not have been notified again before notifyRetryAfter elapsed")
time.Sleep(60 * time.Millisecond) // past notifyRetryAfter
if err := m.processUnprocessedEvents(ctx); err != nil {
t.Fatalf("processUnprocessedEvents() (3rd pass, after retry window) error = %v", err)
}
require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (3rd pass, after retry window)")
spy.waitForCount(t, 2, 2*time.Second)
// the listener "finishes" the first notification late, via the ack it
// was originally handed - not a fresh one from the retry.
spy.ackAt(t, 0)
if !isProcessed(t, pool, shopID, "evt-1") {
t.Fatal("event should be processed once any recorded ack for it is called")
}
require.True(t, isProcessed(t, pool, shopID, "evt-1"), "event should be processed once any recorded ack for it is called")
if err := m.processUnprocessedEvents(ctx); err != nil {
t.Fatalf("processUnprocessedEvents() (4th pass, after ack) error = %v", err)
}
if got := len(spy.eventsSnapshot()); got != 2 {
t.Fatalf("listener was notified again after being acked: got %d calls, want 2", got)
}
require.NoError(t, m.processUnprocessedEvents(ctx), "processUnprocessedEvents() (4th pass, after ack)")
assert.Len(t, spy.eventsSnapshot(), 2, "listener should not be notified again after being acked")
}
// TestProcessEvents_PollFallbackPicksUpRetryDueEvents proves the poll
@@ -341,11 +333,7 @@ func TestProcessEvents_PollFallbackPicksUpRetryDueEvents(t *testing.T) {
errCh <- m.ProcessEvents(ctx)
}()
select {
case <-m.Ready():
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not become ready (LISTEN registered) within 5s")
}
waitReady(t, m, 5*time.Second)
insertRawAmazonEvent(t, pool, shopID, "evt-1")
@@ -358,14 +346,7 @@ func TestProcessEvents_PollFallbackPicksUpRetryDueEvents(t *testing.T) {
spy.waitForCount(t, 2, 3*time.Second)
cancel()
select {
case err := <-errCh:
if err != nil {
t.Fatalf("ProcessEvents() returned error = %v after context cancellation, want nil", err)
}
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not return within 5s of context cancellation")
}
waitStopped(t, errCh, 5*time.Second)
}
// TestProcessEvents_ReactsToNotification drives the actual long-running
@@ -389,36 +370,24 @@ func TestProcessEvents_ReactsToNotification(t *testing.T) {
errCh <- m.ProcessEvents(ctx)
}()
select {
case <-m.Ready():
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not become ready (LISTEN registered) within 5s")
}
waitReady(t, m, 5*time.Second)
insertRawAmazonEvent(t, pool, shopID, "evt-1")
deadline := time.Now().Add(5 * time.Second)
for !isProcessed(t, pool, shopID, "evt-1") {
if time.Now().After(deadline) {
t.Fatal("event was not processed within 5s of insertion - the reactive LISTEN/NOTIFY wake-up did not fire (the 1-minute poll fallback would eventually catch it, but this test intentionally doesn't wait that long)")
}
time.Sleep(20 * time.Millisecond)
}
require.Eventually(t, func() bool {
processed, _ := queryIsProcessed(context.Background(), pool, shopID, "evt-1")
return processed
}, 5*time.Second, 20*time.Millisecond,
"event was not processed within 5s of insertion - the reactive LISTEN/NOTIFY wake-up did not fire "+
"(the 1-minute poll fallback would eventually catch it, but this test intentionally doesn't wait that long)")
spy.waitForCount(t, 1, 2*time.Second)
if got := spy.eventsSnapshot()[0]; got.StoreID != shopID || got.EventID != "evt-1" {
t.Errorf("listener notified with %+v, want StoreID=%q EventID=%q", got, shopID, "evt-1")
}
got := spy.eventsSnapshot()[0]
assert.Equal(t, shopID, got.StoreID, "listener notified with unexpected StoreID")
assert.Equal(t, "evt-1", got.EventID, "listener notified with unexpected EventID")
cancel()
select {
case err := <-errCh:
if err != nil {
t.Fatalf("ProcessEvents() returned error = %v after context cancellation, want nil", err)
}
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not return within 5s of context cancellation")
}
waitStopped(t, errCh, 5*time.Second)
}
// TestProcessEvents_ShutsDownOnContextCancel checks the lifecycle in
@@ -436,21 +405,10 @@ func TestProcessEvents_ShutsDownOnContextCancel(t *testing.T) {
errCh <- m.ProcessEvents(ctx)
}()
select {
case <-m.Ready():
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not become ready (LISTEN registered) within 5s")
}
waitReady(t, m, 5*time.Second)
cancel()
select {
case err := <-errCh:
if err != nil {
t.Fatalf("ProcessEvents() returned error = %v after context cancellation, want nil", err)
}
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not return within 5s of context cancellation")
}
waitStopped(t, errCh, 5*time.Second)
}
// TestProcessEvents_ReconnectsAfterListenConnectionDrops is insight #4's
@@ -474,11 +432,7 @@ func TestProcessEvents_ReconnectsAfterListenConnectionDrops(t *testing.T) {
errCh <- m.ProcessEvents(ctx)
}()
select {
case <-m.Ready():
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not become ready (LISTEN registered) within 5s")
}
waitReady(t, m, 5*time.Second)
terminateListenConnection(t, pool)
@@ -486,27 +440,20 @@ func TestProcessEvents_ReconnectsAfterListenConnectionDrops(t *testing.T) {
// backoff is 1s) - it must not have returned because of this.
select {
case err := <-errCh:
t.Fatalf("ProcessEvents() returned (err = %v) after its LISTEN connection was killed, want it to reconnect and keep running", err)
require.Fail(t, "ProcessEvents() returned after its LISTEN connection was killed, want it to reconnect and keep running",
"err = %v", err)
case <-time.After(2 * time.Second):
}
insertRawAmazonEvent(t, pool, shopID, "evt-1")
deadline := time.Now().Add(5 * time.Second)
for !isProcessed(t, pool, shopID, "evt-1") {
if time.Now().After(deadline) {
t.Fatal("event was not processed within 5s of insertion after the LISTEN connection was forcibly dropped - reconnection did not restore the reactive path")
}
time.Sleep(20 * time.Millisecond)
}
require.Eventually(t, func() bool {
processed, _ := queryIsProcessed(context.Background(), pool, shopID, "evt-1")
return processed
}, 5*time.Second, 20*time.Millisecond,
"event was not processed within 5s of insertion after the LISTEN connection was forcibly dropped - "+
"reconnection did not restore the reactive path")
cancel()
select {
case err := <-errCh:
if err != nil {
t.Fatalf("ProcessEvents() returned error = %v after context cancellation, want nil", err)
}
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not return within 5s of context cancellation")
}
waitStopped(t, errCh, 5*time.Second)
}