test(app): exercise remote probe through the actual pool

commit: e4992c6b3b55ff367914197ba97fab529c765fcd
author: brice · 30d ago · 2026-09-22 21:26 UTC
parent: d60044d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

1 file · +20 -19

M internal/app/broker_remote_probe_shared_test.go +20 -19


@@ -5,25 +5,16 @@ import (
5	5		"errors"
6	6		"testing"
7	7
	8	+	"github.com/bnema/vev/internal/adapters/clock"
8	9		"github.com/bnema/vev/internal/domain"
9	10		"github.com/bnema/vev/internal/ports"
10	11		portsmocks "github.com/bnema/vev/internal/ports/mocks"
11	12		"github.com/bnema/vev/internal/protocol"
	13	+	"github.com/bnema/vev/internal/usecase/broker"
12	14		"github.com/stretchr/testify/mock"
13	15		"github.com/stretchr/testify/require"
14	16		)
15	17
16		-	// sharedPhysicalsFunc adapts a closure to the probe's pool lending seam.
17		-	type sharedPhysicalsFunc func(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
18		-
19		-	func (f sharedPhysicalsFunc) SharedPhysical(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
20		-	return f(identity, policy)
21		-	}
22		-
23		-	func (sharedPhysicalsFunc) AdoptPhysical(ports.BrokerPhysicalConnection, ports.BrokerPolicy) bool {
24		-	return false
25		-	}
26		-
27	18		// TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the
28	19		// broker pool's attached or warm transport instead of bootstrapping SSH/QUIC
29	20		// again, and redials only when the borrowed transport has retired.

@@ -69,8 +60,10 @@ func TestBrokerRemoteProbeBorrowsWarmTransport(t *testing.T) {
69	60		}
70	61		if tt.bound && tt.shared {
71	62		if tt.openErr != nil {
72		-	physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(nil, tt.openErr)
73		-	physical.EXPECT().Done().Return(done)
	63	+	if !tt.retired {
	64	+	physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(nil, tt.openErr)
	65	+	physical.EXPECT().Done().Return(done)
	66	+	}
74	67		} else {
75	68		logical := portsmocks.NewMockBrokerLogicalConnection(t)
76	69		var sent protocol.CommandRequest

@@ -91,12 +84,20 @@ func TestBrokerRemoteProbeBorrowsWarmTransport(t *testing.T) {
91	84		epoch: 7, routes: routes, binder: portsmocks.NewMockBrokerIdentityBinder(t), connector: connector,
92	85		hosts: &routeTestHosts{record: record, found: true},
93	86		}
94		-	probe.shared.share(sharedPhysicalsFunc(func(gotIdentity ports.BrokerDaemonIdentity, gotPolicy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
95		-	if !tt.shared \|\| gotIdentity != identity \|\| gotPolicy != policy {
96		-	return nil, false
97		-	}
98		-	return physical, true
99		-	}))
	87	+	// Use the real pool for borrowing. Stateful pool retirement is
	88	+	// covered with a fake physical in broker's worker tests.
	89	+	pool, err := broker.NewPool(7, routes, probe.binder, connector, clock.New(), broker.PoolLimits{Physical: 1, Clients: 1, Streams: 2, StreamsPerClient: 2, Warm: 1})
	90	+	require.NoError(t, err)
	91	+	t.Cleanup(func() { require.NoError(t, pool.Close()) })
	92	+	if tt.shared && tt.bound && !tt.retired {
	93	+	physical.EXPECT().Policy().Return(policy).Maybe()
	94	+	physical.EXPECT().Identity().Return(identity).Maybe()
	95	+	physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1}).Maybe()
	96	+	physical.EXPECT().Done().Return(done).Maybe()
	97	+	physical.EXPECT().Close().Return(nil).Maybe()
	98	+	require.True(t, pool.AdoptPhysical(physical, policy))
	99	+	}
	100	+	probe.shared.share(pool)
100	101
101	102		observation, err := probe.Probe(context.Background(), registration)
102	103		switch {

 	"errors"
 	"testing"
+	"github.com/bnema/vev/internal/adapters/clock"
 	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 	portsmocks "github.com/bnema/vev/internal/ports/mocks"
 	"github.com/bnema/vev/internal/protocol"
+	"github.com/bnema/vev/internal/usecase/broker"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
-// sharedPhysicalsFunc adapts a closure to the probe's pool lending seam.
-type sharedPhysicalsFunc func(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
-func (f sharedPhysicalsFunc) SharedPhysical(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
-	return f(identity, policy)
-}
-func (sharedPhysicalsFunc) AdoptPhysical(ports.BrokerPhysicalConnection, ports.BrokerPolicy) bool {
-	return false
-}
 // TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the
 // broker pool's attached or warm transport instead of bootstrapping SSH/QUIC
 // again, and redials only when the borrowed transport has retired.
 			}
 			if tt.bound && tt.shared {
 				if tt.openErr != nil {
-					physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(nil, tt.openErr)
+					if !tt.retired {
-					physical.EXPECT().Done().Return(done)
+						physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(nil, tt.openErr)
+						physical.EXPECT().Done().Return(done)
+					}
 				} else {
 					logical := portsmocks.NewMockBrokerLogicalConnection(t)
 					var sent protocol.CommandRequest
 				epoch: 7, routes: routes, binder: portsmocks.NewMockBrokerIdentityBinder(t), connector: connector,
 				hosts: &routeTestHosts{record: record, found: true},
 			}
-			probe.shared.share(sharedPhysicalsFunc(func(gotIdentity ports.BrokerDaemonIdentity, gotPolicy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
+			// Use the real pool for borrowing. Stateful pool retirement is
-				if !tt.shared || gotIdentity != identity || gotPolicy != policy {
+			// covered with a fake physical in broker's worker tests.
-					return nil, false
+			pool, err := broker.NewPool(7, routes, probe.binder, connector, clock.New(), broker.PoolLimits{Physical: 1, Clients: 1, Streams: 2, StreamsPerClient: 2, Warm: 1})
-				}
+			require.NoError(t, err)
-				return physical, true
+			t.Cleanup(func() { require.NoError(t, pool.Close()) })
-			}))
+			if tt.shared && tt.bound && !tt.retired {
+				physical.EXPECT().Policy().Return(policy).Maybe()
+				physical.EXPECT().Identity().Return(identity).Maybe()
+				physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1}).Maybe()
+				physical.EXPECT().Done().Return(done).Maybe()
+				physical.EXPECT().Close().Return(nil).Maybe()
+				require.True(t, pool.AdoptPhysical(physical, policy))
+			}
+			probe.shared.share(pool)
 			observation, err := probe.Probe(context.Background(), registration)
 			switch {