…lines 1–17 unchanged (17 lines)
18
19import (
20 "context"
21 "math/rand"
22 "strings"
23 "testing"
24 "time"
25
26 "github.com/stretchr/testify/assert"
27 "github.com/stretchr/testify/mock"
28 "github.com/stretchr/testify/require"
29 "google.golang.org/grpc"
30 "google.golang.org/protobuf/proto"
31 v1 "k8s.io/api/core/v1"
32 "k8s.io/apimachinery/pkg/api/apitesting/fuzzer"
33 apiequality "k8s.io/apimachinery/pkg/api/equality"
34 metafuzzer "k8s.io/apimachinery/pkg/apis/meta/fuzzer"
35 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
…lines 36–56 unchanged (21 lines)
57 server.OnPodStatusUpdated(pod1, v1.PodStatus{Phase: v1.PodSucceeded})
58 server.OnPodRemoved(pod1)
59
60 event := <-clientChannel
61 assert.Equal(t, "ADDED", string(event.Type))
62 assert.Equal(t, pod1.UID, event.UID)
63 assert.Equal(t, pod1, event.Pod)
64
65 event = <-clientChannel
66 assert.Equal(t, "MODIFIED", string(event.Type))
67 assert.Equal(t, pod1.UID, event.UID)
68 expectedPod := *pod1
69 expectedPod.Status = v1.PodStatus{Phase: v1.PodSucceeded}
70 assert.Equal(t, &expectedPod, event.Pod)
71
72 event = <-clientChannel
73 assert.Equal(t, "DELETED", string(event.Type))
74 assert.Equal(t, pod1.UID, event.UID)
75 assert.Equal(t, pod1, event.Pod)
76
77 select {
78 case event := <-clientChannel:
79 t.Errorf("Unexpected event received: %v", event)
80 default:
81 }
82}
83
84type MockWatchPodsServer struct {
85 grpc.ServerStream
86 Ctx context.Context
87 EventCh chan *podsv1alpha1.WatchPodsEvent
88}
89
90func (m *MockWatchPodsServer) Send(event *podsv1alpha1.WatchPodsEvent) error {
91 select {
92 case m.EventCh <- event:
93 return nil
94 case <-m.Ctx.Done():
95 return m.Ctx.Err()
96 }
97}
98
99func (m *MockWatchPodsServer) Context() context.Context {
100 return m.Ctx
101}
102
103func TestWatchPods(t *testing.T) {
104 broadcaster := podsapi.NewBroadcaster()
105 mockManager := new(kubepodtest.MockManager)
106 server := podsapi.NewPodsServerForTest(broadcaster, mockManager)
107 pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}}
108
109 mockManager.On("GetPods").Return([]*v1.Pod{pod1})
110 mockManager.On("GetPodByUID", types.UID("pod1-uid")).Return(pod1, true)
111
112 ctx, cancel := context.WithCancel(context.Background())
113 defer cancel()
114
115 mockStream := &MockWatchPodsServer{
116 Ctx: ctx,
117 EventCh: make(chan *podsv1alpha1.WatchPodsEvent, 10),
118 }
119
120 go func() {
121 server.WatchPods(&podsv1alpha1.WatchPodsRequest{}, mockStream)
122 }()
123
124 // Verify initial ADDED event
125 event := <-mockStream.EventCh
126 assert.Equal(t, podsv1alpha1.EventType_ADDED, event.Type)
127 podOut := &v1.Pod{}
128 err := podOut.Unmarshal(event.Pod)
129 require.NoError(t, err)
130 assert.Equal(t, "pod1", podOut.Name)
131
132 // Verify INITIAL_SYNC event
133 event = <-mockStream.EventCh
134 assert.Equal(t, podsv1alpha1.EventType_INITIAL_SYNC_COMPLETE, event.Type)
135 assert.Empty(t, event.Pod)
136
137 // Trigger an update
138 server.OnPodUpdated(pod1)
139
140 // Verify MODIFIED event
141 event = <-mockStream.EventCh
142 assert.Equal(t, podsv1alpha1.EventType_MODIFIED, event.Type)
143 podOut = &v1.Pod{}
144 err = podOut.Unmarshal(event.Pod)
145 require.NoError(t, err)
146 assert.Equal(t, "pod1", podOut.Name)
147}
148
149func TestOnPodStatusUpdatedOverlaysStatus(t *testing.T) {
150 broadcaster := podsapi.NewBroadcaster()
151 mockManager := new(kubepodtest.MockManager)
152 server := podsapi.NewPodsServerForTest(broadcaster, mockManager)
153
154 // The pod object as it might exist in the manager (or passed by caller)
155 // It has Stale status.
156 pod := &v1.Pod{
157 ObjectMeta: metav1.ObjectMeta{UID: "pod-1", Name: "pod-1", Namespace: "ns"},
158 Spec: v1.PodSpec{NodeName: "worker-node"},
159 Status: v1.PodStatus{Phase: v1.PodPending},
160 }
161
162 // The new status that is being reported
163 newStatus := v1.PodStatus{Phase: v1.PodRunning, Message: "Pod is running"}
164
165 clientChannel := make(chan podsapi.PodWatchEvent, 1)
166 broadcaster.Register(clientChannel)
167 defer broadcaster.Unregister(clientChannel)
168
169 // Trigger the status update
170 server.OnPodStatusUpdated(pod, newStatus)
171
172 select {
173 case event := <-clientChannel:
174 assert.Equal(t, "MODIFIED", string(event.Type))
175 assert.Equal(t, "pod-1", event.Pod.Name)
176
177 // The status in the event should be the New Status (Running)
178 // NOT the status from the pod object (Pending)
179 assert.Equal(t, v1.PodRunning, event.Pod.Status.Phase)
180 assert.Equal(t, "Pod is running", event.Pod.Status.Message)
181
182 // Verify we didn't mutate the original pod
183 assert.Equal(t, v1.PodPending, pod.Status.Phase, "Original pod object should not be mutated")
184
185 case <-time.After(1 * time.Second):
186 t.Fatal("Timeout waiting for event")
187 }
188}
189
190func TestListPods(t *testing.T) {
191 broadcaster := podsapi.NewBroadcaster()
192 mockManager := new(kubepodtest.MockManager)
193 server := podsapi.NewPodsServerForTest(broadcaster, mockManager)
194 pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}}
…lines 195–331 unchanged (137 lines)
332 if !apiequality.Semantic.DeepEqual(pod, pod2) {
333 t.Fatal("round-tripped objects were different")
334 }
335 }
336}
337
338type MockStatusProvider struct {
339 mock.Mock
340}
341
342func (m *MockStatusProvider) GetPodStatus(uid types.UID) (v1.PodStatus, bool) {
343 args := m.Called(uid)
344 return args.Get(0).(v1.PodStatus), args.Bool(1)
345}
346
347func TestOnPodUpdatedUsesStatusProvider(t *testing.T) {
348 broadcaster := podsapi.NewBroadcaster()
349 mockManager := new(kubepodtest.MockManager)
350 server := podsapi.NewPodsServerForTest(broadcaster, mockManager)
351 mockStatusProvider := new(MockStatusProvider)
352 server.SetStatusProvider(mockStatusProvider)
353
354 // Pod with "Stale" status
355 pod := &v1.Pod{
356 ObjectMeta: metav1.ObjectMeta{UID: "pod-1", Name: "pod-1", Namespace: "ns"},
357 Status: v1.PodStatus{Phase: v1.PodPending},
358 }
359
360 // Status provider has "Fresh" status
361 freshStatus := v1.PodStatus{Phase: v1.PodRunning}
362 mockStatusProvider.On("GetPodStatus", pod.UID).Return(freshStatus, true)
363
364 clientChannel := make(chan podsapi.PodWatchEvent, 1)
365 broadcaster.Register(clientChannel)
366 defer broadcaster.Unregister(clientChannel)
367
368 server.OnPodUpdated(pod)
369
370 select {
371 case event := <-clientChannel:
372 assert.Equal(t, "MODIFIED", string(event.Type))
373 assert.Equal(t, v1.PodRunning, event.Pod.Status.Phase)
374 case <-time.After(1 * time.Second):
375 t.Fatal("Timeout waiting for event")
376 }
377
378 mockStatusProvider.AssertExpectations(t)
379}
380