This is an automated email from the ASF dual-hosted git repository.
mfordjody pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new b3e99787 Supplement test coverage (#960)
b3e99787 is described below
commit b3e99787e02f95a80bd3187cfef8637cdec5072f
Author: mfordjody <[email protected]>
AuthorDate: Sat Jul 18 15:27:48 2026 +0800
Supplement test coverage (#960)
---
.codecov.yml | 9 +-
.../bootstrap/proxyless_grpc_controller_test.go | 88 ++++++++++++++++-
dubbod/discovery/pkg/xds/ads_test.go | 106 +++++++++++++++++++++
dubbod/discovery/pkg/xds/pushqueue_test.go | 48 ++++++++++
4 files changed, 245 insertions(+), 6 deletions(-)
diff --git a/.codecov.yml b/.codecov.yml
index 250ec99e..16aa80f6 100644
--- a/.codecov.yml
+++ b/.codecov.yml
@@ -25,14 +25,13 @@ coverage:
# Pull requests only
patch:
default:
- informational: true
- # Commits pushed to master should not make the overall
- # project coverage decrease:
- target: auto
+ informational: false
+ target: 60%
threshold: 0%
project:
default:
- informational: true
+ informational: false
+ target: auto
threshold: 0%
ignore:
diff --git a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
index b4ad480d..1898e1b4 100644
--- a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
+++ b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
@@ -32,6 +32,7 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
"github.com/apache/dubbo-kubernetes/pkg/grpcxds"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/inject"
"github.com/apache/dubbo-kubernetes/pkg/kube/krt"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
@@ -362,6 +363,74 @@ func TestProxylessGRPCRuntimeConfigNeedsUpdate(t
*testing.T) {
}
}
+func TestHandleRuntimeConfigUpdateEnqueuesManagedPods(t *testing.T) {
+ managed := func(name, namespace string) *corev1.Pod {
+ return &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+ Name: name,
+ Namespace: namespace,
+ Annotations: map[string]string{
+ inject.ProxylessInjectTemplatesAnnoName:
inject.ProxylessGRPCTemplateName,
+ },
+ }}
+ }
+
+ reconciled := make(chan types.NamespacedName, 3)
+ controller := &proxylessGRPCWorkloadController{
+ managedPods: staticProxylessPodIndexer{items: []any{
+ managed("zeta", "team-b"),
+ "not-a-pod",
+ nil,
+ &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name:
"unmanaged", Namespace: "team-a"}},
+ managed("alpha", "team-a"),
+ }},
+ }
+ controller.queue = controllers.NewQueue("proxyless runtime config test",
+ controllers.WithReconciler(func(key types.NamespacedName) error
{
+ reconciled <- key
+ return nil
+ }),
+ controllers.WithMaxAttempts(1),
+ )
+
+ controller.handleRuntimeConfigUpdate(&discoverymodel.PushRequest{
+ ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind:
kind.ConfigMap, Name: "unrelated"}),
+ })
+ controller.handleRuntimeConfigUpdate(&discoverymodel.PushRequest{
+ ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind:
kind.Service, Name: "provider", Namespace: "team-a"}),
+ })
+
+ stop := make(chan struct{})
+ done := make(chan struct{})
+ go func() {
+ controller.queue.Run(stop)
+ close(done)
+ }()
+ t.Cleanup(func() {
+ close(stop)
+ <-done
+ })
+
+ want := []types.NamespacedName{
+ {Name: "alpha", Namespace: "team-a"},
+ {Name: "zeta", Namespace: "team-b"},
+ }
+ for i, expected := range want {
+ select {
+ case got := <-reconciled:
+ if got != expected {
+ t.Fatalf("reconciled key %d = %v, want %v", i,
got, expected)
+ }
+ case <-time.After(time.Second):
+ t.Fatalf("timed out waiting for reconciled key %d", i)
+ }
+ }
+ select {
+ case got := <-reconciled:
+ t.Fatalf("unexpected reconciled key %v from unrelated update",
got)
+ case <-time.After(50 * time.Millisecond):
+ }
+}
+
func TestNextRotationTime(t *testing.T) {
now := time.Now()
rotations := map[types.NamespacedName]time.Time{
@@ -383,6 +452,11 @@ func TestNextRotationTime(t *testing.T) {
}
func newProxylessRuntimeTestPushContext(t *testing.T, configs []config.Config,
services []*discoverymodel.Service) *discoverymodel.PushContext {
+ _, push := newProxylessRuntimeTestEnvironment(t, configs, services)
+ return push
+}
+
+func newProxylessRuntimeTestEnvironment(t *testing.T, configs []config.Config,
services []*discoverymodel.Service) (*discoverymodel.Environment,
*discoverymodel.PushContext) {
t.Helper()
store := memory.Make(collections.DubboGatewayAPI())
@@ -402,7 +476,8 @@ func newProxylessRuntimeTestPushContext(t *testing.T,
configs []config.Config, s
push := discoverymodel.NewPushContext()
push.InitContext(env, nil, nil)
- return push
+ env.SetPushContext(push)
+ return env, push
}
func newProxylessRuntimeTestService(name, namespace, hostname string, port
int) *discoverymodel.Service {
@@ -446,6 +521,17 @@ type proxylessRuntimeStaticServiceDiscovery struct {
services []*discoverymodel.Service
}
+type staticProxylessPodIndexer struct {
+ items []any
+}
+
+func (s staticProxylessPodIndexer) Lookup(key string) []any {
+ if key != proxylessGRPCManagedPodIndexKey {
+ return nil
+ }
+ return s.items
+}
+
func (s proxylessRuntimeStaticServiceDiscovery) Services()
[]*discoverymodel.Service {
return s.services
}
diff --git a/dubbod/discovery/pkg/xds/ads_test.go
b/dubbod/discovery/pkg/xds/ads_test.go
index 7af2933c..830a6047 100644
--- a/dubbod/discovery/pkg/xds/ads_test.go
+++ b/dubbod/discovery/pkg/xds/ads_test.go
@@ -19,6 +19,7 @@ import (
"encoding/json"
"fmt"
"testing"
+ "time"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
v1 "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds/v1"
@@ -79,6 +80,111 @@ func
TestClientsForPushProxylessLargeScaleTargetsWatchedService(t *testing.T) {
}
}
+func TestPushConnectionExercisesProxylessPushPath(t *testing.T) {
+ server, con, stream := newProxylessXDSTestServer(t)
+ req := &model.PushRequest{
+ Full: true,
+ Push: con.proxy.LastPushContext,
+ Reason: model.NewReasonStats(model.ProxyRequest),
+ Start: time.Now(),
+ }
+ proxyNeedsPushCalled := false
+ server.ProxyNeedsPush = func(gotProxy *model.Proxy, gotReq
*model.PushRequest) (*model.PushRequest, bool) {
+ proxyNeedsPushCalled = true
+ if gotProxy != con.proxy || gotReq != req {
+ t.Fatalf("ProxyNeedsPush received proxy=%p request=%p,
want proxy=%p request=%p", gotProxy, gotReq, con.proxy, req)
+ }
+ return gotReq, true
+ }
+
+ if err := server.pushConnection(con, &Event{pushRequest: req}); err !=
nil {
+ t.Fatalf("pushConnection() failed: %v", err)
+ }
+ if !proxyNeedsPushCalled {
+ t.Fatal("ProxyNeedsPush was not called")
+ }
+ responses := stream.takeAllResponses(t, 3)
+ assertResponseTypeOrder(t, responses, v1.ListenerType, v1.ClusterType,
v1.RouteType)
+ if !con.proxy.LastPushTime.Equal(req.Start) {
+ t.Fatalf("LastPushTime = %v, want %v", con.proxy.LastPushTime,
req.Start)
+ }
+}
+
+func TestPushConnectionSkipsUnreadyAndUnneededPushes(t *testing.T) {
+ t.Run("uninitialized connection", func(t *testing.T) {
+ server, _, stream := newProxylessXDSTestServer(t)
+ con := &Connection{}
+ if err := server.pushConnection(con, &Event{pushRequest:
&model.PushRequest{Full: true}}); err != nil {
+ t.Fatalf("pushConnection() failed: %v", err)
+ }
+ stream.assertNoResponse(t)
+ })
+
+ t.Run("nil request", func(t *testing.T) {
+ server, con, stream := newProxylessXDSTestServer(t)
+ if err := server.pushConnection(con, &Event{}); err != nil {
+ t.Fatalf("pushConnection() failed: %v", err)
+ }
+ stream.assertNoResponse(t)
+ })
+
+ t.Run("proxy does not need push", func(t *testing.T) {
+ server, con, stream := newProxylessXDSTestServer(t)
+ server.ProxyNeedsPush = func(_ *model.Proxy, req
*model.PushRequest) (*model.PushRequest, bool) {
+ return req, false
+ }
+ if err := server.pushConnection(con, &Event{pushRequest:
&model.PushRequest{Push: con.proxy.LastPushContext}}); err != nil {
+ t.Fatalf("pushConnection() failed: %v", err)
+ }
+ stream.assertNoResponse(t)
+ })
+}
+
+func TestDiscoveryPushQueuesAndNotifies(t *testing.T) {
+ for _, full := range []bool{false, true} {
+ t.Run(fmt.Sprintf("full=%v", full), func(t *testing.T) {
+ server, con, _ := newProxylessXDSTestServer(t)
+ server.addCon("proxyless-1", con)
+
+ var notified *model.PushRequest
+ server.RuntimeConfigUpdate = func(req
*model.PushRequest) {
+ notified = req
+ }
+ req := &model.PushRequest{
+ Full: full,
+ Reason:
model.NewReasonStats(model.ConfigUpdate),
+ ConfigsUpdated: sets.New(model.ConfigKey{
+ Kind: kind.Service,
+ Name:
"nginx.app.svc.cluster.local",
+ Namespace: "app",
+ }),
+ }
+
+ server.Push(req)
+
+ queuedCon, queuedReq, shutdown :=
server.pushQueue.Dequeue()
+ if shutdown {
+ t.Fatal("push queue unexpectedly shut down")
+ }
+ if queuedCon != con || queuedReq != req {
+ t.Fatalf("queued connection/request = %p/%p,
want %p/%p", queuedCon, queuedReq, con, req)
+ }
+ if req.Push == nil || req.Start.IsZero() {
+ t.Fatalf("Push() left request without context
or start time: %+v", req)
+ }
+ if notified != req {
+ t.Fatalf("RuntimeConfigUpdate received %p, want
%p", notified, req)
+ }
+ if full && req.Push.PushVersion == "test-version" {
+ t.Fatalf("full Push() reused old push version
%q", req.Push.PushVersion)
+ }
+ if !full && req.Push != server.globalPushContext() {
+ t.Fatal("incremental Push() did not reuse the
global push context")
+ }
+ })
+ }
+}
+
func TestPushStatusJSONIncludesOperationalSummary(t *testing.T) {
env := model.NewEnvironment()
push := model.NewPushContext()
diff --git a/dubbod/discovery/pkg/xds/pushqueue_test.go
b/dubbod/discovery/pkg/xds/pushqueue_test.go
index b7f6d05a..a9521063 100644
--- a/dubbod/discovery/pkg/xds/pushqueue_test.go
+++ b/dubbod/discovery/pkg/xds/pushqueue_test.go
@@ -60,3 +60,51 @@ func TestPushQueueEnqueueMergesPendingRequest(t *testing.T) {
t.Fatalf("ConfigsUpdated = %v, want both config keys",
req.ConfigsUpdated)
}
}
+
+func TestPushQueueRequeuesUpdatesArrivingDuringProcessing(t *testing.T) {
+ queue := NewPushQueue()
+ con := &Connection{}
+ initial := &model.PushRequest{
+ Reason: model.NewReasonStats(model.EndpointUpdate),
+ }
+ queue.Enqueue(con, initial)
+
+ gotCon, gotReq, shutdown := queue.Dequeue()
+ if shutdown || gotCon != con || gotReq != initial {
+ t.Fatalf("first Dequeue() = %p, %p, %v; want %p, %p, false",
gotCon, gotReq, shutdown, con, initial)
+ }
+ queue.Enqueue(con, &model.PushRequest{
+ Full: true,
+ Reason: model.NewReasonStats(model.ConfigUpdate),
+ })
+ if stats := queue.Stats(); stats.Processing != 1 || stats.Pending != 0
|| stats.Queued != 0 {
+ t.Fatalf("Stats while processing = %+v, want processing=1
only", stats)
+ }
+
+ queue.MarkDone(con)
+ if stats := queue.Stats(); stats.Processing != 0 || stats.Pending != 1
|| stats.Queued != 1 {
+ t.Fatalf("Stats after MarkDone = %+v, want one requeued
request", stats)
+ }
+ _, merged, shutdown := queue.Dequeue()
+ if shutdown || !merged.Full || merged.Reason[model.ConfigUpdate] != 1 {
+ t.Fatalf("second Dequeue() = %+v, shutdown=%v; want full config
update", merged, shutdown)
+ }
+}
+
+func TestPushQueueShutdownUnblocksDequeueAndRejectsEnqueue(t *testing.T) {
+ queue := NewPushQueue()
+ result := make(chan bool, 1)
+ go func() {
+ _, _, shutdown := queue.Dequeue()
+ result <- shutdown
+ }()
+
+ queue.ShutDown()
+ if shutdown := <-result; !shutdown {
+ t.Fatal("Dequeue() shutdown = false, want true")
+ }
+ queue.Enqueue(&Connection{}, &model.PushRequest{})
+ if stats := queue.Stats(); stats != (pushQueueStats{}) {
+ t.Fatalf("Stats after shutdown enqueue = %+v, want empty",
stats)
+ }
+}
