297 lines
11 KiB
Go
297 lines
11 KiB
Go
package contracttest
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"proxy-pool/internal/domain/adminstate"
|
|
)
|
|
|
|
type Factory func(*testing.T) adminstate.Store
|
|
|
|
func Run(t *testing.T, factory Factory) {
|
|
t.Helper()
|
|
t.Run("config transaction and immutability", func(t *testing.T) {
|
|
runConfigContract(t, factory(t))
|
|
})
|
|
t.Run("upstream idempotency", func(t *testing.T) {
|
|
runUpstreamContract(t, factory(t))
|
|
})
|
|
t.Run("routing compare and swap", func(t *testing.T) {
|
|
runRoutingContract(t, factory(t))
|
|
})
|
|
t.Run("outbox lease and acknowledgement", func(t *testing.T) {
|
|
runOutboxContract(t, factory(t))
|
|
})
|
|
t.Run("context cancellation", func(t *testing.T) {
|
|
runContextContract(t, factory(t))
|
|
})
|
|
}
|
|
|
|
func runConfigContract(t *testing.T, store adminstate.Store) {
|
|
t.Helper()
|
|
command := configCommand("req-config", "cfg-1", strings.Repeat("ab", adminstate.SHA256HexBytes/2))
|
|
result, err := store.CommitConfig(context.Background(), command)
|
|
if err != nil || !result.Changed || result.Revision != 1 || result.RequestID != command.RequestID {
|
|
t.Fatalf("CommitConfig(first) = %+v, %v", result, err)
|
|
}
|
|
|
|
snapshot, err := store.Snapshot(context.Background())
|
|
if err != nil || snapshot.Revision != 1 || snapshot.Config == nil ||
|
|
snapshot.Config.ConfigVersion != "cfg-1" || len(snapshot.Upstreams) != 2 || len(snapshot.Routings) != 1 {
|
|
t.Fatalf("Snapshot() = %+v, %v", snapshot, err)
|
|
}
|
|
snapshot.Config.ConfigVersion = "mutated"
|
|
snapshot.Upstreams[0].Name = "mutated"
|
|
snapshot.Routings[0].Upstreams[0] = "mutated"
|
|
again, err := store.Snapshot(context.Background())
|
|
if err != nil || again.Config.ConfigVersion != "cfg-1" ||
|
|
again.Upstreams[0].Name == "mutated" || again.Routings[0].Upstreams[0] == "mutated" {
|
|
t.Fatalf("Snapshot() leaked mutable state: %+v, %v", again, err)
|
|
}
|
|
|
|
command.RequestID = "req-config-replay"
|
|
command.OccurredAt = command.OccurredAt.Add(time.Second)
|
|
command.Checksum = strings.ToUpper(command.Checksum)
|
|
replayed, err := store.CommitConfig(context.Background(), command)
|
|
if err != nil || replayed.Changed || replayed.Revision != 1 {
|
|
t.Fatalf("CommitConfig(replay) = %+v, %v", replayed, err)
|
|
}
|
|
audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10})
|
|
if err != nil || len(audits) != 2 || !audits[0].Changed || audits[1].Changed ||
|
|
audits[0].Revision != 1 || audits[1].Revision != 1 {
|
|
t.Fatalf("ReadAudit() = %+v, %v", audits, err)
|
|
}
|
|
|
|
conflict := command
|
|
conflict.RequestID = "req-config-conflict"
|
|
conflict.Checksum = strings.Repeat("b", adminstate.SHA256HexBytes)
|
|
if _, err := store.CommitConfig(context.Background(), conflict); !errors.Is(err, adminstate.ErrConflict) {
|
|
t.Fatalf("CommitConfig(conflict) error = %v", err)
|
|
}
|
|
audits, err = store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10})
|
|
if err != nil || len(audits) != 2 {
|
|
t.Fatalf("ReadAudit(after conflict) = %+v, %v", audits, err)
|
|
}
|
|
|
|
invalid := command
|
|
invalid.RequestID = "req-config-invalid"
|
|
invalid.Routings[0].CurrentUpstream = "missing"
|
|
if _, err := store.CommitConfig(context.Background(), invalid); !errors.Is(err, adminstate.ErrInvalidCommand) {
|
|
t.Fatalf("CommitConfig(invalid) error = %v", err)
|
|
}
|
|
final, err := store.Snapshot(context.Background())
|
|
if err != nil || final.Revision != 1 || final.Config.ConfigVersion != "cfg-1" {
|
|
t.Fatalf("Snapshot(after invalid) = %+v, %v", final, err)
|
|
}
|
|
}
|
|
|
|
func runUpstreamContract(t *testing.T, store adminstate.Store) {
|
|
t.Helper()
|
|
now := contractNow()
|
|
commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes)))
|
|
|
|
result, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
|
RequestID: "req-disable", Actor: contractActor(), OccurredAt: now.Add(time.Second),
|
|
Name: "provider-a", Enabled: false,
|
|
})
|
|
if err != nil || !result.Changed || result.Revision != 2 {
|
|
t.Fatalf("SetUpstreamEnabled(disable) = %+v, %v", result, err)
|
|
}
|
|
replayed, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
|
RequestID: "req-disable-replay", Actor: contractActor(), OccurredAt: now.Add(2 * time.Second),
|
|
Name: "provider-a", Enabled: false,
|
|
})
|
|
if err != nil || replayed.Changed || replayed.Revision != 2 {
|
|
t.Fatalf("SetUpstreamEnabled(replay) = %+v, %v", replayed, err)
|
|
}
|
|
if _, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
|
RequestID: "req-missing", Actor: contractActor(), OccurredAt: now.Add(3 * time.Second),
|
|
Name: "missing", Enabled: true,
|
|
}); !errors.Is(err, adminstate.ErrNotFound) {
|
|
t.Fatalf("SetUpstreamEnabled(missing) error = %v", err)
|
|
}
|
|
|
|
snapshot, err := store.Snapshot(context.Background())
|
|
if err != nil || snapshot.Revision != 2 || upstreamEnabled(snapshot, "provider-a") {
|
|
t.Fatalf("Snapshot() = %+v, %v", snapshot, err)
|
|
}
|
|
audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10})
|
|
if err != nil || len(audits) != 3 || !audits[1].Changed || audits[2].Changed {
|
|
t.Fatalf("ReadAudit() = %+v, %v", audits, err)
|
|
}
|
|
events, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
|
ConsumerID: "publisher-a", Now: now.Add(4 * time.Second), Limit: 10, Lease: time.Minute,
|
|
})
|
|
if err != nil || len(events) != 2 || events[0].Revision != 1 || events[1].Revision != 2 {
|
|
t.Fatalf("Claim() = %+v, %v", events, err)
|
|
}
|
|
}
|
|
|
|
func runRoutingContract(t *testing.T, store adminstate.Store) {
|
|
t.Helper()
|
|
now := contractNow()
|
|
commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes)))
|
|
|
|
if _, err := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{
|
|
RequestID: "req-bad-target", Actor: contractActor(), OccurredAt: now.Add(time.Second),
|
|
Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-c",
|
|
}); !errors.Is(err, adminstate.ErrInvalidCommand) {
|
|
t.Fatalf("SwitchRouting(bad target) error = %v", err)
|
|
}
|
|
|
|
start := make(chan struct{})
|
|
var changed atomic.Int64
|
|
var conflicts atomic.Int64
|
|
var unexpectedMu sync.Mutex
|
|
var unexpected []error
|
|
var workers sync.WaitGroup
|
|
for index := range 100 {
|
|
workers.Add(1)
|
|
go func(index int) {
|
|
defer workers.Done()
|
|
<-start
|
|
result, err := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{
|
|
RequestID: fmt.Sprintf("req-switch-%03d", index), Actor: contractActor(),
|
|
OccurredAt: now.Add(2 * time.Second), Name: "checkout",
|
|
ExpectedCurrent: "provider-a", Target: "provider-b", Reason: "capacity",
|
|
})
|
|
switch {
|
|
case err == nil && result.Changed:
|
|
changed.Add(1)
|
|
case errors.Is(err, adminstate.ErrConflict):
|
|
conflicts.Add(1)
|
|
default:
|
|
unexpectedMu.Lock()
|
|
unexpected = append(unexpected, err)
|
|
unexpectedMu.Unlock()
|
|
}
|
|
}(index)
|
|
}
|
|
close(start)
|
|
workers.Wait()
|
|
if changed.Load() != 1 || conflicts.Load() != 99 || len(unexpected) != 0 {
|
|
t.Fatalf("concurrent switch changed=%d conflicts=%d unexpected=%v", changed.Load(), conflicts.Load(), unexpected)
|
|
}
|
|
snapshot, err := store.Snapshot(context.Background())
|
|
if err != nil || snapshot.Revision != 2 || snapshot.Routings[0].CurrentUpstream != "provider-b" {
|
|
t.Fatalf("Snapshot() = %+v, %v", snapshot, err)
|
|
}
|
|
}
|
|
|
|
func runOutboxContract(t *testing.T, store adminstate.Store) {
|
|
t.Helper()
|
|
now := contractNow()
|
|
commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes)))
|
|
result, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
|
RequestID: "req-disable", Actor: contractActor(), OccurredAt: now.Add(time.Second),
|
|
Name: "provider-a", Enabled: false,
|
|
})
|
|
if err != nil || !result.Changed {
|
|
t.Fatalf("SetUpstreamEnabled() = %+v, %v", result, err)
|
|
}
|
|
|
|
first, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
|
ConsumerID: "publisher-a", Now: now.Add(2 * time.Second), Limit: 1, Lease: time.Minute,
|
|
})
|
|
if err != nil || len(first) != 1 || first[0].ID != 1 {
|
|
t.Fatalf("Claim(first) = %+v, %v", first, err)
|
|
}
|
|
first[0].Payload[0] = 'X'
|
|
second, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
|
ConsumerID: "publisher-b", Now: now.Add(2 * time.Second), Limit: 10, Lease: time.Minute,
|
|
})
|
|
if err != nil || len(second) != 1 || second[0].ID != 2 {
|
|
t.Fatalf("Claim(second) = %+v, %v", second, err)
|
|
}
|
|
if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{
|
|
ConsumerID: "publisher-b", Now: now.Add(3 * time.Second), EventIDs: []uint64{1},
|
|
}); !errors.Is(err, adminstate.ErrConflict) {
|
|
t.Fatalf("Acknowledge(wrong owner) error = %v", err)
|
|
}
|
|
|
|
reclaimed, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
|
ConsumerID: "publisher-b", Now: now.Add(2*time.Minute + time.Second), Limit: 10, Lease: time.Minute,
|
|
})
|
|
if err != nil || len(reclaimed) != 2 || reclaimed[0].ID != 1 || reclaimed[1].ID != 2 || reclaimed[0].Payload[0] == 'X' {
|
|
t.Fatalf("Claim(reclaimed) = %+v, %v", reclaimed, err)
|
|
}
|
|
if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{
|
|
ConsumerID: "publisher-b", Now: now.Add(2*time.Minute + 2*time.Second), EventIDs: []uint64{1, 2},
|
|
}); err != nil {
|
|
t.Fatalf("Acknowledge(valid): %v", err)
|
|
}
|
|
empty, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
|
ConsumerID: "publisher-c", Now: now.Add(4 * time.Minute), Limit: 10, Lease: time.Minute,
|
|
})
|
|
if err != nil || len(empty) != 0 {
|
|
t.Fatalf("Claim(after ACK) = %+v, %v", empty, err)
|
|
}
|
|
}
|
|
|
|
func runContextContract(t *testing.T, store adminstate.Store) {
|
|
t.Helper()
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
command := configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))
|
|
if _, err := store.CommitConfig(ctx, command); !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("CommitConfig(canceled) error = %v", err)
|
|
}
|
|
if _, err := store.Snapshot(ctx); !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("Snapshot(canceled) error = %v", err)
|
|
}
|
|
if _, err := store.ReadAudit(ctx, adminstate.AuditQuery{Limit: 1}); !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("ReadAudit(canceled) error = %v", err)
|
|
}
|
|
if _, err := store.Claim(ctx, adminstate.ClaimCommand{
|
|
ConsumerID: "publisher-a", Now: contractNow(), Limit: 1, Lease: time.Second,
|
|
}); !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("Claim(canceled) error = %v", err)
|
|
}
|
|
}
|
|
|
|
func configCommand(requestID, version, checksum string) adminstate.CommitConfigCommand {
|
|
return adminstate.CommitConfigCommand{
|
|
RequestID: requestID, Actor: contractActor(), OccurredAt: contractNow(),
|
|
ConfigVersion: version, Checksum: checksum, Source: "configs/proxy-pool.yaml",
|
|
Upstreams: []adminstate.UpstreamDefinition{
|
|
{Name: "provider-a", Enabled: true},
|
|
{Name: "provider-b", Enabled: true},
|
|
},
|
|
Routings: []adminstate.RoutingDefinition{{
|
|
Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"},
|
|
CurrentUpstream: "provider-a",
|
|
}},
|
|
}
|
|
}
|
|
|
|
func contractActor() adminstate.Actor {
|
|
return adminstate.Actor{ID: "admin-a", SourceIP: "192.0.2.10"}
|
|
}
|
|
|
|
func contractNow() time.Time {
|
|
return time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC)
|
|
}
|
|
|
|
func commit(t *testing.T, store adminstate.Store, command adminstate.CommitConfigCommand) {
|
|
t.Helper()
|
|
if result, err := store.CommitConfig(context.Background(), command); err != nil || !result.Changed {
|
|
t.Fatalf("CommitConfig() = %+v, %v", result, err)
|
|
}
|
|
}
|
|
|
|
func upstreamEnabled(snapshot adminstate.Snapshot, name string) bool {
|
|
for _, upstream := range snapshot.Upstreams {
|
|
if upstream.Name == name {
|
|
return upstream.Enabled
|
|
}
|
|
}
|
|
return false
|
|
}
|