111 lines
3.6 KiB
Go
111 lines
3.6 KiB
Go
package media
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
|
|
"yovision.local/sense/app/sense/adapters/mediamtx"
|
|
"yovision.local/sense/app/sense/admission"
|
|
"yovision.local/sense/app/sense/device"
|
|
"yovision.local/sense/app/sense/identity"
|
|
)
|
|
|
|
type fakeProcess struct {
|
|
state mediamtx.ProcessState
|
|
startErr error
|
|
stopped bool
|
|
}
|
|
|
|
func (p *fakeProcess) Start(context.Context) error {
|
|
if p.startErr != nil {
|
|
return p.startErr
|
|
}
|
|
p.state.Running = true
|
|
p.state.PID = 123
|
|
return nil
|
|
}
|
|
func (p *fakeProcess) Stop(context.Context) error {
|
|
p.stopped = true
|
|
p.state.Running = false
|
|
p.state.PID = 0
|
|
return nil
|
|
}
|
|
func (p *fakeProcess) State() mediamtx.ProcessState { return p.state }
|
|
|
|
type fakeController struct {
|
|
applyErr error
|
|
status mediamtx.PathStatus
|
|
}
|
|
|
|
func (f fakeController) Apply(context.Context, mediamtx.Source) error { return f.applyErr }
|
|
func (f fakeController) Status(context.Context, string) (mediamtx.PathStatus, error) {
|
|
if f.status.Name == "error" {
|
|
return mediamtx.PathStatus{}, errors.New("timeout")
|
|
}
|
|
if !f.status.Ready {
|
|
return mediamtx.PathStatus{Ready: true, Readers: 2}, nil
|
|
}
|
|
return f.status, nil
|
|
}
|
|
|
|
func TestStoreAndProcessBoundary(t *testing.T) {
|
|
store := NewMemoryStore()
|
|
process := &fakeProcess{}
|
|
service := NewService(store, process, fakeController{})
|
|
route := Route{ID: "dev:main", DeviceID: "dev", ProfileToken: "main", Path: "sense_dev_main", Desired: "running", Actual: "pending", Version: 1}
|
|
if err := store.Save(context.Background(), route); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
items, err := service.List(context.Background())
|
|
if err != nil || len(items) != 1 {
|
|
t.Fatalf("items=%#v err=%v", items, err)
|
|
}
|
|
if process.State().Running {
|
|
t.Fatal("process started without explicit reconcile")
|
|
}
|
|
}
|
|
|
|
func preparedService(t *testing.T, process *fakeProcess, controller fakeController) (*Service, Route) {
|
|
t.Helper()
|
|
service := NewService(NewMemoryStore(), process, controller)
|
|
service.profile = func(string, string) (admission.Profile, error) {
|
|
return admission.Profile{StreamURI: "rtsp://camera.invalid/main"}, nil
|
|
}
|
|
service.credential = func(context.Context, string) (device.Credential, error) {
|
|
return device.Credential{Username: "fixture", Password: "fixture"}, nil
|
|
}
|
|
route, err := service.Configure(context.Background(), identity.Principal{}, "device", "main")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return service, route
|
|
}
|
|
|
|
func TestReconcileReadyAndStopOwnedProcess(t *testing.T) {
|
|
process := &fakeProcess{}
|
|
service, route := preparedService(t, process, fakeController{status: mediamtx.PathStatus{Ready: true, Readers: 2}})
|
|
result, err := service.Reconcile(context.Background(), identity.Principal{}, route.ID)
|
|
if err != nil || result.Actual != "ready" || result.Readers != 2 {
|
|
t.Fatalf("result=%#v err=%v", result, err)
|
|
}
|
|
if _, err := service.Stop(context.Background(), identity.Principal{}, route.ID); err != nil || !process.stopped {
|
|
t.Fatalf("stop err=%v stopped=%v", err, process.stopped)
|
|
}
|
|
}
|
|
|
|
func TestReconcileRecordsFailureStates(t *testing.T) {
|
|
process := &fakeProcess{startErr: errors.New("restart limit reached")}
|
|
service, route := preparedService(t, process, fakeController{})
|
|
result, err := service.Reconcile(context.Background(), identity.Principal{}, route.ID)
|
|
if err != nil || result.Actual != "process_failed" {
|
|
t.Fatalf("result=%#v err=%v", result, err)
|
|
}
|
|
process.startErr = nil
|
|
service.controller = fakeController{applyErr: errors.New("control timeout")}
|
|
result, err = service.Reconcile(context.Background(), identity.Principal{}, route.ID)
|
|
if err != nil || result.Actual != "apply_failed" {
|
|
t.Fatalf("result=%#v err=%v", result, err)
|
|
}
|
|
}
|