283 lines
10 KiB
Go
283 lines
10 KiB
Go
package redisactivity
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
|
|
"proxy-pool/internal/domain/activitypool"
|
|
extractionDomain "proxy-pool/internal/domain/extraction"
|
|
proxyDomain "proxy-pool/internal/domain/proxy"
|
|
"proxy-pool/internal/platform/credentials"
|
|
)
|
|
|
|
func TestRunScriptPreservesContextCancellation(t *testing.T) {
|
|
t.Parallel()
|
|
client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"})
|
|
t.Cleanup(func() { _ = client.Close() })
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
_, err := runScript(ctx, client, redis.NewScript("return 1"), nil)
|
|
if !errors.Is(err, context.Canceled) || errors.Is(err, extractionDomain.ErrStoreUnavailable) {
|
|
t.Fatalf("runScript() error = %v, want only context cancellation", err)
|
|
}
|
|
}
|
|
|
|
func TestNewRejectsInvalidDependenciesAndOptions(t *testing.T) {
|
|
t.Parallel()
|
|
client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"})
|
|
t.Cleanup(func() { _ = client.Close() })
|
|
credentialStore, err := credentials.NewMemoryStore(10)
|
|
if err != nil {
|
|
t.Fatalf("NewMemoryStore(): %v", err)
|
|
}
|
|
valid := Options{
|
|
Namespace: "test-a", Credentials: credentialStore,
|
|
OperationTTL: time.Minute, MaxCandidateScan: 2048, CleanupLimit: 128,
|
|
}
|
|
var typedNilClient *redis.Client
|
|
var typedNilCredentials *credentials.MemoryStore
|
|
|
|
tests := []struct {
|
|
name string
|
|
client redis.Scripter
|
|
options Options
|
|
}{
|
|
{name: "nil client", options: valid},
|
|
{name: "typed nil client", client: typedNilClient, options: valid},
|
|
{name: "nil credentials", client: client, options: withCredentials(valid, nil)},
|
|
{name: "typed nil credentials", client: client, options: withCredentials(valid, typedNilCredentials)},
|
|
{name: "empty namespace", client: client, options: withNamespace(valid, "")},
|
|
{name: "braces in namespace", client: client, options: withNamespace(valid, "tenant{other}")},
|
|
{name: "colon in namespace", client: client, options: withNamespace(valid, "tenant:other")},
|
|
{name: "zero operation ttl", client: client, options: withOperationTTL(valid, 0)},
|
|
{name: "zero candidate scan", client: client, options: withMaxCandidateScan(valid, 0)},
|
|
{name: "negative runtime counters", client: client, options: withMaxRuntimeCounters(valid, -1)},
|
|
{name: "negative inventory scan", client: client, options: withMaxInventoryScan(valid, -1)},
|
|
{name: "negative cleanup limit", client: client, options: withCleanupLimit(valid, -1)},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
if adapter, err := New(tt.client, tt.options); err == nil || adapter != nil {
|
|
t.Fatalf("New() = (%v, %v), want nil adapter and error", adapter, err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNewBuildsClusterSafeKeyspaceAndHashesDynamicTokens(t *testing.T) {
|
|
t.Parallel()
|
|
client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"})
|
|
t.Cleanup(func() { _ = client.Close() })
|
|
credentialStore, err := credentials.NewMemoryStore(10)
|
|
if err != nil {
|
|
t.Fatalf("NewMemoryStore(): %v", err)
|
|
}
|
|
adapter, err := New(client, Options{
|
|
Namespace: " test-a ", Credentials: credentialStore,
|
|
OperationTTL: time.Minute, MaxCandidateScan: 2048, CleanupLimit: 128,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("New(): %v", err)
|
|
}
|
|
|
|
if got := adapter.keys.records; got != "pp:{activity}:test-a:records" {
|
|
t.Fatalf("records key = %q", got)
|
|
}
|
|
staticKeys := []string{
|
|
adapter.keys.records, adapter.keys.unique, adapter.keys.idkeys,
|
|
adapter.keys.expiry, adapter.keys.available, adapter.keys.owners,
|
|
adapter.keys.ownerExpiry, adapter.keys.epoch, adapter.keys.inventory,
|
|
adapter.keys.stateInventory, adapter.keys.workerSessions,
|
|
adapter.keys.workerSessionExpiry, adapter.keys.workerRuntime,
|
|
adapter.keys.workerRuntimeExpiry,
|
|
}
|
|
for _, key := range staticKeys {
|
|
if strings.Count(key, "{activity}") != 1 || strings.Count(key, "{") != 1 || strings.Count(key, "}") != 1 {
|
|
t.Fatalf("key %q does not contain exactly one fixed hash tag", key)
|
|
}
|
|
}
|
|
|
|
raw := "tenant:{unsafe}:TOKEN"
|
|
dynamicKeys := []string{
|
|
adapter.keys.idempotency(raw, raw),
|
|
adapter.keys.operation(raw),
|
|
adapter.keys.protocol(raw),
|
|
adapter.keys.region(raw),
|
|
adapter.keys.carrier(raw),
|
|
adapter.keys.upstream(raw),
|
|
adapter.keys.owned(raw),
|
|
}
|
|
for _, key := range dynamicKeys {
|
|
if strings.Contains(key, raw) || strings.Count(key, "{activity}") != 1 || strings.Count(key, "{") != 1 {
|
|
t.Fatalf("dynamic key is unsafe: %q", key)
|
|
}
|
|
}
|
|
if got := digestToken(raw); len(got) != 64 || got != digestToken(raw) || strings.Contains(got, raw) {
|
|
t.Fatalf("digestToken() = %q", got)
|
|
}
|
|
}
|
|
|
|
func TestStateInventoryFieldsAreCollisionFreeAndStatusDoesNotScanRecords(t *testing.T) {
|
|
t.Parallel()
|
|
first := stateInventoryField("provider:a", "FETCHED")
|
|
second := stateInventoryField("provider", "a:FETCHED")
|
|
if first == second || first == "" || second == "" {
|
|
t.Fatalf("state inventory fields collide: %q and %q", first, second)
|
|
}
|
|
upper := strings.ToUpper(statusSource)
|
|
if strings.Contains(upper, "HGETALL") || strings.Contains(upper, "HSCAN") {
|
|
t.Fatal("status script scans Redis hashes")
|
|
}
|
|
}
|
|
|
|
func TestReadStateInventoryRejectsInvalidCalls(t *testing.T) {
|
|
t.Parallel()
|
|
now := time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC)
|
|
var adapter *Adapter
|
|
if _, err := adapter.ReadStateInventory(context.Background(), []string{"provider-a"}, now); !errors.Is(err, activitypool.ErrInvalidInventory) {
|
|
t.Fatalf("nil adapter error = %v", err)
|
|
}
|
|
if _, err := adapter.ReadStateInventory(nil, []string{"provider-a"}, now); !errors.Is(err, activitypool.ErrInvalidInventory) {
|
|
t.Fatalf("nil context error = %v", err)
|
|
}
|
|
adapter = &Adapter{}
|
|
if _, err := adapter.ReadStateInventory(context.Background(), []string{""}, now); !errors.Is(err, activitypool.ErrInvalidInventory) {
|
|
t.Fatalf("empty upstream error = %v", err)
|
|
}
|
|
if _, err := adapter.ReadStateInventory(context.Background(), []string{"provider-a"}, time.Time{}); !errors.Is(err, activitypool.ErrInvalidInventory) {
|
|
t.Fatalf("zero time error = %v", err)
|
|
}
|
|
}
|
|
|
|
func TestProxyRecordCodecIsDeterministicStrictAndRedacted(t *testing.T) {
|
|
t.Parallel()
|
|
record := proxyRecord{
|
|
Version: 1, ID: "proxy-a", Scheme: string(proxyDomain.SchemeHTTP),
|
|
Host: "192.0.2.10", Port: 8080, Username: "user", Password: "top-secret",
|
|
SourceUpstream: "provider-a", CreatedAtMS: 1_000, ExpiresAtMS: 61_000,
|
|
UsableUntilMS: 58_000, LastCheckedAtMS: 2_000, LastSuccessAtMS: 2_000,
|
|
LatencyNS: int64(25 * time.Millisecond), MaxConcurrency: 8,
|
|
State: string(proxyDomain.StateAvailable), Tags: map[string]string{"region": "cn", "carrier": "ct"},
|
|
OwnerIndexKey: "pp:{activity}:test:owned:index",
|
|
}
|
|
first, err := encodeProxyRecord(record)
|
|
if err != nil {
|
|
t.Fatalf("encodeProxyRecord(): %v", err)
|
|
}
|
|
second, err := encodeProxyRecord(record)
|
|
if err != nil || first != second {
|
|
t.Fatalf("deterministic encode = (%q, %q, %v)", first, second, err)
|
|
}
|
|
decoded, err := decodeProxyRecord(first)
|
|
if err != nil || decoded.Password != "top-secret" || decoded.State != string(proxyDomain.StateAvailable) {
|
|
t.Fatalf("decodeProxyRecord() = (%+v, %v)", decoded, err)
|
|
}
|
|
if formatted := fmt.Sprintf("%+v", record); strings.Contains(formatted, "top-secret") || !strings.Contains(formatted, "<redacted>") {
|
|
t.Fatalf("proxyRecord formatting leaked password: %s", formatted)
|
|
}
|
|
|
|
invalid := []string{
|
|
strings.Replace(first, `"state":"AVAILABLE"`, `"state":"UNKNOWN"`, 1),
|
|
strings.Replace(first, `"latencyNs":25000000`, `"latencyNs":-1`, 1),
|
|
strings.TrimSuffix(first, "}") + `,"unexpected":true}`,
|
|
}
|
|
for _, payload := range invalid {
|
|
if _, err := decodeProxyRecord(payload); err == nil {
|
|
t.Fatalf("decodeProxyRecord(%s) error = nil", payload)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestOwnershipAndIdempotencyCodecsValidateAndRedact(t *testing.T) {
|
|
t.Parallel()
|
|
assignment := ownershipRecord{
|
|
Version: 1, ProxyID: "proxy-a", WorkerID: "worker-a",
|
|
Epoch: 2, AssignmentVersion: 3, ExpiresAtMS: 5_000, Draining: true,
|
|
}
|
|
payload, err := encodeOwnershipRecord(assignment)
|
|
if err != nil {
|
|
t.Fatalf("encodeOwnershipRecord(): %v", err)
|
|
}
|
|
if decoded, err := decodeOwnershipRecord(payload); err != nil || decoded != assignment {
|
|
t.Fatalf("decodeOwnershipRecord() = (%+v, %v)", decoded, err)
|
|
}
|
|
if _, err := decodeOwnershipRecord(strings.Replace(payload, `"epoch":2`, `"epoch":0`, 1)); err == nil {
|
|
t.Fatal("decodeOwnershipRecord() accepted zero epoch")
|
|
}
|
|
|
|
idempotency := idempotencyRecord{
|
|
Version: 1, RequestDigest: digestToken("request"), ExpiresAtMS: 10_000,
|
|
Result: extractionDomain.Result{
|
|
Requested: 1, Returned: 1, ExtractedAt: time.UnixMilli(1_000).UTC(),
|
|
Items: []extractionDomain.Candidate{{
|
|
ID: "proxy-a", Protocol: "http", Host: "192.0.2.10", Port: 8080,
|
|
Username: "user", Password: "top-secret", Upstream: "provider-a", State: extractionDomain.Extracted,
|
|
ExpiresAt: time.UnixMilli(10_000).UTC(),
|
|
}},
|
|
},
|
|
}
|
|
payload, err = encodeIdempotencyRecord(idempotency)
|
|
if err != nil {
|
|
t.Fatalf("encodeIdempotencyRecord(): %v", err)
|
|
}
|
|
decodedID, err := decodeIdempotencyRecord(payload)
|
|
if err != nil || decodedID.Result.Items[0].Password != "top-secret" {
|
|
t.Fatalf("decodeIdempotencyRecord() = (%+v, %v)", decodedID, err)
|
|
}
|
|
invalidPayloads := []string{
|
|
strings.Replace(payload, `"port":8080`, `"port":65537`, 1),
|
|
strings.Replace(payload, `"state":"EXTRACTED","expiresAtMs":10000`, `"state":"EXTRACTED","expiresAtMs":10000,"lastCheckedAtMs":-1`, 1),
|
|
}
|
|
for _, invalidPayload := range invalidPayloads {
|
|
if _, err := decodeIdempotencyRecord(invalidPayload); err == nil {
|
|
t.Fatalf("decodeIdempotencyRecord(%s) error = nil", invalidPayload)
|
|
}
|
|
}
|
|
if formatted := fmt.Sprintf("%+v", idempotency); strings.Contains(formatted, "top-secret") || !strings.Contains(formatted, "<redacted>") {
|
|
t.Fatalf("idempotencyRecord formatting leaked password: %s", formatted)
|
|
}
|
|
}
|
|
|
|
func withCredentials(options Options, store credentials.Store) Options {
|
|
options.Credentials = store
|
|
return options
|
|
}
|
|
|
|
func withNamespace(options Options, namespace string) Options {
|
|
options.Namespace = namespace
|
|
return options
|
|
}
|
|
|
|
func withOperationTTL(options Options, ttl time.Duration) Options {
|
|
options.OperationTTL = ttl
|
|
return options
|
|
}
|
|
|
|
func withMaxCandidateScan(options Options, limit int) Options {
|
|
options.MaxCandidateScan = limit
|
|
return options
|
|
}
|
|
|
|
func withMaxRuntimeCounters(options Options, limit int) Options {
|
|
options.MaxRuntimeCounters = limit
|
|
return options
|
|
}
|
|
|
|
func withMaxInventoryScan(options Options, limit int) Options {
|
|
options.MaxInventoryScan = limit
|
|
return options
|
|
}
|
|
|
|
func withCleanupLimit(options Options, limit int) Options {
|
|
options.CleanupLimit = limit
|
|
return options
|
|
}
|