“Naeel”
5d52c9ba94
TestStartManagedNamespaceWatcherIntegration проверяет полный маршрут горячей регистрации namespace без real cluster: 1. RunManagedNamespaceWatcher запускается с k8sfake.NewSimpleClientset() 2. В fake client создаётся Namespace с label fission.io/managed=true 3. Kubernetes informer детектирует событие (без polling, через Watch) 4. SubscriberFuncs.OnNamespaceAdd вызывается 5. NamespaceManager содержит запись со статусом Active Тест доказывает, что вся цепочка fake k8s event → informer → AddFunc → subscriber → manager работает корректно без rolling restart процесса. Также добавлен import metav1 в test file (требовался для CreateOptions).
923 lines
34 KiB
Go
923 lines
34 KiB
Go
package utils
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"reflect"
|
||
"testing"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
"go.uber.org/zap/zaptest/observer"
|
||
corev1 "k8s.io/api/core/v1"
|
||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||
k8sfake "k8s.io/client-go/kubernetes/fake"
|
||
k8sCache "k8s.io/client-go/tools/cache"
|
||
|
||
managerPkg "github.com/fission/fission/pkg/utils/manager"
|
||
)
|
||
|
||
type testNamespaceSubscriber struct {
|
||
name string
|
||
addErr error
|
||
removeErr error
|
||
resyncErr error
|
||
addCalls int
|
||
removeCalls int
|
||
resyncCalls int
|
||
}
|
||
|
||
func (s *testNamespaceSubscriber) Name() string {
|
||
return s.name
|
||
}
|
||
|
||
func (s *testNamespaceSubscriber) OnNamespaceAdd(ctx context.Context, record NamespaceRecord) error {
|
||
s.addCalls++
|
||
return s.addErr
|
||
}
|
||
|
||
func (s *testNamespaceSubscriber) OnNamespaceRemove(ctx context.Context, record NamespaceRecord) error {
|
||
s.removeCalls++
|
||
return s.removeErr
|
||
}
|
||
|
||
func (s *testNamespaceSubscriber) OnNamespaceResync(ctx context.Context, record NamespaceRecord) error {
|
||
s.resyncCalls++
|
||
return s.resyncErr
|
||
}
|
||
|
||
func TestNamespaceManagerSnapshotAndGet(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
now := time.Now().UTC()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-b", Source: NamespaceSourceWatcher, ObservedAt: now})
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher, ObservedAt: now})
|
||
|
||
expected := []string{"tenant-a", "tenant-b"}
|
||
if !reflect.DeepEqual(expected, manager.Snapshot()) {
|
||
t.Fatalf("expected snapshot %v, got %v", expected, manager.Snapshot())
|
||
}
|
||
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok {
|
||
t.Fatalf("expected tenant-a to exist")
|
||
}
|
||
if record.Name != "tenant-a" || record.Phase != NamespacePhaseDiscovered {
|
||
t.Fatalf("unexpected record returned: %+v", record)
|
||
}
|
||
summary := manager.Summary()
|
||
if summary.TotalNamespaces != 2 {
|
||
t.Fatalf("expected summary total namespaces to be 2, got %d", summary.TotalNamespaces)
|
||
}
|
||
if summary.LiveNamespaces != 2 {
|
||
t.Fatalf("expected live namespaces to be 2, got %d", summary.LiveNamespaces)
|
||
}
|
||
if summary.SourceCounts[NamespaceSourceWatcher] != 2 {
|
||
t.Fatalf("expected watcher source count to be 2")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerUpsertIncrementsGeneration(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
now := time.Now().UTC()
|
||
|
||
first := manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher, ObservedAt: now})
|
||
second := manager.Upsert(NamespaceEvent{Type: NamespaceEventUpdate, Name: "tenant-a", Source: NamespaceSourceWatcher, ObservedAt: now.Add(time.Second)})
|
||
|
||
if first.Generation != 1 {
|
||
t.Fatalf("expected first generation to be 1, got %d", first.Generation)
|
||
}
|
||
if second.Generation != 2 {
|
||
t.Fatalf("expected second generation to be 2, got %d", second.Generation)
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerMarkPartState(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
|
||
record, ok := manager.MarkPartState("tenant-a", "router", NamespacePartState{State: NamespacePartStateActive})
|
||
if !ok {
|
||
t.Fatalf("expected mark part state to succeed")
|
||
}
|
||
if record.RegisteredParts["router"].State != NamespacePartStateActive {
|
||
t.Fatalf("expected router part state to be stored")
|
||
}
|
||
if record.Phase != NamespacePhaseActive {
|
||
t.Fatalf("expected phase to become active, got %s", record.Phase)
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerRemove(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
|
||
if !manager.Remove("tenant-a") {
|
||
t.Fatalf("expected remove to succeed")
|
||
}
|
||
if _, ok := manager.Get("tenant-a"); ok {
|
||
t.Fatalf("expected record to be removed")
|
||
}
|
||
if manager.Remove("tenant-a") {
|
||
t.Fatalf("expected second remove to report false")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerSnapshotRecordsReturnsCopies(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{
|
||
Type: NamespaceEventAdd,
|
||
Name: "tenant-a",
|
||
Source: NamespaceSourceWatcher,
|
||
Labels: map[string]string{"fission.io/managed": "true"},
|
||
})
|
||
_, _ = manager.MarkPartState("tenant-a", "router", NamespacePartState{State: NamespacePartStateActive})
|
||
|
||
records := manager.SnapshotRecords()
|
||
records[0].Labels["fission.io/managed"] = "false"
|
||
records[0].RegisteredParts["router"] = NamespacePartState{State: "failed"}
|
||
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok {
|
||
t.Fatalf("expected tenant-a to exist")
|
||
}
|
||
if record.Labels["fission.io/managed"] != "true" {
|
||
t.Fatalf("expected snapshot records to be detached copies")
|
||
}
|
||
if record.RegisteredParts["router"].State != NamespacePartStateActive {
|
||
t.Fatalf("expected part states to be detached copies")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerSubscribers(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Subscribe(&testNamespaceSubscriber{name: "router"})
|
||
manager.Subscribe(&testNamespaceSubscriber{name: "buildermgr"})
|
||
manager.Subscribe(&testNamespaceSubscriber{name: "router"})
|
||
|
||
expected := []string{"buildermgr", "router"}
|
||
if !reflect.DeepEqual(expected, manager.SnapshotSubscribers()) {
|
||
t.Fatalf("expected subscribers %v, got %v", expected, manager.SnapshotSubscribers())
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerMarkPartStateDerivesPhase(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
|
||
record, ok := manager.MarkPartState("tenant-a", "router", NamespacePartState{State: NamespacePartStateRegistering})
|
||
if !ok {
|
||
t.Fatalf("expected registering state update to succeed")
|
||
}
|
||
if record.Phase != NamespacePhaseRegistering {
|
||
t.Fatalf("expected phase registering, got %s", record.Phase)
|
||
}
|
||
|
||
record, ok = manager.MarkPartState("tenant-a", "router", NamespacePartState{State: NamespacePartStateFailed})
|
||
if !ok {
|
||
t.Fatalf("expected failed state update to succeed")
|
||
}
|
||
if record.Phase != NamespacePhaseFailed {
|
||
t.Fatalf("expected phase failed, got %s", record.Phase)
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerBootstrap(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
now := time.Now().UTC()
|
||
records := manager.Bootstrap([]string{"tenant-b", "tenant-a"}, NamespaceSourceBackfill, now)
|
||
|
||
if len(records) != 2 {
|
||
t.Fatalf("expected 2 bootstrap records, got %d", len(records))
|
||
}
|
||
if !reflect.DeepEqual([]string{"tenant-a", "tenant-b"}, manager.Snapshot()) {
|
||
t.Fatalf("expected manager snapshot to contain bootstrapped namespaces")
|
||
}
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok {
|
||
t.Fatalf("expected tenant-a after bootstrap")
|
||
}
|
||
if record.Source != NamespaceSourceBackfill {
|
||
t.Fatalf("expected bootstrap source backfill, got %s", record.Source)
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerBootstrapAndDispatch(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
builder := &testNamespaceSubscriber{name: "buildermgr"}
|
||
manager.Subscribe(router)
|
||
manager.Subscribe(builder)
|
||
|
||
records, err := manager.BootstrapAndDispatch(context.Background(), []string{"tenant-b", "tenant-a"}, NamespaceSourceBackfill, time.Now().UTC())
|
||
if err != nil {
|
||
t.Fatalf("expected bootstrap and dispatch success: %v", err)
|
||
}
|
||
if len(records) != 2 {
|
||
t.Fatalf("expected 2 records, got %d", len(records))
|
||
}
|
||
if router.addCalls != 2 || builder.addCalls != 2 {
|
||
t.Fatalf("expected both subscribers to be called for each namespace")
|
||
}
|
||
if records[0].Phase != NamespacePhaseActive || records[1].Phase != NamespacePhaseActive {
|
||
t.Fatalf("expected active records after bootstrap dispatch")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerBootstrapAndDispatchFailure(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router", addErr: errors.New("router failed")}
|
||
builder := &testNamespaceSubscriber{name: "buildermgr"}
|
||
manager.Subscribe(router)
|
||
manager.Subscribe(builder)
|
||
|
||
records, err := manager.BootstrapAndDispatch(context.Background(), []string{"tenant-a"}, NamespaceSourceBackfill, time.Now().UTC())
|
||
if err == nil {
|
||
t.Fatalf("expected bootstrap dispatch failure")
|
||
}
|
||
if len(records) != 1 {
|
||
t.Fatalf("expected single record result")
|
||
}
|
||
if records[0].Phase != NamespacePhaseFailed {
|
||
t.Fatalf("expected failed phase after subscriber error, got %s", records[0].Phase)
|
||
}
|
||
if records[0].RegisteredParts["router"].LastError != "router failed" {
|
||
t.Fatalf("expected router failure to be stored")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerPartStateHelpers(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
|
||
record, ok := manager.MarkPartRegistering("tenant-a", "router")
|
||
if !ok || record.Phase != NamespacePhaseRegistering {
|
||
t.Fatalf("expected registering helper to set registering phase")
|
||
}
|
||
|
||
record, ok = manager.MarkPartActive("tenant-a", "router")
|
||
if !ok || record.Phase != NamespacePhaseActive {
|
||
t.Fatalf("expected active helper to set active phase")
|
||
}
|
||
|
||
record, ok = manager.MarkPartFailed("tenant-a", "router", errors.New("boom"))
|
||
if !ok || record.Phase != NamespacePhaseFailed {
|
||
t.Fatalf("expected failed helper to set failed phase")
|
||
}
|
||
if record.RegisteredParts["router"].LastError != "boom" {
|
||
t.Fatalf("expected failed helper to persist error")
|
||
}
|
||
}
|
||
|
||
func TestNewBootstrappedNamespaceManager(t *testing.T) {
|
||
resolver := &NamespaceResolver{
|
||
FissionResourceNS: map[string]string{
|
||
"tenant-b": "tenant-b",
|
||
"tenant-a": "tenant-a",
|
||
},
|
||
}
|
||
|
||
manager := NewBootstrappedNamespaceManager(resolver, NamespaceSourceEnv, time.Now().UTC())
|
||
if !reflect.DeepEqual([]string{"tenant-a", "tenant-b"}, manager.Snapshot()) {
|
||
t.Fatalf("expected bootstrapped manager snapshot from resolver")
|
||
}
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok {
|
||
t.Fatalf("expected tenant-a in bootstrapped manager")
|
||
}
|
||
if record.Source != NamespaceSourceEnv {
|
||
t.Fatalf("expected env source, got %s", record.Source)
|
||
}
|
||
}
|
||
|
||
func TestNewWatcherNamespaceManager(t *testing.T) {
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
builder := &testNamespaceSubscriber{name: "buildermgr"}
|
||
manager, err := NewWatcherNamespaceManager(context.Background(), []string{"tenant-b", "tenant-a"}, NamespaceSourceEnv, time.Now().UTC(), router, builder)
|
||
if err != nil {
|
||
t.Fatalf("expected watcher manager bootstrap success: %v", err)
|
||
}
|
||
if !reflect.DeepEqual([]string{"tenant-a", "tenant-b"}, manager.Snapshot()) {
|
||
t.Fatalf("expected watcher manager snapshot to match bootstrapped namespaces")
|
||
}
|
||
if router.addCalls != 2 || builder.addCalls != 2 {
|
||
t.Fatalf("expected all subscribers to receive bootstrap dispatch calls")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceBecameUnmanaged(t *testing.T) {
|
||
oldNamespace := &corev1.Namespace{}
|
||
oldNamespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
newNamespace := &corev1.Namespace{}
|
||
newNamespace.Labels = map[string]string{}
|
||
|
||
if !NamespaceBecameUnmanaged(oldNamespace, newNamespace) {
|
||
t.Fatalf("expected namespace to become unmanaged")
|
||
}
|
||
if NamespaceBecameUnmanaged(nil, newNamespace) {
|
||
t.Fatalf("expected nil old namespace to report false")
|
||
}
|
||
}
|
||
|
||
func TestDispatchNamespaceAddAndResync(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
manager.Subscribe(router)
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
|
||
record, ok, err := DispatchNamespaceAdd(context.Background(), manager, namespace, NamespaceSourceWatcher, time.Now().UTC())
|
||
if !ok || err != nil || record.Phase != NamespacePhaseActive {
|
||
t.Fatalf("expected add dispatch success, ok=%v err=%v phase=%s", ok, err, record.Phase)
|
||
}
|
||
record, ok, err = DispatchNamespaceResync(context.Background(), manager, namespace, NamespaceSourceWatcher, time.Now().UTC())
|
||
if !ok || err != nil || record.Phase != NamespacePhaseActive {
|
||
t.Fatalf("expected resync dispatch success, ok=%v err=%v phase=%s", ok, err, record.Phase)
|
||
}
|
||
if router.addCalls != 1 || router.resyncCalls != 1 {
|
||
t.Fatalf("expected add and resync subscriber calls")
|
||
}
|
||
}
|
||
|
||
func TestRecordNamespaceRemoval(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
record, ok := RecordNamespaceRemoval(manager, k8sCache.DeletedFinalStateUnknown{Obj: namespace}, NamespaceSourceWatcher, time.Now().UTC())
|
||
if !ok {
|
||
t.Fatalf("expected removal bookkeeping to succeed")
|
||
}
|
||
if record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected removed phase after bookkeeping, got %s", record.Phase)
|
||
}
|
||
}
|
||
|
||
func TestHandleWatcherNamespaceAddUpdateDelete(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
manager.Subscribe(router)
|
||
logger := zap.NewNop()
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
|
||
HandleWatcherNamespaceAdd(context.Background(), logger, "router.NSWatcher", manager, namespace)
|
||
if router.addCalls != 1 {
|
||
t.Fatalf("expected add handler to dispatch add")
|
||
}
|
||
|
||
HandleWatcherNamespaceUpdate(context.Background(), logger, "router.NSWatcher", manager, namespace, namespace, NamespaceRemovalStrategyTrackOnly)
|
||
if router.resyncCalls != 1 {
|
||
t.Fatalf("expected update handler to dispatch resync")
|
||
}
|
||
|
||
HandleWatcherNamespaceDelete(context.Background(), logger, "router.NSWatcher", manager, k8sCache.DeletedFinalStateUnknown{Obj: namespace}, NamespaceRemovalStrategyTrackOnly)
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok || record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected delete handler to mark namespace removed")
|
||
}
|
||
}
|
||
|
||
func TestHandleWatcherNamespaceAddLogsSummary(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
manager.Subscribe(router)
|
||
core, logs := observer.New(zap.InfoLevel)
|
||
logger := zap.New(core)
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
|
||
HandleWatcherNamespaceAdd(context.Background(), logger, "router.NSWatcher", manager, namespace)
|
||
|
||
entries := logs.AllUntimed()
|
||
if len(entries) != 1 {
|
||
t.Fatalf("expected one summary log entry, got %d", len(entries))
|
||
}
|
||
if entries[0].Message != "router.NSWatcher: namespace manager summary" {
|
||
t.Fatalf("unexpected log message: %s", entries[0].Message)
|
||
}
|
||
fields := entries[0].ContextMap()
|
||
active, ok := fields["has_active_namespaces"].(bool)
|
||
if !ok || !active {
|
||
t.Fatalf("expected summary log to report active namespaces")
|
||
}
|
||
if live, ok := fields["live_namespaces"].(int64); !ok || live != 1 {
|
||
t.Fatalf("expected summary log to report one live namespace, got %v", fields["live_namespaces"])
|
||
}
|
||
}
|
||
|
||
func TestHandleWatcherNamespaceAddWithNilLogger(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router", addErr: errors.New("add failed")}
|
||
manager.Subscribe(router)
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
|
||
HandleWatcherNamespaceAdd(context.Background(), nil, "router.NSWatcher", manager, namespace)
|
||
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok {
|
||
t.Fatalf("expected tenant-a to exist after add handling")
|
||
}
|
||
if record.Phase != NamespacePhaseFailed {
|
||
t.Fatalf("expected failed phase after subscriber error, got %s", record.Phase)
|
||
}
|
||
if router.addCalls != 1 {
|
||
t.Fatalf("expected failing subscriber to be called once")
|
||
}
|
||
}
|
||
|
||
func TestHandleWatcherNamespaceRemovalDispatch(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
manager.Subscribe(router)
|
||
logger := zap.NewNop()
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
|
||
HandleWatcherNamespaceRemoval(context.Background(), logger, "router.NSWatcher", manager, namespace, NamespaceRemovalStrategyDispatchRemove)
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok || record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected dispatch removal to keep removed record")
|
||
}
|
||
if router.removeCalls != 1 {
|
||
t.Fatalf("expected remove subscriber to be called once")
|
||
}
|
||
}
|
||
|
||
func TestHandleWatcherNamespaceUpdateDispatchRemoval(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
manager.Subscribe(router)
|
||
logger := zap.NewNop()
|
||
oldNamespace := &corev1.Namespace{}
|
||
oldNamespace.Name = "tenant-a"
|
||
oldNamespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
newNamespace := &corev1.Namespace{}
|
||
newNamespace.Name = "tenant-a"
|
||
newNamespace.Labels = map[string]string{}
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
|
||
HandleWatcherNamespaceUpdate(context.Background(), logger, "router.NSWatcher", manager, oldNamespace, newNamespace, NamespaceRemovalStrategyDispatchRemove)
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok || record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected dispatch removal on managed->unmanaged transition")
|
||
}
|
||
if router.removeCalls != 1 {
|
||
t.Fatalf("expected remove subscriber to be called once on label drop")
|
||
}
|
||
}
|
||
|
||
func TestNewNamespaceWatcherEventHandlers(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
manager.Subscribe(router)
|
||
logger := zap.NewNop()
|
||
handlers := NewNamespaceWatcherEventHandlers(context.Background(), logger, "router.NSWatcher", manager, NamespaceRemovalStrategyTrackOnly)
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
|
||
handlers.AddFunc(namespace)
|
||
handlers.UpdateFunc(namespace, namespace)
|
||
handlers.DeleteFunc(k8sCache.DeletedFinalStateUnknown{Obj: namespace})
|
||
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok || record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected watcher event handlers to drive namespace lifecycle")
|
||
}
|
||
if router.addCalls != 1 || router.resyncCalls != 1 {
|
||
t.Fatalf("expected add and resync calls through event handlers")
|
||
}
|
||
}
|
||
|
||
func TestPrepareManagedNamespaceWatcher(t *testing.T) {
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
logger := zap.NewNop()
|
||
manager, handlers, err := PrepareManagedNamespaceWatcher(context.Background(), logger, ManagedNamespaceWatcherConfig{
|
||
Component: "router.NSWatcher",
|
||
Namespaces: []string{"tenant-a"},
|
||
RemovalStrategy: NamespaceRemovalStrategyTrackOnly,
|
||
Subscriber: router,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("expected watcher preparation success: %v", err)
|
||
}
|
||
if !reflect.DeepEqual([]string{"tenant-a"}, manager.Snapshot()) {
|
||
t.Fatalf("expected watcher manager snapshot to contain bootstrapped namespace")
|
||
}
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
handlers.UpdateFunc(namespace, namespace)
|
||
if router.addCalls != 1 || router.resyncCalls != 1 {
|
||
t.Fatalf("expected bootstrap add and handler resync calls")
|
||
}
|
||
}
|
||
|
||
func TestRunManagedNamespaceWatcher(t *testing.T) {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
defer cancel()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
logger := zap.NewNop()
|
||
manager, err := RunManagedNamespaceWatcher(ctx, logger, k8sfake.NewSimpleClientset(), managerPkg.New(), ManagedNamespaceWatcherConfig{
|
||
Component: "router.NSWatcher",
|
||
Namespaces: []string{"tenant-a"},
|
||
RemovalStrategy: NamespaceRemovalStrategyTrackOnly,
|
||
Subscriber: router,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("expected managed namespace watcher run success: %v", err)
|
||
}
|
||
if !reflect.DeepEqual([]string{"tenant-a"}, manager.Snapshot()) {
|
||
t.Fatalf("expected watcher manager snapshot to contain bootstrapped namespace")
|
||
}
|
||
}
|
||
|
||
func TestRunManagedNamespaceWatcherLogsStartedSummary(t *testing.T) {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
defer cancel()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
core, logs := observer.New(zap.InfoLevel)
|
||
logger := zap.New(core)
|
||
|
||
_, err := RunManagedNamespaceWatcher(ctx, logger, k8sfake.NewSimpleClientset(), managerPkg.New(), ManagedNamespaceWatcherConfig{
|
||
Component: "router.NSWatcher",
|
||
Namespaces: []string{"tenant-a"},
|
||
RemovalStrategy: NamespaceRemovalStrategyTrackOnly,
|
||
Subscriber: router,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("expected managed namespace watcher run success: %v", err)
|
||
}
|
||
|
||
entries := logs.FilterMessage("router.NSWatcher: started namespace watcher").AllUntimed()
|
||
if len(entries) != 1 {
|
||
t.Fatalf("expected one started-summary log entry, got %d", len(entries))
|
||
}
|
||
fields := entries[0].ContextMap()
|
||
if live, ok := fields["live_namespaces"].(int64); !ok || live != 1 {
|
||
t.Fatalf("expected started summary to report one live namespace, got %v", fields["live_namespaces"])
|
||
}
|
||
if active, ok := fields["has_active_namespaces"].(bool); !ok || !active {
|
||
t.Fatalf("expected started summary to report active namespaces")
|
||
}
|
||
if total, ok := fields["total_namespaces"].(int64); !ok || total != 1 {
|
||
t.Fatalf("expected started summary to report one total namespace, got %v", fields["total_namespaces"])
|
||
}
|
||
if got := fields["subscribers"]; got == nil || !reflect.DeepEqual(got, []interface{}{"router"}) && !reflect.DeepEqual(got, []string{"router"}) {
|
||
t.Fatalf("expected started summary to report router subscriber, got %v", got)
|
||
}
|
||
if got := fields["phase_counts"]; got == nil {
|
||
t.Fatalf("expected started summary to report phase counts")
|
||
}
|
||
}
|
||
|
||
func TestRunManagedNamespaceWatcherWithNilLogger(t *testing.T) {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
defer cancel()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
|
||
manager, err := RunManagedNamespaceWatcher(ctx, nil, k8sfake.NewSimpleClientset(), managerPkg.New(), ManagedNamespaceWatcherConfig{
|
||
Component: "router.NSWatcher",
|
||
Namespaces: []string{"tenant-a"},
|
||
RemovalStrategy: NamespaceRemovalStrategyTrackOnly,
|
||
Subscriber: router,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("expected managed namespace watcher run success: %v", err)
|
||
}
|
||
if !reflect.DeepEqual([]string{"tenant-a"}, manager.Snapshot()) {
|
||
t.Fatalf("expected watcher manager snapshot to contain bootstrapped namespace")
|
||
}
|
||
}
|
||
|
||
func TestNewDefaultManagedNamespaceWatcherConfig(t *testing.T) {
|
||
resolver := DefaultNSResolver()
|
||
original := resolver.FissionResourceNS
|
||
resolver.FissionResourceNS = map[string]string{"tenant-a": "tenant-a"}
|
||
defer func() {
|
||
resolver.FissionResourceNS = original
|
||
}()
|
||
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
config := NewDefaultManagedNamespaceWatcherConfig("router.NSWatcher", router)
|
||
if config.Component != "router.NSWatcher" {
|
||
t.Fatalf("expected component name to be preserved")
|
||
}
|
||
if config.RemovalStrategy != NamespaceRemovalStrategyTrackOnly {
|
||
t.Fatalf("expected default removal strategy track-only")
|
||
}
|
||
if !reflect.DeepEqual([]string{"tenant-a"}, config.Namespaces) {
|
||
t.Fatalf("expected namespaces from resolver snapshot")
|
||
}
|
||
if config.Subscriber != router {
|
||
t.Fatalf("expected subscriber to be preserved")
|
||
}
|
||
}
|
||
|
||
func TestPrepareManagedNamespaceWatcherDefaultsToTrackOnly(t *testing.T) {
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
logger := zap.NewNop()
|
||
manager, handlers, err := PrepareManagedNamespaceWatcher(context.Background(), logger, ManagedNamespaceWatcherConfig{
|
||
Component: "router.NSWatcher",
|
||
Namespaces: []string{"tenant-a"},
|
||
Subscriber: router,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("expected watcher preparation success: %v", err)
|
||
}
|
||
namespace := &corev1.Namespace{}
|
||
namespace.Name = "tenant-a"
|
||
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
handlers.DeleteFunc(k8sCache.DeletedFinalStateUnknown{Obj: namespace})
|
||
record, ok := manager.Get("tenant-a")
|
||
if !ok || record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected default strategy to keep track-only removal semantics")
|
||
}
|
||
if router.removeCalls != 0 {
|
||
t.Fatalf("expected default strategy not to dispatch remove")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerSummary(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-b", Source: NamespaceSourceWatcher})
|
||
_, _ = manager.MarkPartRegistering("tenant-a", "router")
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventRemove, Name: "tenant-b", Source: NamespaceSourceWatcher})
|
||
manager.Subscribe(&testNamespaceSubscriber{name: "router"})
|
||
|
||
summary := manager.Summary()
|
||
if summary.TotalNamespaces != 2 {
|
||
t.Fatalf("expected total namespaces 2, got %d", summary.TotalNamespaces)
|
||
}
|
||
if summary.LiveNamespaces != 1 {
|
||
t.Fatalf("expected one live namespace, got %d", summary.LiveNamespaces)
|
||
}
|
||
if summary.PhaseCounts[NamespacePhaseRegistering] != 1 {
|
||
t.Fatalf("expected one registering namespace")
|
||
}
|
||
if summary.PhaseCounts[NamespacePhaseRemoved] != 1 {
|
||
t.Fatalf("expected one removed namespace")
|
||
}
|
||
if !reflect.DeepEqual([]string{"router"}, summary.Subscribers) {
|
||
t.Fatalf("expected router subscriber in summary")
|
||
}
|
||
if summary.SourceCounts[NamespaceSourceWatcher] != 2 {
|
||
t.Fatalf("expected two watcher-sourced namespaces")
|
||
}
|
||
}
|
||
|
||
func TestLogNamespaceManagerSummary(t *testing.T) {
|
||
LogNamespaceManagerSummary(nil, "ignored", NamespaceManagerSummary{})
|
||
|
||
logger := zap.NewNop()
|
||
summary := NamespaceManagerSummary{
|
||
TotalNamespaces: 2,
|
||
LiveNamespaces: 1,
|
||
PhaseCounts: map[NamespacePhase]int{
|
||
NamespacePhaseActive: 1,
|
||
NamespacePhaseRemoved: 1,
|
||
},
|
||
SourceCounts: map[NamespaceSource]int{
|
||
NamespaceSourceWatcher: 1,
|
||
NamespaceSourceEnv: 1,
|
||
},
|
||
Subscribers: []string{"router", "buildermgr"},
|
||
}
|
||
|
||
LogNamespaceManagerSummary(logger, "namespace summary", summary)
|
||
}
|
||
|
||
func TestLogNamespaceManagerSummaryIncludesHasActiveNamespaces(t *testing.T) {
|
||
core, logs := observer.New(zap.InfoLevel)
|
||
logger := zap.New(core)
|
||
|
||
LogNamespaceManagerSummary(logger, "namespace summary", NamespaceManagerSummary{
|
||
LiveNamespaces: 1,
|
||
PhaseCounts: map[NamespacePhase]int{},
|
||
SourceCounts: map[NamespaceSource]int{},
|
||
Subscribers: []string{},
|
||
})
|
||
|
||
entries := logs.AllUntimed()
|
||
if len(entries) != 1 {
|
||
t.Fatalf("expected one log entry, got %d", len(entries))
|
||
}
|
||
fields := entries[0].ContextMap()
|
||
active, ok := fields["has_active_namespaces"].(bool)
|
||
if !ok {
|
||
t.Fatalf("expected has_active_namespaces field in log context")
|
||
}
|
||
if !active {
|
||
t.Fatalf("expected has_active_namespaces=true in log context")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerSummaryEmptyContract(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
summary := manager.Summary()
|
||
if summary.PhaseCounts == nil {
|
||
t.Fatalf("expected phase counts map to be initialized")
|
||
}
|
||
if summary.SourceCounts == nil {
|
||
t.Fatalf("expected source counts map to be initialized")
|
||
}
|
||
if summary.Subscribers == nil {
|
||
t.Fatalf("expected subscribers slice to be initialized")
|
||
}
|
||
if len(summary.Subscribers) != 0 {
|
||
t.Fatalf("expected no subscribers in empty summary")
|
||
}
|
||
}
|
||
func TestLogNamespaceManagerSummaryWithNilLogger(t *testing.T) {
|
||
LogNamespaceManagerSummary(nil, "ignored", NamespaceManagerSummary{TotalNamespaces: 1})
|
||
}
|
||
|
||
func TestNamespaceManagerDispatchAdd(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
builder := &testNamespaceSubscriber{name: "buildermgr"}
|
||
manager.Subscribe(router)
|
||
manager.Subscribe(builder)
|
||
|
||
record, ok, err := manager.DispatchAdd(context.Background(), "tenant-a")
|
||
if !ok || err != nil {
|
||
t.Fatalf("expected dispatch add success, ok=%v err=%v", ok, err)
|
||
}
|
||
if router.addCalls != 1 || builder.addCalls != 1 {
|
||
t.Fatalf("expected both subscribers to receive add call")
|
||
}
|
||
if record.Phase != NamespacePhaseActive {
|
||
t.Fatalf("expected active phase after successful dispatch, got %s", record.Phase)
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerDispatchResyncFailure(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
builder := &testNamespaceSubscriber{name: "buildermgr", resyncErr: errors.New("resync failed")}
|
||
manager.Subscribe(router)
|
||
manager.Subscribe(builder)
|
||
|
||
record, ok, err := manager.DispatchResync(context.Background(), "tenant-a")
|
||
if !ok || err == nil {
|
||
t.Fatalf("expected dispatch resync failure, ok=%v err=%v", ok, err)
|
||
}
|
||
if builder.resyncCalls != 1 {
|
||
t.Fatalf("expected failing subscriber to receive resync call")
|
||
}
|
||
if record.Phase != NamespacePhaseFailed {
|
||
t.Fatalf("expected failed phase after dispatch error, got %s", record.Phase)
|
||
}
|
||
if record.RegisteredParts["buildermgr"].LastError != "resync failed" {
|
||
t.Fatalf("expected subscriber error to be stored")
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerDispatchRemove(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
builder := &testNamespaceSubscriber{name: "buildermgr"}
|
||
manager.Subscribe(router)
|
||
manager.Subscribe(builder)
|
||
|
||
record, ok, err := manager.DispatchRemove(context.Background(), "tenant-a")
|
||
if !ok || err != nil {
|
||
t.Fatalf("expected dispatch remove success, ok=%v err=%v", ok, err)
|
||
}
|
||
if router.removeCalls != 1 || builder.removeCalls != 1 {
|
||
t.Fatalf("expected both subscribers to receive remove call")
|
||
}
|
||
if record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected removed phase after dispatch remove, got %s", record.Phase)
|
||
}
|
||
}
|
||
|
||
func TestNamespaceManagerDispatchRemoveFailure(t *testing.T) {
|
||
manager := NewNamespaceManager()
|
||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||
router := &testNamespaceSubscriber{name: "router", removeErr: errors.New("remove failed")}
|
||
builder := &testNamespaceSubscriber{name: "buildermgr"}
|
||
manager.Subscribe(router)
|
||
manager.Subscribe(builder)
|
||
|
||
record, ok, err := manager.DispatchRemove(context.Background(), "tenant-a")
|
||
if !ok || err == nil {
|
||
t.Fatalf("expected dispatch remove failure, ok=%v err=%v", ok, err)
|
||
}
|
||
if record.Phase != NamespacePhaseRemoved {
|
||
t.Fatalf("expected removed phase even on dispatch remove error, got %s", record.Phase)
|
||
}
|
||
}
|
||
|
||
// TestStartManagedNamespaceWatcherIntegration проверяет полный маршрут:
|
||
// fake k8s client → информер → AddFunc → subscriber.OnNamespaceAdd.
|
||
// Это интеграционный тест без real cluster — использует k8sfake.
|
||
func TestStartManagedNamespaceWatcherIntegration(t *testing.T) {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
defer cancel()
|
||
|
||
fakeClient := k8sfake.NewSimpleClientset()
|
||
mgr := managerPkg.New()
|
||
router := &testNamespaceSubscriber{name: "router"}
|
||
|
||
// Запускаем watcher — не Bootstrap, только informer.
|
||
// config.Namespaces пусто, чтобы не было pre-loaded namespace-ов.
|
||
manager, err := RunManagedNamespaceWatcher(ctx, zap.NewNop(), fakeClient, mgr, ManagedNamespaceWatcherConfig{
|
||
Component: "router.NSWatcher",
|
||
Namespaces: nil,
|
||
RemovalStrategy: NamespaceRemovalStrategyTrackOnly,
|
||
Subscriber: router,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("RunManagedNamespaceWatcher failed: %v", err)
|
||
}
|
||
|
||
// Создаём namespace с managed label в fake client.
|
||
ns := &corev1.Namespace{}
|
||
ns.Name = "tenant-integration-a"
|
||
ns.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||
if _, err := fakeClient.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}); err != nil {
|
||
t.Fatalf("failed to create test namespace in fake client: %v", err)
|
||
}
|
||
|
||
// Ждём, пока informer обработает event и subscriber получит вызов.
|
||
// Timeout намеренно короткий (3 секунды) — fake client синхронный.
|
||
deadline := time.Now().Add(3 * time.Second)
|
||
for time.Now().Before(deadline) {
|
||
if router.addCalls >= 1 {
|
||
break
|
||
}
|
||
time.Sleep(10 * time.Millisecond)
|
||
}
|
||
|
||
if router.addCalls == 0 {
|
||
t.Fatalf("expected subscriber.OnNamespaceAdd to be called after namespace creation, got 0 calls")
|
||
}
|
||
if router.addCalls != 1 {
|
||
t.Fatalf("expected exactly 1 add call (namespace is new), got %d", router.addCalls)
|
||
}
|
||
|
||
// Namespace должен быть в manager snapshot.
|
||
snapshot := manager.Snapshot()
|
||
found := false
|
||
for _, name := range snapshot {
|
||
if name == ns.Name {
|
||
found = true
|
||
break
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatalf("expected %s in manager snapshot after informer event, got %v", ns.Name, snapshot)
|
||
}
|
||
|
||
record, ok := manager.Get(ns.Name)
|
||
if !ok {
|
||
t.Fatalf("expected record for %s in manager", ns.Name)
|
||
}
|
||
if record.Phase != NamespacePhaseActive {
|
||
t.Fatalf("expected active phase after informer add, got %s", record.Phase)
|
||
}
|
||
}
|
||
|
||
func TestNamespaceSubscriberFuncs(t *testing.T) {
|
||
addCalls := 0
|
||
removeCalls := 0
|
||
resyncCalls := 0
|
||
subscriber := NamespaceSubscriberFuncs{
|
||
SubscriberName: "router",
|
||
AddFunc: func(ctx context.Context, record NamespaceRecord) error {
|
||
addCalls++
|
||
return nil
|
||
},
|
||
RemoveFunc: func(ctx context.Context, record NamespaceRecord) error {
|
||
removeCalls++
|
||
return nil
|
||
},
|
||
ResyncFunc: func(ctx context.Context, record NamespaceRecord) error {
|
||
resyncCalls++
|
||
return nil
|
||
},
|
||
}
|
||
|
||
if subscriber.Name() != "router" {
|
||
t.Fatalf("expected subscriber name router")
|
||
}
|
||
_ = subscriber.OnNamespaceAdd(context.Background(), NamespaceRecord{Name: "tenant-a"})
|
||
_ = subscriber.OnNamespaceRemove(context.Background(), NamespaceRecord{Name: "tenant-a"})
|
||
_ = subscriber.OnNamespaceResync(context.Background(), NamespaceRecord{Name: "tenant-a"})
|
||
|
||
if addCalls != 1 || removeCalls != 1 || resyncCalls != 1 {
|
||
t.Fatalf("expected all functional callbacks to run once")
|
||
}
|
||
}
|