Files
QiuSW 12857fdf32
Harness governance / validate (push) Has been cancelled
Harness governance / validate (pull_request) Has been cancelled
feat(sense): add reconciliation safety controls [T-012]
2026-08-07 23:00:03 +08:00

233 lines
6.8 KiB
Go

package mtx
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"sort"
"strconv"
"strings"
"sync"
"testing"
)
type fakeMediaMTX struct {
mu sync.Mutex
paths map[string]string
mutations int
}
func (f *fakeMediaMTX) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
f.mu.Lock()
defer f.mu.Unlock()
writer.Header().Set("Content-Type", "application/json")
if request.URL.Path == "/v3/config/paths/list" {
page, _ := strconv.Atoi(request.URL.Query().Get("page"))
limit, _ := strconv.Atoi(request.URL.Query().Get("itemsPerPage"))
if limit <= 0 {
limit = 100
}
names := make([]string, 0, len(f.paths))
for name := range f.paths {
names = append(names, name)
}
sort.Strings(names)
start := page * limit
if start > len(names) {
start = len(names)
}
end := start + limit
if end > len(names) {
end = len(names)
}
items := make([]map[string]any, 0, end-start)
for _, name := range names[start:end] {
items = append(items, map[string]any{"name": name})
}
pageCount := 0
if len(names) > 0 {
pageCount = (len(names) + limit - 1) / limit
}
_ = json.NewEncoder(writer).Encode(map[string]any{
"itemCount": len(names), "pageCount": pageCount, "items": items,
})
return
}
prefixes := map[string]string{
"/v3/config/paths/get/": "get",
"/v3/config/paths/add/": "add",
"/v3/config/paths/patch/": "patch",
"/v3/config/paths/delete/": "delete",
"/v3/paths/get/": "runtime",
}
for prefix, operation := range prefixes {
if !strings.HasPrefix(request.URL.Path, prefix) {
continue
}
name := strings.TrimPrefix(request.URL.Path, prefix)
source, exists := f.paths[name]
switch operation {
case "get":
if !exists {
http.Error(writer, `{"error":"not found"}`, http.StatusNotFound)
return
}
_ = json.NewEncoder(writer).Encode(map[string]any{"name": name, "source": source})
case "add", "patch":
var body struct {
Source string `json:"source"`
}
if err := json.NewDecoder(request.Body).Decode(&body); err != nil {
http.Error(writer, `{}`, http.StatusBadRequest)
return
}
f.paths[name] = body.Source
f.mutations++
_, _ = writer.Write([]byte(`{}`))
case "delete":
if !exists {
http.Error(writer, `{"error":"not found"}`, http.StatusNotFound)
return
}
delete(f.paths, name)
f.mutations++
_, _ = writer.Write([]byte(`{}`))
case "runtime":
if !exists {
http.Error(writer, `{"error":"not found"}`, http.StatusNotFound)
return
}
_, _ = writer.Write([]byte(`{"online":true,"available":true}`))
}
return
}
http.NotFound(writer, request)
}
func TestListPathNamesUsesPaginationAndDoesNotReturnSources(t *testing.T) {
t.Parallel()
paths := make(map[string]string)
for index := 0; index < 205; index++ {
paths[fmt.Sprintf("camera-%03d", index)] = fmt.Sprintf("rtsp://secret.invalid/%d", index)
}
fake := &fakeMediaMTX{paths: paths}
server := httptest.NewServer(fake)
defer server.Close()
client, err := NewClient(server.URL, server.Client())
if err != nil {
t.Fatal(err)
}
values, err := client.ListPathNames(context.Background())
if err != nil {
t.Fatal(err)
}
if len(values) != 205 || values[0] != "camera-000" || values[204] != "camera-204" {
t.Fatalf("unexpected path inventory: len=%d first=%q last=%q", len(values), values[0], values[len(values)-1])
}
for _, value := range values {
if strings.Contains(value, "rtsp") || strings.Contains(value, "secret") {
t.Fatalf("source leaked from path inventory: %q", value)
}
}
}
func TestListPathNamesRejectsRepeatedPage(t *testing.T) {
t.Parallel()
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
writer.Header().Set("Content-Type", "application/json")
if request.URL.Path != "/v3/config/paths/list" {
http.NotFound(writer, request)
return
}
_, _ = writer.Write([]byte(`{"itemCount":2,"pageCount":2,"items":[{"name":"same"}]}`))
}))
defer server.Close()
client, err := NewClient(server.URL, server.Client())
if err != nil {
t.Fatal(err)
}
if _, err := client.ListPathNames(context.Background()); err == nil ||
!strings.Contains(err.Error(), "repeated page") {
t.Fatalf("repeated pagination was accepted: %v", err)
}
}
func TestGeneratedClientCreateReadDeleteMapping(t *testing.T) {
t.Parallel()
fake := &fakeMediaMTX{paths: make(map[string]string)}
server := httptest.NewServer(fake)
defer server.Close()
client, err := NewClient(server.URL, server.Client())
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
if err := client.CreatePath(ctx, "camera-1", "rtsp://media.invalid/camera-1"); err != nil {
t.Fatal(err)
}
path, err := client.GetPath(ctx, "camera-1")
if err != nil {
t.Fatal(err)
}
if path.Source != "rtsp://media.invalid/camera-1" {
t.Fatalf("unexpected source mapping: %+v", path)
}
if err := client.DeletePath(ctx, "camera-1"); err != nil {
t.Fatal(err)
}
if _, err := client.GetPath(ctx, "camera-1"); err != ErrPathNotFound {
t.Fatalf("expected not found after delete, got %v", err)
}
}
func TestEnsurePathIsIdempotentAndCanPatch(t *testing.T) {
t.Parallel()
fake := &fakeMediaMTX{paths: make(map[string]string)}
server := httptest.NewServer(fake)
defer server.Close()
client, err := NewClient(server.URL, server.Client())
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
changed, err := client.EnsurePath(ctx, "camera-2", "rtsp://media.invalid/first")
if err != nil || !changed {
t.Fatalf("first ensure: changed=%v err=%v", changed, err)
}
changed, err = client.EnsurePath(ctx, "camera-2", "rtsp://media.invalid/first")
if err != nil || changed {
t.Fatalf("second ensure must be idempotent: changed=%v err=%v", changed, err)
}
changed, err = client.EnsurePath(ctx, "camera-2", "rtsp://media.invalid/second")
if err != nil || !changed {
t.Fatalf("changed source must patch: changed=%v err=%v", changed, err)
}
if fake.mutations != 2 {
t.Fatalf("expected create + patch, got %d mutations", fake.mutations)
}
ready, err := client.PathReady(ctx, "camera-2")
if err != nil || !ready {
t.Fatalf("runtime probe: ready=%v err=%v", ready, err)
}
}
func TestAPIErrorDoesNotLeakSource(t *testing.T) {
t.Parallel()
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
writer.WriteHeader(http.StatusInternalServerError)
_, _ = fmt.Fprint(writer, `{"error":"upstream included a secret"}`)
}))
defer server.Close()
client, err := NewClient(server.URL, server.Client())
if err != nil {
t.Fatal(err)
}
secretSource := "rtsp://" + "user" + ":" + "redacted" + "@camera.invalid/live"
err = client.CreatePath(context.Background(), "camera", secretSource)
if err == nil || strings.Contains(err.Error(), secretSource) || strings.Contains(err.Error(), "secret") {
t.Fatalf("error must be redacted, got %v", err)
}
}