test(mediaserver): fetchIngressState

.mockery.yaml

@@ -9,3 +9,8 @@ packages:
   git.netflux.io/rob/termstream/container:
     interfaces:
       DockerClient:
+  git.netflux.io/rob/termstream/mediaserver:
+    interfaces:
+      httpClient:
+        config:
+          mockname: HTTPClient

generated/mocks/mediaserver/httpclient_mock.go

@@ -0,0 +1,94 @@
+// Code generated by mockery v2.52.2. DO NOT EDIT.
+
+package mediaserver
+
+import (
+	http "net/http"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// HTTPClient is an autogenerated mock type for the httpClient type
+type HTTPClient struct {
+	mock.Mock
+}
+
+type HTTPClient_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *HTTPClient) EXPECT() *HTTPClient_Expecter {
+	return &HTTPClient_Expecter{mock: &_m.Mock}
+}
+
+// Do provides a mock function with given fields: _a0
+func (_m *HTTPClient) Do(_a0 *http.Request) (*http.Response, error) {
+	ret := _m.Called(_a0)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Do")
+	}
+
+	var r0 *http.Response
+	var r1 error
+	if rf, ok := ret.Get(0).(func(*http.Request) (*http.Response, error)); ok {
+		return rf(_a0)
+	}
+	if rf, ok := ret.Get(0).(func(*http.Request) *http.Response); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*http.Response)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(*http.Request) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// HTTPClient_Do_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Do'
+type HTTPClient_Do_Call struct {
+	*mock.Call
+}
+
+// Do is a helper method to define mock.On call
+//   - _a0 *http.Request
+func (_e *HTTPClient_Expecter) Do(_a0 interface{}) *HTTPClient_Do_Call {
+	return &HTTPClient_Do_Call{Call: _e.mock.On("Do", _a0)}
+}
+
+func (_c *HTTPClient_Do_Call) Run(run func(_a0 *http.Request)) *HTTPClient_Do_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(*http.Request))
+	})
+	return _c
+}
+
+func (_c *HTTPClient_Do_Call) Return(_a0 *http.Response, _a1 error) *HTTPClient_Do_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *HTTPClient_Do_Call) RunAndReturn(run func(*http.Request) (*http.Response, error)) *HTTPClient_Do_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// NewHTTPClient creates a new instance of HTTPClient. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewHTTPClient(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *HTTPClient {
+	mock := &HTTPClient{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

mediaserver/actor.go

@@ -186,7 +186,7 @@ func (s *Actor) actorLoop(containerStateC <-chan domain.Container, errC <-chan e
 
 			sendState()
 		case <-fetchStateT.C:
-			ingressState, err := s.fetchIngressState()
+			ingressState, err := fetchIngressState(s.apiURL(), s.httpClient)
 			if err != nil {
 				s.logger.Error("Error fetching server state", "error", err)
 				continue

mediaserver/api.go

@@ -8,6 +8,10 @@ import (
 	"time"
 )
 
+type httpClient interface {
+	Do(*http.Request) (*http.Response, error)
+}
+
 type apiResponse[T any] struct {
 	Items []T `json:"items"`
 }
@@ -27,13 +31,13 @@ type ingressStreamState struct {
 	listeners int
 }
 
-func (s *Actor) fetchIngressState() (state ingressStreamState, _ error) {
+func fetchIngressState(apiURL string, httpClient httpClient) (state ingressStreamState, _ error) {
-	req, err := http.NewRequest(http.MethodGet, s.apiURL(), nil)
+	req, err := http.NewRequest(http.MethodGet, apiURL, nil)
 	if err != nil {
 		return state, fmt.Errorf("new request: %w", err)
 	}
 
-	httpResp, err := s.httpClient.Do(req)
+	httpResp, err := httpClient.Do(req)
 	if err != nil {
 		return state, fmt.Errorf("do request: %w", err)
 	}

mediaserver/api_test.go

@@ -0,0 +1,91 @@
+package mediaserver
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"net/http"
+	"testing"
+
+	mocks "git.netflux.io/rob/termstream/generated/mocks/mediaserver"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFetchIngressState(t *testing.T) {
+	const URL = "http://localhost:8989/v3/rtmpconns/list"
+
+	testCases := []struct {
+		name         string
+		httpResponse *http.Response
+		httpError    error
+		wantState    ingressStreamState
+		wantErr      error
+	}{
+		{
+			name:         "non-200 status",
+			httpResponse: &http.Response{StatusCode: http.StatusNotFound},
+			wantErr:      errors.New("unexpected status code: 404"),
+		},
+		{
+			name: "unparseable response",
+			httpResponse: &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       io.NopCloser(bytes.NewReader([]byte("invalid json"))),
+			},
+			wantErr: errors.New("unmarshal: invalid character 'i' looking for beginning of value"),
+		},
+		{
+			name: "successful response, no streams",
+			httpResponse: &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       io.NopCloser(bytes.NewReader([]byte(`{"itemCount":0,"pageCount":0,"items":[]}`))),
+			},
+			wantState: ingressStreamState{ready: false, listeners: 0},
+		},
+		{
+			name: "successful response, not yet ready",
+			httpResponse: &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       io.NopCloser(bytes.NewReader([]byte(`{"itemCount":1,"pageCount":1,"items":[{"id":"d2953cf8-9cd6-4c30-816f-807b80b6a71f","created":"2025-02-15T08:19:00.616220354Z","remoteAddr":"172.17.0.1:32972","state":"publish","path":"live","query":"","bytesReceived":15462,"bytesSent":3467}]}`))),
+			},
+			wantState: ingressStreamState{ready: false, listeners: 0},
+		},
+		{
+			name: "successful response, ready",
+			httpResponse: &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       io.NopCloser(bytes.NewReader([]byte(`{"itemCount":1,"pageCount":1,"items":[{"id":"d2953cf8-9cd6-4c30-816f-807b80b6a71f","created":"2025-02-15T08:19:00.616220354Z","remoteAddr":"172.17.0.1:32972","state":"publish","path":"live","query":"","bytesReceived":27832,"bytesSent":3467}]}`))),
+			},
+			wantState: ingressStreamState{ready: true, listeners: 0},
+		},
+		{
+			name: "successful response, ready, with listeners",
+			httpResponse: &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       io.NopCloser(bytes.NewReader([]byte(`{"itemCount":2,"pageCount":1,"items":[{"id":"12668315-0572-41f1-8384-fe7047cc73be","created":"2025-02-15T08:23:43.836589664Z","remoteAddr":"172.17.0.1:40026","state":"publish","path":"live","query":"","bytesReceived":7180753,"bytesSent":3467},{"id":"079370fd-43bb-4798-b079-860cc3159e4e","created":"2025-02-15T08:24:32.396794364Z","remoteAddr":"192.168.48.3:44736","state":"read","path":"live","query":"","bytesReceived":333435,"bytesSent":24243}]}`))),
+			},
+			wantState: ingressStreamState{ready: true, listeners: 1},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var httpClient mocks.HTTPClient
+			httpClient.
+				EXPECT().
+				Do(mock.MatchedBy(func(req *http.Request) bool {
+					return req.URL.String() == URL && req.Method == http.MethodGet
+				})).
+				Return(tc.httpResponse, tc.httpError)
+
+			state, err := fetchIngressState(URL, &httpClient)
+			if tc.wantErr != nil {
+				require.EqualError(t, err, tc.wantErr.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tc.wantState, state)
+			}
+		})
+	}
+}