Documentation
¶
Overview ¶
Package testserver provides an in-process fake admin server used in agent integration tests. It implements just enough of AgentService to observe what the agent sends, drive subscribe/unsubscribe scenarios, and exercise reconnect paths.
The package is internal so it cannot be imported outside admin/agent; production code MUST NOT depend on it.
Index ¶
- type Server
- func (s *Server) Addr() net.Addr
- func (s *Server) Close()
- func (s *Server) RejectedStreams() int
- func (s *Server) Sessions() []*StreamSession
- func (s *Server) SetAuthReject(reject bool)
- func (s *Server) SetHealthzFail(fail bool)
- func (s *Server) Stream(ctx context.Context, stream *connect.BidiStream[adminv1.Frame, adminv1.Frame]) error
- func (s *Server) URL() string
- func (s *Server) WaitForRegistration(d time.Duration) (*adminv1.NodeRegistration, error)
- type StreamSession
- func (sess *StreamSession) Closed() <-chan struct{}
- func (sess *StreamSession) EventCount() int
- func (sess *StreamSession) EventsCh() <-chan *adminv1.Event
- func (sess *StreamSession) Goodbye() string
- func (sess *StreamSession) HeartbeatCount() int
- func (sess *StreamSession) SendGoodbye(reason string) error
- func (sess *StreamSession) SendSnapshotRequest(requestID string, t adminv1.SnapshotType) error
- func (sess *StreamSession) SendSubscribe(id string, filter *adminv1.Filter, rates map[string]float32) error
- func (sess *StreamSession) SendUnsubscribe(id string) error
- func (sess *StreamSession) WaitForEvent(d time.Duration) (*adminv1.Event, error)
- func (sess *StreamSession) WaitForRegistration(d time.Duration) (*adminv1.NodeRegistration, error)
- func (sess *StreamSession) WaitForSnapshotResponse(d time.Duration) (*adminv1.SnapshotResponse, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is a fake admin server. Construct with Start, drop with Close. Use the methods on Server to observe registrations, drive subscribes, and watch incoming events.
func Start ¶
func Start() *Server
Start launches a new fake server on a fresh ephemeral port. The returned Server is ready to accept connections; agents should be pointed at Server.URL().
func (*Server) Addr ¶
Listener returns the bound TCP address (host:port). Useful for tests that want to break the connection by closing the listener manually.
func (*Server) Close ¶
func (s *Server) Close()
Close shuts the server down. Safe to call from defer.
func (*Server) RejectedStreams ¶
RejectedStreams returns how many non-/healthz requests have been answered with 401 so far (one per agent stream attempt).
func (*Server) Sessions ¶
func (s *Server) Sessions() []*StreamSession
Sessions returns a snapshot of sessions opened against this server.
func (*Server) SetAuthReject ¶
SetAuthReject toggles 401-for-everything-but-/healthz, mimicking the real admin server rejecting the agent's token. Established streams are not torn down; use SendGoodbye on the live session to force a reconnect into the rejecting server.
func (*Server) SetHealthzFail ¶
SetHealthzFail toggles /healthz between 200 and 503. While failing, the agent's dialer skips this endpoint and fails over to the next one — use it to steer a multi-endpoint agent at a chosen server.
func (*Server) Stream ¶
func (s *Server) Stream(ctx context.Context, stream *connect.BidiStream[adminv1.Frame, adminv1.Frame]) error
Stream is the AgentService.Stream handler.
func (*Server) URL ¶
URL returns the URL agents should connect to (e.g. http://127.0.0.1:NNNN).
func (*Server) WaitForRegistration ¶
WaitForRegistration blocks until any agent sends a NodeRegistration on any stream, or the deadline elapses.
type StreamSession ¶
type StreamSession struct {
// contains filtered or unexported fields
}
StreamSession represents one bidi stream the agent has opened. Methods are goroutine-safe and may be called from the test goroutine.
func (*StreamSession) Closed ¶
func (sess *StreamSession) Closed() <-chan struct{}
Closed returns a channel that is closed once the session has ended (either side closed the stream).
func (*StreamSession) EventCount ¶
func (sess *StreamSession) EventCount() int
EventCount returns the count of events received so far.
func (*StreamSession) EventsCh ¶
func (sess *StreamSession) EventsCh() <-chan *adminv1.Event
EventsCh exposes the raw events channel. Tests use it for select-form checks like "no event arrives within X". Each item is owned by the caller; the testserver does not retain references after sending.
func (*StreamSession) Goodbye ¶
func (sess *StreamSession) Goodbye() string
Goodbye returns the reason string if the agent sent a Goodbye, or "".
func (*StreamSession) HeartbeatCount ¶
func (sess *StreamSession) HeartbeatCount() int
HeartbeatCount returns the number of heartbeats received so far.
func (*StreamSession) SendGoodbye ¶
func (sess *StreamSession) SendGoodbye(reason string) error
SendGoodbye pushes a server-initiated Goodbye and closes the session.
func (*StreamSession) SendSnapshotRequest ¶
func (sess *StreamSession) SendSnapshotRequest(requestID string, t adminv1.SnapshotType) error
SendSnapshotRequest routes a SnapshotRequest command to the agent.
func (*StreamSession) SendSubscribe ¶
func (sess *StreamSession) SendSubscribe(id string, filter *adminv1.Filter, rates map[string]float32) error
SendSubscribe pushes a Subscribe command to the latest session.
func (*StreamSession) SendUnsubscribe ¶
func (sess *StreamSession) SendUnsubscribe(id string) error
SendUnsubscribe pushes an Unsubscribe command.
func (*StreamSession) WaitForEvent ¶
WaitForEvent blocks until the session receives an event from the agent or the timeout elapses.
func (*StreamSession) WaitForRegistration ¶
func (sess *StreamSession) WaitForRegistration(d time.Duration) (*adminv1.NodeRegistration, error)
WaitForRegistration blocks for this specific session.
func (*StreamSession) WaitForSnapshotResponse ¶
func (sess *StreamSession) WaitForSnapshotResponse(d time.Duration) (*adminv1.SnapshotResponse, error)
WaitForSnapshotResponse blocks until the agent answers a SnapshotRequest or the timeout elapses.
Source Files