pkg/k8score/cache_test.go
Rama main · réplica de lectura del repositorio original.
package k8score
import (
"context"
"errors"
"fmt"
"io"
"log"
"reflect"
"slices"
"sync"
"sync/atomic"
"testing"
"time"
corev1 "k8s.io/api/core/v1"
rbacv1 "k8s.io/api/rbac/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
dynamicfake "k8s.io/client-go/dynamic/fake"
"k8s.io/client-go/kubernetes/fake"
k8stesting "k8s.io/client-go/testing"
"k8s.io/client-go/tools/cache"
)
func TestNewResourceCache_Basic(t *testing.T) {
client := fake.NewSimpleClientset()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Services: true,
Deployments: true,
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
if !rc.IsSyncComplete() {
t.Error("expected IsSyncComplete() = true after NewResourceCache returns")
}
if rc.Pods() == nil {
t.Error("expected Pods() lister to be non-nil")
}
if rc.Services() == nil {
t.Error("expected Services() lister to be non-nil")
}
if rc.Deployments() == nil {
t.Error("expected Deployments() lister to be non-nil")
}
// Disabled resources should return nil listers
if rc.Secrets() != nil {
t.Error("expected Secrets() lister to be nil (not enabled)")
}
if rc.Nodes() != nil {
t.Error("expected Nodes() lister to be nil (not enabled)")
}
if !rc.IsKindClusterWide(Pods) {
t.Error("legacy cluster-wide ResourceTypes config must report cluster-wide authority")
}
if got := rc.KindNamespaces(Pods); got != nil {
t.Fatalf("cluster-wide Pod namespaces = %v, want nil", got)
}
}
func TestNewResourceCache_DeferredSync(t *testing.T) {
client := fake.NewSimpleClientset()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Services: true,
ConfigMaps: true,
Secrets: true,
},
DeferredTypes: map[string]bool{
ConfigMaps: true,
Secrets: true,
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Critical resources should be available immediately
if rc.Pods() == nil {
t.Error("expected Pods() lister to be available (critical)")
}
// Wait for deferred to complete (fake client syncs immediately)
select {
case <-rc.DeferredDone():
case <-time.After(5 * time.Second):
t.Fatal("timed out waiting for deferred sync")
}
if !rc.IsDeferredSynced() {
t.Error("expected IsDeferredSynced() = true")
}
if rc.ConfigMaps() == nil {
t.Error("expected ConfigMaps() lister to be available after deferred sync")
}
if rc.Secrets() == nil {
t.Error("expected Secrets() lister to be available after deferred sync")
}
}
// TestNewResourceCache_DeferredSync_PartialFailure verifies that a permanently
// failing deferred informer (e.g. HPA autoscaling/v2 on a K8s <1.23 cluster,
// which responds with "the server could not find the requested resource")
// does not block sibling deferred informers from becoming ready. It also
// verifies the DeferredSyncTimeout path flips deferredFailed so stragglers
// return false from IsDeferredPending.
func TestNewResourceCache_DeferredSync_PartialFailure(t *testing.T) {
client := fake.NewSimpleClientset()
// Make HPA LIST fail forever, as happens when the v2 API isn't served.
client.PrependReactor("list", "horizontalpodautoscalers", func(action k8stesting.Action) (bool, runtime.Object, error) {
return true, nil, fmt.Errorf("the server could not find the requested resource")
})
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
ConfigMaps: true,
Secrets: true,
HorizontalPodAutoscalers: true,
},
DeferredTypes: map[string]bool{
ConfigMaps: true,
Secrets: true,
HorizontalPodAutoscalers: true,
},
DeferredSyncTimeout: 500 * time.Millisecond,
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Poll for ConfigMaps and Secrets to become ready. A failing sibling
// (HPA) must not block them.
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if rc.ConfigMaps() != nil && rc.Secrets() != nil {
break
}
time.Sleep(10 * time.Millisecond)
}
if rc.ConfigMaps() == nil {
t.Fatal("expected ConfigMaps() to become ready despite sibling HPA failing")
}
if rc.Secrets() == nil {
t.Fatal("expected Secrets() to become ready despite sibling HPA failing")
}
// Pre-timeout contract check: while HPA is still stuck and the deadline
// hasn't fired, IsDeferredPending must report HPA pending (HTTP handlers
// return 503) and ConfigMaps not-pending (handlers serve data). This is
// the 503-vs-403 distinction the fix is built around.
if !rc.IsDeferredPending(HorizontalPodAutoscalers) {
t.Error("pre-timeout: expected IsDeferredPending(HPA)=true while informer still stuck")
}
if rc.IsDeferredPending(ConfigMaps) {
t.Error("pre-timeout: ConfigMaps synced, expected IsDeferredPending=false")
}
// deferredDone must close even though HPA never syncs — otherwise the
// SSE warmup completion never fires.
select {
case <-rc.DeferredDone():
case <-time.After(3 * time.Second):
t.Fatal("deferredDone never closed after DeferredSyncTimeout")
}
// Post-timeout: HPA flips from pending to not-pending because
// deferredFailed is now set — stops the perpetual-503 spinner.
// ConfigMaps stays not-pending (it was already synced).
if rc.IsDeferredPending(HorizontalPodAutoscalers) {
t.Error("post-timeout: expected IsDeferredPending(HPA)=false (deferredFailed signals give-up)")
}
if rc.IsDeferredPending(ConfigMaps) {
t.Error("post-timeout: ConfigMaps synced, expected IsDeferredPending=false")
}
}
// TestNewResourceCache_MinimalSet_AllFast verifies that when the patience
// window is set but every informer syncs quickly, NewResourceCache returns
// via the all-critical-synced path (not the minimal-set fallback). No
// informers should be promoted, syncProgress should fire to completion.
func TestNewResourceCache_MinimalSet_AllFast(t *testing.T) {
client := fake.NewSimpleClientset()
var lastSynced, lastTotal int
var lastReady bool
var progMu sync.Mutex
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Services: true,
Deployments: true,
Ingresses: true,
},
PatienceWindow: 2 * time.Second,
MinimalSet: map[string]bool{
Pods: true,
Services: true,
},
SyncProgress: func(synced, total int, ready bool) {
progMu.Lock()
lastSynced, lastTotal, lastReady = synced, total, ready
progMu.Unlock()
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// All four critical informers should be synced — no promotion.
if got := rc.PromotedKinds(); len(got) != 0 {
t.Errorf("expected no promoted kinds when all critical synced fast, got %v", got)
}
progMu.Lock()
defer progMu.Unlock()
if lastTotal != 4 {
t.Errorf("expected SyncProgress.total=4, got %d", lastTotal)
}
if lastSynced != 4 {
t.Errorf("expected SyncProgress.synced=4, got %d", lastSynced)
}
if !lastReady {
t.Error("expected final SyncProgress to report minimalReady=true")
}
}
// TestNewResourceCache_MinimalSet_Promotion verifies the slow-cluster path:
// when a non-minimal critical informer can't sync within the patience
// window but minimal-set members are ready, NewResourceCache returns and
// promotes the slow informer to deferred. The promoted informer must
// continue running and eventually become available.
func TestNewResourceCache_MinimalSet_Promotion(t *testing.T) {
client := fake.NewSimpleClientset()
// Make ingresses LIST fail. The reflector retries with backoff, so
// HasSynced() stays false, but the failure returns immediately —
// no shared tracker lock held, pods/services LIST proceed normally.
// Ingress is critical (not in MinimalSet) so it should be the one
// that gets promoted.
client.PrependReactor("list", "ingresses", func(action k8stesting.Action) (bool, runtime.Object, error) {
return true, nil, fmt.Errorf("simulated slow API: ingresses unavailable")
})
start := time.Now()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Services: true,
Ingresses: true,
},
PatienceWindow: 200 * time.Millisecond,
MinimalSet: map[string]bool{
Pods: true,
Services: true,
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
elapsed := time.Since(start)
if elapsed < 200*time.Millisecond {
t.Errorf("expected return after at least the patience window (200ms), got %v", elapsed)
}
// Cap upper bound generously: patience window + a few jitter ticks.
if elapsed > 3*time.Second {
t.Errorf("returned far later than patience window — minimal-set fallback didn't trigger? elapsed=%v", elapsed)
}
promoted := rc.PromotedKinds()
if len(promoted) != 1 || promoted[0] != "Ingress" {
t.Errorf("expected Ingress to be promoted, got %v", promoted)
}
// Minimal-set listers should be available immediately.
if rc.Pods() == nil {
t.Error("expected Pods() lister available after first paint")
}
if rc.Services() == nil {
t.Error("expected Services() lister available after first paint")
}
}
// TestNewResourceCache_SyncTimeout_Promotion covers the legacy hard-cap
// path used by skyhook-connector. Without PatienceWindow/MinimalSet, a
// stuck critical informer must promote at SyncTimeout and the cache
// must return.
func TestNewResourceCache_SyncTimeout_Promotion(t *testing.T) {
client := fake.NewSimpleClientset()
client.PrependReactor("list", "ingresses", func(action k8stesting.Action) (bool, runtime.Object, error) {
return true, nil, fmt.Errorf("simulated stuck API")
})
start := time.Now()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Ingresses: true,
},
SyncTimeout: 200 * time.Millisecond,
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
elapsed := time.Since(start)
if elapsed < 200*time.Millisecond {
t.Errorf("expected return after SyncTimeout (200ms), got %v", elapsed)
}
if elapsed > 3*time.Second {
t.Errorf("returned far later than SyncTimeout — promotion didn't fire? elapsed=%v", elapsed)
}
promoted := rc.PromotedKinds()
if len(promoted) != 1 || promoted[0] != "Ingress" {
t.Errorf("expected Ingress to be promoted, got %v", promoted)
}
}
// TestNewResourceCache_MinimalSet_UnknownKey verifies the validation
// log path: a typo or kind not enabled in ResourceTypes results in an
// empty effective minimal set; the cache returns at PatienceWindow with
// nothing meaningfully gating first paint. This is intentionally
// permissive (we don't fail construction) but the operator must see a
// warning. We don't capture log output here — just verify the cache
// still returns and shape is consistent.
func TestNewResourceCache_MinimalSet_UnknownKey(t *testing.T) {
client := fake.NewSimpleClientset()
start := time.Now()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
},
PatienceWindow: 200 * time.Millisecond,
MinimalSet: map[string]bool{
"pod": true, // typo — should be "pods"
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Pods syncs instantly on fake client, so allCritical fires before
// patience even elapses — promoted should be empty.
if got := rc.PromotedKinds(); len(got) != 0 {
t.Errorf("expected no promoted kinds (everything synced), got %v", got)
}
if rc.Pods() == nil {
t.Error("expected Pods() lister to be available")
}
// Sanity: returned in reasonable time despite the bogus minimal-set key
if elapsed := time.Since(start); elapsed > 3*time.Second {
t.Errorf("returned far too late with bogus MinimalSet key: %v", elapsed)
}
}
// TestNewResourceCache_PatienceWindow_WithoutMinimalSet verifies the
// edge case where only PatienceWindow is set: useMinimalSet is false,
// so behavior degrades to "wait indefinitely for all critical" with no
// hard cap. With a fake client that syncs instantly, this should just
// return on the all-synced path.
func TestNewResourceCache_PatienceWindow_WithoutMinimalSet(t *testing.T) {
client := fake.NewSimpleClientset()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
},
PatienceWindow: 100 * time.Millisecond,
// MinimalSet intentionally nil
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
if got := rc.PromotedKinds(); len(got) != 0 {
t.Errorf("expected no promotion without MinimalSet, got %v", got)
}
if rc.Pods() == nil {
t.Error("expected Pods() lister to be available")
}
}
// TestPendingPromotedKinds verifies the live-filtered accessor that the
// dashboard banner relies on: starts equal to PromotedKinds, and shrinks
// as informers eventually sync.
func TestPendingPromotedKinds(t *testing.T) {
client := fake.NewSimpleClientset()
client.PrependReactor("list", "ingresses", func(action k8stesting.Action) (bool, runtime.Object, error) {
return true, nil, fmt.Errorf("simulated slow API")
})
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Services: true,
Ingresses: true,
},
PatienceWindow: 200 * time.Millisecond,
MinimalSet: map[string]bool{
Pods: true,
Services: true,
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// At first paint, Ingress is both Promoted and Pending.
if got := rc.PromotedKinds(); len(got) != 1 || got[0] != "Ingress" {
t.Fatalf("PromotedKinds: expected [Ingress], got %v", got)
}
if got := rc.PendingPromotedKinds(); len(got) != 1 || got[0] != "Ingress" {
t.Errorf("PendingPromotedKinds: expected [Ingress], got %v", got)
}
// PromotedKinds is a stable historical snapshot, PendingPromotedKinds
// is the live view — verify they are distinct concepts and don't
// share backing storage.
promoted := rc.PromotedKinds()
pending := rc.PendingPromotedKinds()
if len(promoted) > 0 && len(pending) > 0 && &promoted[0] == &pending[0] {
t.Error("PromotedKinds and PendingPromotedKinds must not share backing array")
}
}
// TestPendingPromotedKinds_Drains verifies the live-filtering claim: as
// a promoted informer eventually catches up and reports HasSynced=true,
// it leaves PendingPromotedKinds. PromotedKinds (the snapshot) does not
// change. Without this, a UI banner would list kinds forever even after
// they finished loading.
func TestPendingPromotedKinds_Drains(t *testing.T) {
client := fake.NewSimpleClientset()
// Toggle: fail Ingress LIST until we flip the flag, then succeed.
// The reflector retries on backoff so HasSynced flips when LIST
// stops failing.
var failIngress atomic.Bool
failIngress.Store(true)
client.PrependReactor("list", "ingresses", func(action k8stesting.Action) (bool, runtime.Object, error) {
if failIngress.Load() {
return true, nil, fmt.Errorf("simulated transient failure")
}
return false, nil, nil // pass through to default tracker
})
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Services: true,
Ingresses: true,
},
PatienceWindow: 200 * time.Millisecond,
MinimalSet: map[string]bool{
Pods: true,
Services: true,
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Sanity: at construction, Ingress is pending.
if got := rc.PendingPromotedKinds(); len(got) != 1 || got[0] != "Ingress" {
t.Fatalf("PendingPromotedKinds at start: expected [Ingress], got %v", got)
}
// Flip the reactor to succeed, then poll for the live view to drain.
failIngress.Store(false)
deadline := time.Now().Add(15 * time.Second)
for time.Now().Before(deadline) {
if len(rc.PendingPromotedKinds()) == 0 {
break
}
time.Sleep(50 * time.Millisecond)
}
if got := rc.PendingPromotedKinds(); len(got) != 0 {
t.Errorf("PendingPromotedKinds did not drain after Ingress LIST began succeeding; still pending: %v", got)
}
// PromotedKinds is the snapshot — must NOT shrink.
if got := rc.PromotedKinds(); len(got) != 1 || got[0] != "Ingress" {
t.Errorf("PromotedKinds (snapshot) should not shrink; expected [Ingress], got %v", got)
}
}
// TestNewResourceCache_MinimalSet_BackstopFires covers the worst case on
// the patience+minimal-set path: a kind that's IN the minimal set never
// syncs. Without a backstop the cache would block in NewResourceCache
// forever, trapping the caller on a connecting screen. SyncTimeout is
// the safety net — it must promote everything still pending (including
// minimal-set members) and let the cache return.
func TestNewResourceCache_MinimalSet_BackstopFires(t *testing.T) {
client := fake.NewSimpleClientset()
// Pods is in MinimalSet — make its LIST fail forever.
client.PrependReactor("list", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
return true, nil, fmt.Errorf("simulated permanently-stuck API")
})
start := time.Now()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
Services: true,
},
PatienceWindow: 100 * time.Millisecond,
MinimalSet: map[string]bool{
Pods: true,
Services: true,
},
SyncTimeout: 500 * time.Millisecond,
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
elapsed := time.Since(start)
if elapsed < 500*time.Millisecond {
t.Errorf("expected return after at least SyncTimeout (500ms), got %v", elapsed)
}
if elapsed > 3*time.Second {
t.Errorf("returned far later than SyncTimeout — backstop didn't fire? elapsed=%v", elapsed)
}
// Pods stuck → must be promoted; Services synced → must not be promoted.
promoted := rc.PromotedKinds()
if len(promoted) != 1 || promoted[0] != "Pod" {
t.Errorf("expected only Pod to be promoted by backstop, got %v", promoted)
}
// Lister still available (it's just empty).
if rc.Pods() == nil {
t.Error("expected Pods() lister to be available even after backstop promotion")
}
}
func TestNewResourceCache_Callbacks(t *testing.T) {
// Pre-create a pod so the informer fires an add event
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test-pod",
Namespace: "default",
UID: "test-uid",
},
}
client := fake.NewSimpleClientset(pod)
var mu sync.Mutex
var changes []ResourceChange
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
},
OnChange: func(change ResourceChange, obj, oldObj any) {
mu.Lock()
changes = append(changes, change)
mu.Unlock()
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Wait for the informer add event to propagate
time.Sleep(200 * time.Millisecond)
mu.Lock()
got := len(changes)
mu.Unlock()
if got == 0 {
t.Error("expected OnChange to be called for the pre-existing pod add")
}
}
func TestNewResourceCache_SuppressInitialAdds(t *testing.T) {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "existing-pod",
Namespace: "default",
UID: "uid-1",
},
}
client := fake.NewSimpleClientset(pod)
var mu sync.Mutex
var callbackChanges []ResourceChange
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
},
SuppressInitialAdds: true,
OnChange: func(change ResourceChange, obj, oldObj any) {
mu.Lock()
callbackChanges = append(callbackChanges, change)
mu.Unlock()
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Wait briefly for any events
time.Sleep(200 * time.Millisecond)
mu.Lock()
got := len(callbackChanges)
mu.Unlock()
// With SuppressInitialAdds, the OnChange callback should NOT fire for
// pre-existing resources during sync. However, the add events still
// go to the changes channel. Note: since NewResourceCache blocks until
// sync completes, the add fires DURING construction when syncComplete
// is still false, so callback should be suppressed.
if got != 0 {
t.Errorf("expected 0 callback changes with SuppressInitialAdds, got %d", got)
}
}
func TestNewResourceCache_OnReceived(t *testing.T) {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test-pod",
Namespace: "default",
UID: "test-uid",
},
}
client := fake.NewSimpleClientset(pod)
var mu sync.Mutex
receivedKinds := map[string]int{}
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
},
OnReceived: func(kind string) {
mu.Lock()
receivedKinds[kind]++
mu.Unlock()
},
// Even with noisy filter that always returns true, OnReceived should fire
IsNoisyResource: func(kind, name, op string) bool {
return true // everything is noisy
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
time.Sleep(200 * time.Millisecond)
mu.Lock()
podCount := receivedKinds["Pod"]
mu.Unlock()
if podCount == 0 {
t.Error("expected OnReceived to fire even when IsNoisyResource returns true")
}
}
func TestNewResourceCache_OnObservedChangeSurvivesFiltering(t *testing.T) {
pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
Name: "test-pod", Namespace: "default", UID: "test-uid",
}}
client := fake.NewSimpleClientset(pod)
var mu sync.Mutex
var observed, delivered []ResourceChange
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{Pods: true},
IsNoisyResource: func(string, string, string) bool {
return true
},
OnObservedChange: func(change ResourceChange, _, _ any) {
mu.Lock()
observed = append(observed, change)
mu.Unlock()
},
OnChange: func(change ResourceChange, _, _ any) {
mu.Lock()
delivered = append(delivered, change)
mu.Unlock()
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
time.Sleep(200 * time.Millisecond)
mu.Lock()
defer mu.Unlock()
if len(observed) != 1 || observed[0].Kind != "Pod" || observed[0].Name != "test-pod" {
t.Fatalf("observed changes = %+v, want filtered Pod add", observed)
}
if len(delivered) != 0 {
t.Fatalf("OnChange received filtered changes: %+v", delivered)
}
}
func TestNewResourceCache_NamespaceScopedValidation(t *testing.T) {
client := fake.NewSimpleClientset()
_, err := NewResourceCache(CacheConfig{
Client: client,
NamespaceScoped: true,
Namespace: "", // empty namespace with NamespaceScoped=true
ResourceTypes: map[string]bool{Pods: true},
})
if err == nil {
t.Fatal("expected error when NamespaceScoped=true with empty Namespace")
}
}
func TestNewResourceCache_CallbackPanicRecovery(t *testing.T) {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "panic-pod",
Namespace: "default",
UID: "panic-uid",
},
}
client := fake.NewSimpleClientset(pod)
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
},
OnChange: func(change ResourceChange, obj, oldObj any) {
panic("test panic in OnChange")
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Wait for event to fire — the panic should be recovered, not crash
time.Sleep(200 * time.Millisecond)
// If we get here without crashing, the test passes
if rc.Pods() == nil {
t.Error("expected Pods() lister to still work after callback panic")
}
}
func TestNewResourceCache_MapCloning(t *testing.T) {
client := fake.NewSimpleClientset()
resourceTypes := map[string]bool{
Pods: true,
Services: true,
}
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: resourceTypes,
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
// Mutate the original map after construction
resourceTypes[Pods] = false
resourceTypes["bogus"] = true
// The cache should not be affected
if rc.Pods() == nil {
t.Error("expected Pods() lister to still work after caller mutates resourceTypes map")
}
enabled := rc.GetEnabledResources()
if !enabled[Pods] {
t.Error("expected Pods to still be enabled after caller mutates resourceTypes map")
}
}
func TestNewResourceCache_NilClient(t *testing.T) {
_, err := NewResourceCache(CacheConfig{
Client: nil,
})
if err == nil {
t.Fatal("expected error with nil client")
}
}
func TestNewResourceCache_NoEnabledResources(t *testing.T) {
client := fake.NewSimpleClientset()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{},
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
defer rc.Stop()
if !rc.IsSyncComplete() {
t.Error("expected IsSyncComplete() = true even with no resources")
}
if rc.GetResourceCount() != 0 {
t.Errorf("expected 0 resource count, got %d", rc.GetResourceCount())
}
}
func TestNewResourceCache_StopLifecycle(t *testing.T) {
client := fake.NewSimpleClientset()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{
Pods: true,
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
// Stop should be safe to call multiple times
rc.Stop()
rc.Stop()
// Methods should be safe to call after stop
_ = rc.Pods()
_ = rc.Changes()
_ = rc.GetResourceCount()
}
func TestDropManagedFields(t *testing.T) {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test",
Namespace: "default",
ManagedFields: []metav1.ManagedFieldsEntry{
{Manager: "test-manager"},
},
Annotations: map[string]string{
"kubectl.kubernetes.io/last-applied-configuration": `{"big":"json"}`,
"keep-this": "yes",
},
},
}
result, err := DropManagedFields(pod)
if err != nil {
t.Fatalf("DropManagedFields failed: %v", err)
}
p := result.(*corev1.Pod)
if len(p.ManagedFields) != 0 {
t.Error("expected managedFields to be nil/empty")
}
if _, ok := p.Annotations["kubectl.kubernetes.io/last-applied-configuration"]; ok {
t.Error("expected last-applied-configuration annotation to be removed")
}
if p.Annotations["keep-this"] != "yes" {
t.Error("expected other annotations to be preserved")
}
}
func TestResourceCacheOnTransformSeesManagedFieldsBeforeStripping(t *testing.T) {
dataWrite := time.Date(2026, 7, 23, 8, 0, 0, 0, time.UTC)
secret := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "credentials",
Namespace: "default",
ManagedFields: []metav1.ManagedFieldsEntry{
{
Manager: "secret-controller",
Operation: metav1.ManagedFieldsOperationUpdate,
Time: &metav1.Time{Time: dataWrite},
FieldsType: "FieldsV1",
FieldsV1: &metav1.FieldsV1{Raw: []byte(`{"f:data":{"f:password":{}}}`)},
},
},
},
Data: map[string][]byte{"password": []byte("must-not-leak")},
}
client := fake.NewSimpleClientset(secret)
var captured time.Time
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceTypes: map[string]bool{Secrets: true},
OnTransform: func(obj any) {
if transformed, ok := obj.(*corev1.Secret); ok && len(transformed.ManagedFields) == 1 {
captured = transformed.ManagedFields[0].Time.Time
}
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
if !captured.Equal(dataWrite) {
t.Fatalf("OnTransform captured %v, want %v", captured, dataWrite)
}
cached, err := rc.Secrets().Secrets("default").Get("credentials")
if err != nil {
t.Fatalf("cached Secret lookup failed: %v", err)
}
if len(cached.ManagedFields) != 0 {
t.Fatalf("cached Secret leaked %d managedFields entries", len(cached.ManagedFields))
}
}
func TestDropManagedFields_EventSeries(t *testing.T) {
now := time.Now()
event := &corev1.Event{
EventTime: metav1.NewMicroTime(now.Add(-time.Hour)),
Series: &corev1.EventSeries{Count: 42, LastObservedTime: metav1.NewMicroTime(now)},
}
result, err := DropManagedFields(event)
if err != nil {
t.Fatal(err)
}
got := result.(*corev1.Event)
if got.EventTime != event.EventTime || got.Series == nil || *got.Series != *event.Series {
t.Fatalf("event occurrence evidence stripped: %+v", got)
}
}
func TestDropManagedFields_Event(t *testing.T) {
event := &corev1.Event{
ObjectMeta: metav1.ObjectMeta{
Name: "test-event",
Namespace: "default",
UID: "event-uid",
ManagedFields: []metav1.ManagedFieldsEntry{
{Manager: "test-manager"},
},
Labels: map[string]string{"should": "be-stripped"},
},
Reason: "Created",
Message: "Pod created",
Type: "Normal",
Count: 1,
}
result, err := DropManagedFields(event)
if err != nil {
t.Fatalf("DropManagedFields failed: %v", err)
}
e := result.(*corev1.Event)
if e.Reason != "Created" {
t.Errorf("expected Reason=Created, got %s", e.Reason)
}
if e.Labels != nil {
t.Error("expected Labels to be stripped from event")
}
if len(e.ManagedFields) != 0 {
t.Error("expected managedFields to be stripped from event")
}
}
// TestNewResourceCache_ResourceScopesMixed verifies a mixed-scope user
// (some kinds cluster-wide, some kinds namespace-scoped) ends up with
// informers wired into the right factories. The fake client lets every
// list succeed; the assertion is structural — that listers exist for
// every enabled kind and the per-namespace factory is created.
func TestNewResourceCache_ResourceScopesMixed(t *testing.T) {
const ns = "dev-ns-1"
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: ns, UID: "u"},
}
client := fake.NewSimpleClientset(pod)
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceScopes: map[string]ResourceScope{
Pods: {Enabled: true, Namespace: ns}, // namespace-scoped
Deployments: {Enabled: true, Namespace: ns}, // namespace-scoped
Nodes: {Enabled: true, Namespace: ns}, // cluster-scoped kinds ignore namespace fallback
Services: {Enabled: false}, // denied — no informer
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
if rc.Pods() == nil {
t.Error("Pods lister should exist for an enabled kind")
}
if rc.Deployments() == nil {
t.Error("Deployments lister should exist for an enabled kind")
}
if rc.Nodes() == nil {
t.Error("Nodes lister should exist for an enabled kind")
}
if rc.Services() != nil {
t.Error("Services lister should be nil — kind was disabled")
}
if !rc.IsKindClusterWide(Nodes) || rc.KindNamespaces(Nodes) != nil {
t.Fatal("cluster-scoped Node informer did not report its effective cluster-wide scope")
}
enabled := rc.GetEnabledResources()
if !enabled[Pods] || !enabled[Deployments] || !enabled[Nodes] {
t.Errorf("enabled map missing kinds: %+v", enabled)
}
if enabled[Services] {
t.Errorf("disabled Services should not appear in enabled map")
}
// Internal: a namespace-scoped factory should have been created for
// the namespace used by Pods/Deployments.
if _, ok := rc.nsFactories[ns]; !ok {
t.Errorf("expected nsFactories to contain %q, got keys %v", ns, mapKeys(rc.nsFactories))
}
}
func TestNewResourceCache_ResourceScopeNamespacesUnion(t *testing.T) {
const nsA, nsB, nsOther = "team-a", "team-b", "team-c"
client := fake.NewSimpleClientset(
&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-a", Namespace: nsA, UID: "pod-a"}},
&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-b", Namespace: nsB, UID: "pod-b"}},
&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-c", Namespace: nsOther, UID: "pod-c"}},
)
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceScopes: map[string]ResourceScope{
Pods: {Enabled: true, Namespace: nsA},
},
ResourceScopeNamespaces: map[string][]string{
Pods: {nsA, nsB},
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
lister := rc.Pods()
if lister == nil {
t.Fatal("Pods lister should exist")
}
all, err := lister.List(labels.Everything())
if err != nil {
t.Fatalf("list all pods: %v", err)
}
if len(all) != 2 {
t.Fatalf("all pods = %d, want 2 from %s/%s (items=%v)", len(all), nsA, nsB, all)
}
for _, tc := range []struct {
namespace string
want int
}{
{nsA, 1},
{nsB, 1},
{nsOther, 0},
} {
items, err := lister.Pods(tc.namespace).List(labels.Everything())
if err != nil {
t.Fatalf("list pods in %s: %v", tc.namespace, err)
}
if len(items) != tc.want {
t.Fatalf("pods in %s = %d, want %d", tc.namespace, len(items), tc.want)
}
}
if got := ListCountNamespaced(lister, []string{nsA, nsB}); got != 2 {
t.Fatalf("ListCountNamespaced = %d, want 2", got)
}
if rc.IsKindClusterWide(Pods) {
t.Fatal("multi-namespace scoped Pods must not report cluster-wide authority")
}
if got := rc.KindNamespaces(Pods); !slices.Equal(got, []string{nsA, nsB}) {
t.Fatalf("Pod informer namespaces = %v, want [%s %s]", got, nsA, nsB)
}
got := rc.KindNamespaces(Pods)
got[0] = "mutated"
if next := rc.KindNamespaces(Pods); !slices.Equal(next, []string{nsA, nsB}) {
t.Fatalf("caller mutation changed Pod informer namespaces: %v", next)
}
}
func TestUnionIndexer_ReadFanout(t *testing.T) {
const nsA, nsB = "team-a", "team-b"
mkIndexer := func(pods ...*corev1.Pod) cache.Indexer {
idx := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
for _, p := range pods {
if err := idx.Add(p); err != nil {
t.Fatalf("seed indexer: %v", err)
}
}
return idx
}
podA := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-a", Namespace: nsA, UID: "pod-a"}}
podB := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-b", Namespace: nsB, UID: "pod-b"}}
u := &unionIndexer{indexers: []cache.Indexer{mkIndexer(podA), mkIndexer(podB)}}
if got := len(u.List()); got != 2 {
t.Fatalf("List() = %d items, want 2", got)
}
if got := len(u.ListKeys()); got != 2 {
t.Fatalf("ListKeys() = %d, want 2", got)
}
for key, want := range map[string]bool{nsA + "/pod-a": true, nsB + "/pod-b": true, "other/pod-x": false} {
_, exists, err := u.GetByKey(key)
if err != nil {
t.Fatalf("GetByKey(%s): %v", key, err)
}
if exists != want {
t.Fatalf("GetByKey(%s) exists = %v, want %v", key, exists, want)
}
}
byNs, err := u.ByIndex(cache.NamespaceIndex, nsB)
if err != nil {
t.Fatalf("ByIndex: %v", err)
}
if len(byNs) != 1 || byNs[0].(*corev1.Pod).Name != "pod-b" {
t.Fatalf("ByIndex(%s) = %v, want [pod-b]", nsB, byNs)
}
if vals := u.ListIndexFuncValues(cache.NamespaceIndex); len(vals) != 2 {
t.Fatalf("ListIndexFuncValues = %v, want two namespaces", vals)
}
}
func TestUnionIndexer_WritesRejected(t *testing.T) {
u := &unionIndexer{indexers: []cache.Indexer{cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)}}
pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "ns"}}
for name, err := range map[string]error{
"Add": u.Add(pod),
"Update": u.Update(pod),
"Delete": u.Delete(pod),
"Replace": u.Replace(nil, ""),
"Resync": u.Resync(),
} {
if err == nil {
t.Fatalf("%s on read-only union indexer returned nil error", name)
}
}
if got := len(u.List()); got != 0 {
t.Fatalf("rejected writes still stored %d items", got)
}
}
// TestNewResourceCache_ResourceScopesEmpty verifies that an explicitly
// empty ResourceScopes map (no kinds enabled) produces a cache with no
// informers, vs nil ResourceScopes which falls through to the legacy
// ResourceTypes path.
func TestNewResourceCache_ResourceScopesEmpty(t *testing.T) {
client := fake.NewSimpleClientset()
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceScopes: map[string]ResourceScope{}, // explicitly empty
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
if rc.Pods() != nil {
t.Error("expected no Pods lister when ResourceScopes is empty")
}
}
func TestNewResourceCache_RbacResources(t *testing.T) {
client := fake.NewSimpleClientset(
&rbacv1.Role{ObjectMeta: metav1.ObjectMeta{Name: "reader", Namespace: "dev"}},
&rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{Name: "cluster-reader"}},
&rbacv1.RoleBinding{ObjectMeta: metav1.ObjectMeta{Name: "reader-bind", Namespace: "dev"}},
&rbacv1.ClusterRoleBinding{ObjectMeta: metav1.ObjectMeta{Name: "cluster-reader-bind"}},
)
rc, err := NewResourceCache(CacheConfig{
Client: client,
ResourceScopes: map[string]ResourceScope{
Roles: {Enabled: true, Namespace: "dev"},
ClusterRoles: {Enabled: true},
RoleBindings: {Enabled: true, Namespace: "dev"},
ClusterRoleBindings: {Enabled: true},
},
})
if err != nil {
t.Fatalf("NewResourceCache failed: %v", err)
}
defer rc.Stop()
if rc.Roles() == nil || rc.ClusterRoles() == nil || rc.RoleBindings() == nil || rc.ClusterRoleBindings() == nil {
t.Fatalf("expected all RBAC listers to be non-nil")
}
counts := rc.GetKindObjectCounts()
for _, kind := range []string{"Role", "ClusterRole", "RoleBinding", "ClusterRoleBinding"} {
if counts[kind] == 0 {
t.Fatalf("expected %s to be counted, got counts=%v", kind, counts)
}
}
enabled := rc.GetEnabledResources()
for _, key := range []string{Roles, ClusterRoles, RoleBindings, ClusterRoleBindings} {
if !enabled[key] {
t.Fatalf("expected %s to be enabled, got %v", key, enabled)
}
}
}
func mapKeys[K comparable, V any](m map[K]V) []K {
keys := make([]K, 0, len(m))
for k := range m {
keys = append(keys, k)
}
return keys
}
// TestRecordWatchError verifies that the WatchErrorHandler hook updates the
// per-informer status when a 403 surfaces from the reflector. Validates the
// observable behavior of the safety net rather than relying on simulating a
// full reflector watch failure (which is racy with the fake client).
func TestRecordWatchError(t *testing.T) {
rc := &ResourceCache{
stdlog: log.New(io.Discard, "", 0),
informerStatuses: []InformerSyncStatus{
{Kind: "Pod", Key: Pods},
{Kind: "Service", Key: Services},
},
}
rc.recordWatchError(0, Pods, "Pod", apierrors.NewForbidden(
schema.GroupResource{Resource: "pods"}, "", nil,
))
rc.informerMu.RLock()
defer rc.informerMu.RUnlock()
pod := rc.informerStatuses[0]
if !pod.ForbiddenSeen {
t.Errorf("ForbiddenSeen should be true after a 403")
}
if pod.LastError == "" {
t.Errorf("LastError should be set")
}
if pod.LastErrorAt == "" {
t.Errorf("LastErrorAt should be set")
}
// Untouched informer should remain pristine.
svc := rc.informerStatuses[1]
if svc.ForbiddenSeen || svc.LastError != "" {
t.Errorf("Service status should not be affected by Pod's 403, got %+v", svc)
}
}
func TestRecordWatchError_TransientNotForbidden(t *testing.T) {
rc := &ResourceCache{
stdlog: log.New(io.Discard, "", 0),
informerStatuses: []InformerSyncStatus{
{Kind: "Pod", Key: Pods},
},
}
rc.recordWatchError(0, Pods, "Pod", errors.New("dial tcp: connection refused"))
rc.informerMu.RLock()
defer rc.informerMu.RUnlock()
if rc.informerStatuses[0].ForbiddenSeen {
t.Errorf("ForbiddenSeen should NOT be set for a transient connection error")
}
if rc.informerStatuses[0].LastError == "" {
t.Errorf("LastError should still be recorded for transient errors")
}
}
func TestDynamicResourceCache_NamespaceFallbackIsPerGVR(t *testing.T) {
const ns = "dev-ns-1"
clusterGVR := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "clusterthings"}
namespacedGVR := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "namespacedthings"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
clusterGVR: "ClusterThingList",
namespacedGVR: "NamespacedThingList",
}, func(gvr schema.GroupVersionResource, namespace string) bool {
if gvr == namespacedGVR {
return namespace == ns
}
return namespace == ""
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceFallback: ns,
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
clusterScopes, _, err := d.probeScopes(clusterGVR, "")
if err != nil {
t.Fatalf("cluster GVR probe failed: %v", err)
}
if !reflect.DeepEqual(clusterScopes, []string{""}) {
t.Errorf("cluster GVR scopes = %q, want cluster-wide", clusterScopes)
}
nsScopes, _, err := d.probeScopes(namespacedGVR, "")
if err != nil {
t.Fatalf("namespaced GVR probe failed: %v", err)
}
if !reflect.DeepEqual(nsScopes, []string{ns}) {
t.Errorf("namespaced GVR scopes = %q, want [%q]", nsScopes, ns)
}
}
func TestDynamicResourceCache_ForcedNamespaceScopesEveryGVR(t *testing.T) {
const ns = "dev-ns-1"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == ns
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceScoped: true,
Namespace: ns,
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
scopes, _, err := d.probeScopes(gvr, "")
if err != nil {
t.Fatalf("forced namespace probe failed: %v", err)
}
if !reflect.DeepEqual(scopes, []string{ns}) {
t.Errorf("GVR scopes = %q, want [%q]", scopes, ns)
}
}
// The #768 core: when cluster-wide list is denied but the caller names a
// specific namespace they CAN list, the probe scopes to that namespace —
// not to the configured single fallback. This is what lets a namespace-
// restricted user read a CRD in the namespaces they actually have access to.
func TestDynamicResourceCache_ProbeScope_HonorsRequestedNamespace(t *testing.T) {
const fallbackNs, wantedNs = "fallback-ns", "argocd"
gvr := schema.GroupVersionResource{Group: "argoproj.io", Version: "v1alpha1", Resource: "applications"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "ApplicationList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == wantedNs // not cluster-wide, not the fallback
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceFallback: fallbackNs,
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
// Requesting the namespace the user can list scopes the informer there.
scopes, _, err := d.probeScopes(gvr, wantedNs)
if err != nil {
t.Fatalf("probeScopes(%q) failed: %v", wantedNs, err)
}
if !reflect.DeepEqual(scopes, []string{wantedNs}) {
t.Errorf("scopes = %q, want [%q]", scopes, wantedNs)
}
// With no requested namespace, it falls back to the configured fallback,
// which the user cannot list — so the probe is forbidden.
if _, _, err := d.probeScopes(gvr, ""); !apierrors.IsForbidden(err) {
t.Errorf("probeScopes(\"\") err = %v, want forbidden", err)
}
}
// List(gvr, "") must be served ONLY by a cluster-wide informer — it must not
// union incidental per-namespace informers, which would make results depend on
// what other requests warmed. ListNamespaces is the explicit union path.
func TestDynamicResourceCache_ListEmptyNamespaceDoesNotUnion(t *testing.T) {
const nsA, nsB = "team-a", "team-b"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == nsA || namespace == nsB // namespaced only, never cluster-wide
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
// Warm per-namespace informers for both namespaces.
if _, err := d.ListBlocking(gvr, nsA, 2*time.Second); err != nil {
t.Fatalf("ListBlocking(%q) failed: %v", nsA, err)
}
if _, err := d.ListBlocking(gvr, nsB, 2*time.Second); err != nil {
t.Fatalf("ListBlocking(%q) failed: %v", nsB, err)
}
// List(gvr, "") finds no cluster-wide informer and must NOT union the two
// per-namespace informers.
if got := d.readEntries(gvr, ""); got != nil {
t.Errorf("readEntries(gvr, \"\") returned %d entries, want none (no cluster-wide informer)", len(got))
}
}
// Mirror of ListEmptyNamespaceDoesNotUnion for namespace-scoped mode: when Radar
// connects with a namespace-restricted identity (NamespaceFallback set),
// cluster-wide list is denied so a GVR's only informers are per-namespace. An
// all-namespaces read — List(gvr, "") and Count(gvr, nil) — MUST union them, or
// namespaced CRDs show 0 in the UI despite synced per-namespace informers. The
// NamespaceFallback gate is what keeps the no-union contract intact for a
// cluster-wide cache (the sibling test) while fixing the namespace-scoped case.
func TestDynamicResourceCache_ListEmptyNamespaceUnionsWhenNamespaceScoped(t *testing.T) {
const nsA, nsB = "team-a", "team-b"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == nsA || namespace == nsB // namespaced only, never cluster-wide
})
for _, ns := range []string{nsA, nsB} {
obj := &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "example.com/v1",
"kind": "Widget",
"metadata": map[string]any{"name": "w-" + ns, "namespace": ns},
}}
if _, err := dyn.Resource(gvr).Namespace(ns).Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
t.Fatalf("seed %s: %v", ns, err)
}
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, NamespaceFallback: nsA})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
for _, ns := range []string{nsA, nsB} {
if _, err := d.ListBlocking(gvr, ns, 2*time.Second); err != nil {
t.Fatalf("ListBlocking(%q) failed: %v", ns, err)
}
}
got, err := d.List(gvr, "")
if err != nil {
t.Fatalf("List(gvr, \"\") failed: %v", err)
}
if len(got) != 2 {
t.Errorf("List(gvr, \"\") = %d objects, want 2 (union of per-namespace informers in namespace-scoped mode)", len(got))
}
// Count gates on the cache's own per-informer `synced` flag, which a
// background goroutine sets shortly AFTER informer.HasSynced() (what
// ListBlocking waited on) — so poll briefly rather than racing that
// propagation.
var n int
deadline := time.Now().Add(2 * time.Second)
for {
n, err = d.Count(gvr, nil)
if err == nil || time.Now().After(deadline) {
break
}
time.Sleep(10 * time.Millisecond)
}
if err != nil {
t.Fatalf("Count(gvr, nil) failed: %v", err)
}
if n != 2 {
t.Errorf("Count(gvr, nil) = %d, want 2", n)
}
}
// ListWatched is the internal "scan what's already cached" path: it unions
// every watched scope so namespace-restricted callers (audit, PolicyReport
// indexing) don't silently drop namespace-scoped contents the way List(gvr,
// "") would.
func TestDynamicResourceCache_ListWatchedUnionsNamespaceInformers(t *testing.T) {
const nsA, nsB = "team-a", "team-b"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == nsA || namespace == nsB // namespaced only, never cluster-wide
})
// Seed one object in each namespace so the informers have content.
for _, ns := range []string{nsA, nsB} {
obj := &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "example.com/v1",
"kind": "Widget",
"metadata": map[string]any{"name": "w-" + ns, "namespace": ns},
}}
if _, err := dyn.Resource(gvr).Namespace(ns).Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
t.Fatalf("seed %s: %v", ns, err)
}
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
for _, ns := range []string{nsA, nsB} {
if _, err := d.ListBlocking(gvr, ns, 2*time.Second); err != nil {
t.Fatalf("ListBlocking(%q) failed: %v", ns, err)
}
}
got, err := d.ListWatched(gvr)
if err != nil {
t.Fatalf("ListWatched failed: %v", err)
}
if len(got) != 2 {
t.Errorf("ListWatched returned %d objects, want 2 (union of both namespace-scoped informers)", len(got))
}
readOnly, err := d.ListWatchedReadOnly(gvr)
if err != nil {
t.Fatalf("ListWatchedReadOnly failed: %v", err)
}
if len(readOnly) != 2 {
t.Fatalf("ListWatchedReadOnly returned %d objects, want 2", len(readOnly))
}
copiesByName := make(map[string]*unstructured.Unstructured, len(got))
for _, item := range got {
copiesByName[item.GetName()] = item
}
for _, item := range readOnly {
if item == copiesByName[item.GetName()] {
t.Fatalf("ListWatched returned cached pointer for %s instead of a defensive copy", item.GetName())
}
}
readOnlyAgain, err := d.ListWatchedReadOnly(gvr)
if err != nil {
t.Fatalf("second ListWatchedReadOnly failed: %v", err)
}
readOnlyByName := make(map[string]*unstructured.Unstructured, len(readOnly))
for _, item := range readOnly {
readOnlyByName[item.GetName()] = item
}
for _, item := range readOnlyAgain {
if item != readOnlyByName[item.GetName()] {
t.Fatalf("ListWatchedReadOnly copied cached object %s", item.GetName())
}
}
item, err := d.GetWatched(gvr, nsB, "w-"+nsB)
if err != nil {
t.Fatalf("GetWatched failed: %v", err)
}
if item.GetNamespace() != nsB || item.GetName() != "w-"+nsB {
t.Fatalf("GetWatched returned %s/%s, want %s/%s", item.GetNamespace(), item.GetName(), nsB, "w-"+nsB)
}
item.SetName("mutated-copy")
again, err := d.GetWatched(gvr, nsB, "w-"+nsB)
if err != nil || again.GetName() != "w-"+nsB {
t.Fatalf("GetWatched mutated informer object: item=%v err=%v", again, err)
}
if _, err := d.GetWatched(gvr, nsB, "absent"); !errors.Is(err, ErrResourceNotFound) {
t.Fatalf("GetWatched missing object error = %v, want ErrResourceNotFound", err)
}
}
// ListNamespaces must short-circuit a cluster-scoped GVR to a cluster-wide
// read even when given a namespace set: cluster-scoped objects live under the
// "" namespace, so a per-namespace filter would index them to nothing. Pins
// the gvrIsNamespaced guard the topology paths rely on for Karpenter
// NodePool/NodeClaim and other cluster-scoped CRDs.
func TestDynamicResourceCache_ListNamespacesClusterScopedReadsClusterWide(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "karpenter.sh", Version: "v1", Resource: "nodepools"}
disc := &ResourceDiscovery{
resources: []APIResource{{
Group: gvr.Group, Version: gvr.Version, Kind: "NodePool", Name: gvr.Resource,
Namespaced: false, IsCRD: true, Verbs: []string{"get", "list", "watch"},
}},
resourceMap: map[string]APIResource{},
gvrMap: map[string]schema.GroupVersionResource{},
lastRefresh: time.Now(),
cacheTTL: time.Hour,
}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "NodePoolList",
}, func(schema.GroupVersionResource, string) bool { return true }) // cluster-wide allowed
// Seed a cluster-scoped object (no namespace).
obj := &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "karpenter.sh/v1",
"kind": "NodePool",
"metadata": map[string]any{"name": "default"},
}}
if _, err := dyn.Resource(gvr).Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
t.Fatalf("seed cluster-scoped object: %v", err)
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, Discovery: disc})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
if _, err := d.ListBlocking(gvr, "", 2*time.Second); err != nil {
t.Fatalf("ListBlocking failed: %v", err)
}
// A namespace filter must NOT zero out a cluster-scoped resource.
got, err := d.ListNamespaces(gvr, []string{"ns-a"})
if err != nil {
t.Fatalf("ListNamespaces failed: %v", err)
}
if len(got) != 1 {
t.Errorf("ListNamespaces(clusterScopedGVR, [ns-a]) = %d objects, want 1 (cluster-wide short-circuit)", len(got))
}
}
// Starting a cluster-wide informer must supersede (stop + drop) any
// namespace-scoped informers for the same GVR — the "never both" invariant —
// so reads (ListWatched) and change callbacks don't double up.
func TestDynamicResourceCache_ClusterWideSupersedesNamespaceInformers(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(schema.GroupVersionResource, string) bool { return true })
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
defer d.Stop()
has := func(ns string) bool {
d.mu.RLock()
defer d.mu.RUnlock()
_, ok := d.informers[informerKey{gvr: gvr, ns: ns}]
return ok
}
if err := d.startWatching(gvr, "team-a"); err != nil {
t.Fatalf("startWatching(team-a): %v", err)
}
if err := d.startWatching(gvr, "team-b"); err != nil {
t.Fatalf("startWatching(team-b): %v", err)
}
if !has("team-a") || !has("team-b") {
t.Fatal("expected both namespace-scoped informers before supersede")
}
if err := d.startWatching(gvr, ""); err != nil {
t.Fatalf("startWatching(cluster-wide): %v", err)
}
if !has("") {
t.Error("expected cluster-wide informer after supersede")
}
if has("team-a") || has("team-b") {
t.Error("namespace-scoped informers must be superseded by the cluster-wide one")
}
}
// A handler registered via AddGVRChangeHandler must reach informers created
// AFTER registration (lazy per-namespace watches, reap re-creations) — not
// only those present at call time — or derived caches miss those events.
func TestDynamicResourceCache_GVRChangeHandlerAppliesToLaterInformers(t *testing.T) {
const nsA, nsB = "team-a", "team-b"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, ns string) bool { return ns != "" }) // namespaced only
// Seed an object in nsB so its (later-created) informer has an add to deliver.
obj := &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "example.com/v1",
"kind": "Widget",
"metadata": map[string]any{"name": "w", "namespace": nsB},
}}
if _, err := dyn.Resource(gvr).Namespace(nsB).Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
t.Fatalf("seed nsB: %v", err)
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
defer d.Stop()
// Warm nsA so the handler has an existing informer to attach to at registration.
if _, err := d.ListBlocking(gvr, nsA, 2*time.Second); err != nil {
t.Fatalf("ListBlocking(nsA): %v", err)
}
var adds atomic.Int64
h := cache.ResourceEventHandlerFuncs{AddFunc: func(any) { adds.Add(1) }}
if err := d.AddGVRChangeHandler(gvr, h); err != nil {
t.Fatalf("AddGVRChangeHandler: %v", err)
}
// nsB's informer is created lazily after registration; it must still get h.
if _, err := d.ListBlocking(gvr, nsB, 2*time.Second); err != nil {
t.Fatalf("ListBlocking(nsB): %v", err)
}
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) && adds.Load() == 0 {
time.Sleep(10 * time.Millisecond)
}
if adds.Load() == 0 {
t.Error("handler did not fire for an informer created after registration")
}
}
// GVR-level status APIs must span namespace-scoped informers: with only
// per-namespace watches (no cluster-wide one), the kind is still watched and
// synced, so IsSynced/WaitForSync must report true and AddGVRChangeHandler
// must register — they resolve via entriesForGVR, not readEntries(gvr, "").
func TestDynamicResourceCache_StatusAPIsSpanNamespaceInformers(t *testing.T) {
const ns = "team-a"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == ns // namespaced only, never cluster-wide
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
if _, err := d.ListBlocking(gvr, ns, 2*time.Second); err != nil {
t.Fatalf("ListBlocking(%q) failed: %v", ns, err)
}
if !d.IsSynced(gvr) {
t.Error("IsSynced = false for a GVR watched only via a namespace-scoped informer, want true")
}
if !d.WaitForSync(gvr, 2*time.Second) {
t.Error("WaitForSync = false for a namespace-scoped-only GVR, want true")
}
if err := d.AddGVRChangeHandler(gvr, cache.ResourceEventHandlerFuncs{}); err != nil {
t.Errorf("AddGVRChangeHandler failed for a namespace-scoped-only GVR: %v", err)
}
// A subsequent List(gvr, "") must NOT short-circuit on the existing
// namespace-scoped informer and then find nothing. With no cluster-wide
// informer it re-probes cluster-wide (denied here) and returns a clean
// forbidden — never a spurious "informer not found".
_, err = d.List(gvr, "")
if err == nil {
t.Fatal("List(gvr, \"\") = nil error, want forbidden (cluster-wide denied)")
}
if !apierrors.IsForbidden(err) {
t.Errorf("List(gvr, \"\") err = %v, want a re-probed forbidden (not 'informer not found')", err)
}
}
// EnsureWatching must surface a probe that's forbidden everywhere it looks
// (cluster-wide AND the fallback namespace) as an apierrors.IsForbidden-
// classifiable error through the %w wrap — that's what lets the resources
// handler map a denied CRD list to 403 instead of 500. Mirrors #768: a user
// who can read the CRD only in a namespace the cache never probes.
func TestDynamicResourceCache_EnsureWatching_ForbiddenIsClassifiable(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "argoproj.io", Version: "v1alpha1", Resource: "applications"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "ApplicationList",
}, func(schema.GroupVersionResource, string) bool { return false })
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceFallback: "other-ns",
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
err = d.EnsureWatching(gvr)
if err == nil {
t.Fatal("EnsureWatching succeeded, want forbidden error")
}
if !apierrors.IsForbidden(err) {
t.Fatalf("EnsureWatching error not classifiable as forbidden (handler would 500 instead of 403): %v", err)
}
}
func fakeDynamicForListAccess(
t *testing.T,
listKinds map[schema.GroupVersionResource]string,
allow func(schema.GroupVersionResource, string) bool,
) *dynamicfake.FakeDynamicClient {
t.Helper()
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), listKinds)
dyn.PrependReactor("list", "*", func(action k8stesting.Action) (bool, runtime.Object, error) {
listAction, ok := action.(k8stesting.ListAction)
if !ok {
return false, nil, nil
}
gvr := listAction.GetResource()
if allow(gvr, listAction.GetNamespace()) {
return false, nil, nil
}
return true, nil, apierrors.NewForbidden(schema.GroupResource{Group: gvr.Group, Resource: gvr.Resource}, "", nil)
})
return dyn
}
// TestKindCoversNamespace pins the empty-vs-unreadable distinction: an empty list
// is only authoritative when the informer actually covers the namespace.
func TestKindCoversNamespace(t *testing.T) {
// nil ResourceScopes = legacy / cluster-wide default → covers everything.
if !(&ResourceCache{}).KindCoversNamespace("pods", "any") {
t.Error("nil scopes must cover all (legacy cluster-wide)")
}
clusterWide := &ResourceCache{config: CacheConfig{ResourceScopes: map[string]ResourceScope{"pods": {Enabled: true, Namespace: ""}}}}
if !clusterWide.KindCoversNamespace("pods", "prod") {
t.Error("cluster-wide pods must cover any namespace")
}
scoped := &ResourceCache{config: CacheConfig{ResourceScopes: map[string]ResourceScope{"pods": {Enabled: true, Namespace: "prod"}}}}
if !scoped.KindCoversNamespace("pods", "prod") {
t.Error("namespace-scoped pods must cover its own namespace")
}
if scoped.KindCoversNamespace("pods", "other") {
t.Error("pods scoped to prod must NOT cover another namespace (empty there is out-of-scope, not 0 pods)")
}
disabled := &ResourceCache{config: CacheConfig{ResourceScopes: map[string]ResourceScope{}}}
if disabled.KindCoversNamespace("pods", "prod") {
t.Error("a kind absent from an authoritative scope set is not covered")
}
}
// The PR core for CRDs: cluster-wide denied but several fallback namespaces
// granted → one scope per granted namespace, denied candidates skipped.
func TestDynamicResourceCache_ProbeScopesFansOutAcrossFallbacks(t *testing.T) {
const nsA, nsB, nsDenied = "team-a", "team-b", "team-c"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == nsA || namespace == nsB
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceFallbacks: []string{nsA, nsB, nsDenied},
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
scopes, complete, err := d.probeScopes(gvr, "")
if err != nil {
t.Fatalf("probeScopes failed: %v", err)
}
if !reflect.DeepEqual(scopes, []string{nsA, nsB}) {
t.Fatalf("scopes = %v, want [%s %s]", scopes, nsA, nsB)
}
if !complete {
t.Fatal("healthy fanout walk must report complete")
}
}
// End-to-end: an all-namespaces read of a fallback-scoped GVR starts one
// informer per granted namespace, unions their contents, and does not
// re-probe on subsequent reads.
func TestDynamicResourceCache_EnsureWatchingFansOutAndUnions(t *testing.T) {
const nsA, nsB, nsDenied = "team-a", "team-b", "team-c"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == nsA || namespace == nsB
})
for _, ns := range []string{nsA, nsB} {
obj := &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "example.com/v1",
"kind": "Widget",
"metadata": map[string]any{"name": "w-" + ns, "namespace": ns},
}}
if _, err := dyn.Resource(gvr).Namespace(ns).Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
t.Fatalf("seed %s: %v", ns, err)
}
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceFallbacks: []string{nsA, nsB, nsDenied},
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
items, err := d.ListBlocking(gvr, "", 3*time.Second)
if err != nil {
t.Fatalf("ListBlocking failed: %v", err)
}
if len(items) != 2 {
t.Fatalf("union read = %d items, want 2 (one per granted namespace)", len(items))
}
if !d.hasCoveringInformer(gvr, "") {
t.Fatal("all-namespaces scope not marked resolved — every read would re-probe cluster + candidates")
}
if n, err := d.Count(gvr, nil); err != nil || n != 2 {
t.Fatalf("Count = %d, %v; want 2", n, err)
}
}
// Pins the unavailable-over-partial contract: a fanout GVR whose informers
// exist only from namespace-specific reads (all-namespaces walk never
// settled) must refuse an all-namespace count rather than sum the subset —
// and must start counting once an all-namespaces read settles the walk.
func TestDynamicResourceCache_CountRefusesUnsettledFanout(t *testing.T) {
const nsA, nsB = "team-a", "team-b"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == nsA || namespace == nsB
})
for _, ns := range []string{nsA, nsB} {
obj := &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "example.com/v1",
"kind": "Widget",
"metadata": map[string]any{"name": "w-" + ns, "namespace": ns},
}}
if _, err := dyn.Resource(gvr).Namespace(ns).Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
t.Fatalf("seed %s: %v", ns, err)
}
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceFallbacks: []string{nsA, nsB},
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
// Namespace-specific read only — informer for nsA exists, walk unsettled.
if _, err := d.ListBlocking(gvr, nsA, 3*time.Second); err != nil {
t.Fatalf("ListBlocking(%s): %v", nsA, err)
}
if n, err := d.Count(gvr, nil); err == nil {
t.Fatalf("all-namespace count on unsettled fanout returned %d, want error", n)
}
// All-namespaces read settles the walk; the count becomes authoritative.
if _, err := d.ListBlocking(gvr, "", 3*time.Second); err != nil {
t.Fatalf("ListBlocking(all): %v", err)
}
if n, err := d.Count(gvr, nil); err != nil || n != 2 {
t.Fatalf("settled count = %d, %v; want 2", n, err)
}
}
// Legacy single-fallback regression: with exactly one configured fallback
// namespace, an informer created by an explicit-namespace read fully covers
// the configured scope — Count(gvr, nil) must work without ever seeing an
// all-namespaces read.
func TestDynamicResourceCache_SingleFallbackCountsWithoutAllNamespacesRead(t *testing.T) {
const ns = "team-a"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == ns
})
obj := &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "example.com/v1",
"kind": "Widget",
"metadata": map[string]any{"name": "w-1", "namespace": ns},
}}
if _, err := dyn.Resource(gvr).Namespace(ns).Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
t.Fatalf("seed: %v", err)
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
NamespaceFallback: ns,
})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
// Explicit-namespace read only — never an all-namespaces read.
if _, err := d.ListBlocking(gvr, ns, 3*time.Second); err != nil {
t.Fatalf("ListBlocking(%s): %v", ns, err)
}
if n, err := d.Count(gvr, nil); err != nil || n != 1 {
t.Fatalf("single-fallback Count = %d, %v; want 1", n, err)
}
}
// A namespace-scoped watch covers its own namespace only: another namespace
// is not covered, so a caller can tell "not watched" from "still syncing".
func TestDynamicResourceCache_HasCoveringInformerIsPerNamespace(t *testing.T) {
const nsA, nsB = "team-a", "team-b"
gvr := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
dyn := fakeDynamicForListAccess(t, map[schema.GroupVersionResource]string{
gvr: "WidgetList",
}, func(_ schema.GroupVersionResource, namespace string) bool {
return namespace == nsA
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, NamespaceFallbacks: []string{nsA}})
if err != nil {
t.Fatalf("NewDynamicResourceCache failed: %v", err)
}
if _, err := d.ListBlocking(gvr, nsA, 3*time.Second); err != nil {
t.Fatalf("ListBlocking failed: %v", err)
}
if !d.HasCoveringInformer(gvr, nsA) || !d.IsNamespaceSynced(gvr, nsA) {
t.Fatal("the watched namespace is not covered and synced")
}
if d.HasCoveringInformer(gvr, nsB) || d.IsNamespaceSynced(gvr, nsB) {
t.Fatal("a namespace with no informer reads as covered")
}
var missing *DynamicResourceCache
if missing.HasCoveringInformer(gvr, nsA) {
t.Fatal("a nil cache covers nothing")
}
}