fromozuzhouzzz
Deploy grok2api v3.0.11 to HF Spaces
f1dd159
Raw
History Blame Contribute Delete
4.08 kB
package invalidation
import (
"context"
"errors"
"log/slog"
"sync"
"testing"
"time"
"github.com/chenyme/grok2api/backend/internal/domain/account"
"github.com/chenyme/grok2api/backend/internal/repository"
)
type testBus struct {
mu sync.Mutex
published []repository.InvalidationEvent
events []repository.InvalidationEvent
publishErr error
}
func (b *testBus) PublishInvalidation(_ context.Context, event repository.InvalidationEvent) error {
b.mu.Lock()
b.published = append(b.published, event)
b.mu.Unlock()
return b.publishErr
}
func (b *testBus) ListenInvalidations(ctx context.Context, handler func(context.Context, repository.InvalidationEvent) error) error {
for _, event := range b.events {
if err := handler(ctx, event); err != nil {
return err
}
}
return nil
}
func TestNotifyAlwaysAppliesLocalInvalidation(t *testing.T) {
var applied []repository.InvalidationEvent
service := NewService(nil, "local", func(event repository.InvalidationEvent) {
applied = append(applied, event)
}, slog.Default())
service.Notify(context.Background(), repository.InvalidationEvent{
Kind: repository.InvalidationAccountStateChanged, Provider: account.ProviderBuild,
})
if len(applied) != 1 || applied[0].SourceInstance != "local" || applied[0].PublishedAt.IsZero() {
t.Fatalf("local invalidation = %#v", applied)
}
}
func TestNotifyAppliesLocallyWhenRemoteQueueIsFull(t *testing.T) {
var applied int
service := NewService(&testBus{}, "local", func(repository.InvalidationEvent) { applied++ }, slog.Default())
for range cap(service.queue) {
service.queue <- repository.InvalidationEvent{Kind: repository.InvalidationAccountStateChanged}
}
service.Notify(context.Background(), repository.InvalidationEvent{Kind: repository.InvalidationAccountBillingChanged})
if applied != 1 || service.dropped.Load() != 1 {
t.Fatalf("applied=%d dropped=%d", applied, service.dropped.Load())
}
}
func TestRunSubscriberIgnoresInvalidAndSameSourceEvents(t *testing.T) {
bus := &testBus{events: []repository.InvalidationEvent{
{Kind: "unknown", SourceInstance: "remote"},
{Kind: repository.InvalidationAccountStateChanged, Provider: account.ProviderBuild, SourceInstance: "local"},
{Kind: repository.InvalidationAccountQuotaChanged, Provider: account.ProviderWeb, SourceInstance: "remote", Revision: 2},
}}
var applied []repository.InvalidationEvent
service := NewService(bus, "local", func(event repository.InvalidationEvent) {
applied = append(applied, event)
}, slog.Default())
if err := service.RunSubscriber(context.Background()); err != nil {
t.Fatal(err)
}
if len(applied) != 1 || applied[0].Kind != repository.InvalidationAccountQuotaChanged {
t.Fatalf("applied events = %#v", applied)
}
}
func TestRunPublisherDoesNotStopAfterPublishFailure(t *testing.T) {
bus := &testBus{publishErr: errors.New("redis unavailable")}
service := NewService(bus, "local", nil, slog.Default())
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- service.RunPublisher(ctx) }()
service.Notify(context.Background(), repository.InvalidationEvent{Kind: repository.InvalidationAccountStateChanged})
deadline := time.Now().Add(time.Second)
for service.failures.Load() == 0 && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
cancel()
if err := <-done; err != nil {
t.Fatal(err)
}
if service.failures.Load() != 1 {
t.Fatalf("publish failures = %d", service.failures.Load())
}
}
func TestEventKeyKeepsClientKeyInvalidationsDistinct(t *testing.T) {
first := eventKey(repository.InvalidationEvent{Kind: repository.InvalidationClientKeyChanged, ClientKeyID: 1})
second := eventKey(repository.InvalidationEvent{Kind: repository.InvalidationClientKeyChanged, ClientKeyID: 2})
global := eventKey(repository.InvalidationEvent{Kind: repository.InvalidationClientKeyChanged})
if first == second || first == global || second == global {
t.Fatalf("client-key invalidation keys were coalesced: first=%#v second=%#v global=%#v", first, second, global)
}
}