iop/apps/control-plane/cmd/control-plane/main_test.go

546 lines
17 KiB
Go

package main
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
"google.golang.org/protobuf/types/known/structpb"
"iop/apps/control-plane/internal/wire"
iop "iop/proto/gen/iop"
)
func TestLoadConfigDatabaseDefaultsToLocal(t *testing.T) {
t.Setenv("IOP_DATABASE_URL", "")
t.Setenv("IOP_REDIS_URL", "")
t.Setenv("IOP_REDIS_KEY_PREFIX", "")
cfg, err := loadConfig("")
if err != nil {
t.Fatalf("load config: %v", err)
}
const wantDatabase = "postgres://nomadcode:nomadcode@code-server-postgres:5432/iop-control-plane-local?sslmode=disable"
if cfg.Database.URL != wantDatabase {
t.Fatalf("database url: got %q want %q", cfg.Database.URL, wantDatabase)
}
const wantRedis = "redis://code-server-redis:6379/1"
if cfg.Redis.URL != wantRedis {
t.Fatalf("redis url: got %q want %q", cfg.Redis.URL, wantRedis)
}
const wantRedisKeyPrefix = "iop:control-plane:local:"
if cfg.Redis.KeyPrefix != wantRedisKeyPrefix {
t.Fatalf("redis key prefix: got %q want %q", cfg.Redis.KeyPrefix, wantRedisKeyPrefix)
}
}
func TestLoadConfigDatabaseFromYAML(t *testing.T) {
t.Setenv("IOP_DATABASE_URL", "")
t.Setenv("IOP_REDIS_URL", "")
t.Setenv("IOP_REDIS_KEY_PREFIX", "")
path := writeConfig(t, `
server:
listen: "127.0.0.1:9080"
database:
url: "postgres://user:pass@db:5432/iop-control-plane-local?sslmode=disable"
redis:
url: "redis://cache:6379/1"
key_prefix: "iop:test:"
`)
cfg, err := loadConfig(path)
if err != nil {
t.Fatalf("load config: %v", err)
}
const want = "postgres://user:pass@db:5432/iop-control-plane-local?sslmode=disable"
if cfg.Database.URL != want {
t.Fatalf("database url: got %q want %q", cfg.Database.URL, want)
}
if cfg.Redis.URL != "redis://cache:6379/1" {
t.Fatalf("redis url: got %q", cfg.Redis.URL)
}
if cfg.Redis.KeyPrefix != "iop:test:" {
t.Fatalf("redis key prefix: got %q", cfg.Redis.KeyPrefix)
}
}
func TestLoadConfigRepositoryLocalConfig(t *testing.T) {
t.Setenv("IOP_DATABASE_URL", "")
t.Setenv("IOP_REDIS_URL", "")
t.Setenv("IOP_REDIS_KEY_PREFIX", "")
t.Setenv("IOP_WIRE_LISTEN", "")
t.Setenv("IOP_EDGE_WIRE_LISTEN", "")
cfg, err := loadConfig("../../../../configs/control-plane.yaml")
if err != nil {
t.Fatalf("load config: %v", err)
}
if cfg.Redis.URL != "redis://code-server-redis:6379/1" {
t.Fatalf("redis url: got %q", cfg.Redis.URL)
}
if cfg.Redis.KeyPrefix != "iop:control-plane:local:" {
t.Fatalf("redis key prefix: got %q", cfg.Redis.KeyPrefix)
}
if cfg.Server.WireListen != "0.0.0.0:19080" {
t.Fatalf("wire_listen: got %q", cfg.Server.WireListen)
}
if cfg.Server.EdgeWireListen != "0.0.0.0:19081" {
t.Fatalf("edge_wire_listen: got %q", cfg.Server.EdgeWireListen)
}
}
func TestLoadConfigEnvOverrides(t *testing.T) {
path := writeConfig(t, `
database:
url: "postgres://user:pass@db:5432/iop-control-plane-local?sslmode=disable"
redis:
url: "redis://cache:6379/1"
key_prefix: "iop:control-plane:local:"
`)
const wantDatabase = "postgres://user:pass@db:5432/iop-control-plane-dev?sslmode=disable"
const wantRedis = "redis://redis:6379/2"
const wantRedisKeyPrefix = "iop:control-plane:dev:"
t.Setenv("IOP_DATABASE_URL", wantDatabase)
t.Setenv("IOP_REDIS_URL", wantRedis)
t.Setenv("IOP_REDIS_KEY_PREFIX", wantRedisKeyPrefix)
cfg, err := loadConfig(path)
if err != nil {
t.Fatalf("load config: %v", err)
}
if cfg.Database.URL != wantDatabase {
t.Fatalf("database url: got %q want %q", cfg.Database.URL, wantDatabase)
}
if cfg.Redis.URL != wantRedis {
t.Fatalf("redis url: got %q want %q", cfg.Redis.URL, wantRedis)
}
if cfg.Redis.KeyPrefix != wantRedisKeyPrefix {
t.Fatalf("redis key prefix: got %q want %q", cfg.Redis.KeyPrefix, wantRedisKeyPrefix)
}
}
func TestDatabaseLogFieldsDoNotExposeCredentials(t *testing.T) {
fields := databaseLogFields("postgres://user:secret@db.example:5432/iop-control-plane-local?sslmode=disable")
var host, database string
for _, field := range fields {
switch field.Key {
case "host":
host = field.String
case "database":
database = field.String
}
}
if host != "db.example:5432" {
t.Fatalf("host field: got %q", host)
}
if database != "iop-control-plane-local" {
t.Fatalf("database field: got %q", database)
}
if strings.Contains(fmt.Sprint(fields), "secret") {
t.Fatal("secret leaked into log fields")
}
}
func TestRedisLogFields(t *testing.T) {
fields := redisLogFields("redis://:secret@cache.example:6379/2", "iop:control-plane:dev:")
var host, database, keyPrefix string
for _, field := range fields {
switch field.Key {
case "host":
host = field.String
case "database":
database = field.String
case "key_prefix":
keyPrefix = field.String
}
}
if host != "cache.example:6379" {
t.Fatalf("host field: got %q", host)
}
if database != "2" {
t.Fatalf("database field: got %q", database)
}
if keyPrefix != "iop:control-plane:dev:" {
t.Fatalf("key_prefix field: got %q", keyPrefix)
}
if strings.Contains(fmt.Sprint(fields), "secret") {
t.Fatal("secret leaked into log fields")
}
}
func writeConfig(t *testing.T, body string) string {
t.Helper()
path := filepath.Join(t.TempDir(), "control-plane.yaml")
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatalf("write config: %v", err)
}
return path
}
func TestLoadConfigWireListenEnvOverrides(t *testing.T) {
path := writeConfig(t, `
server:
listen: "0.0.0.0:9080"
wire_listen: "0.0.0.0:19080"
`)
const wantListen = "127.0.0.1:9081"
const wantWireListen = "127.0.0.1:19081"
t.Setenv("IOP_LISTEN", wantListen)
t.Setenv("IOP_WIRE_LISTEN", wantWireListen)
cfg, err := loadConfig(path)
if err != nil {
t.Fatalf("load config: %v", err)
}
if cfg.Server.Listen != wantListen {
t.Fatalf("listen: got %q want %q", cfg.Server.Listen, wantListen)
}
if cfg.Server.WireListen != wantWireListen {
t.Fatalf("wire_listen: got %q want %q", cfg.Server.WireListen, wantWireListen)
}
}
func TestLoadConfigEdgeWireListenDefault(t *testing.T) {
t.Setenv("IOP_EDGE_WIRE_LISTEN", "")
cfg, err := loadConfig("")
if err != nil {
t.Fatalf("load config: %v", err)
}
if cfg.Server.EdgeWireListen != "0.0.0.0:19081" {
t.Fatalf("edge_wire_listen default: got %q want %q", cfg.Server.EdgeWireListen, "0.0.0.0:19081")
}
if cfg.Server.EdgeWireListen == cfg.Server.WireListen {
t.Fatalf("edge_wire_listen must not share the client wire_listen address %q", cfg.Server.WireListen)
}
}
func TestLoadConfigEdgeWireListenEnvOverrides(t *testing.T) {
path := writeConfig(t, `
server:
listen: "0.0.0.0:9080"
wire_listen: "0.0.0.0:19080"
edge_wire_listen: "0.0.0.0:19081"
`)
const wantEdgeWireListen = "127.0.0.1:29081"
t.Setenv("IOP_EDGE_WIRE_LISTEN", wantEdgeWireListen)
cfg, err := loadConfig(path)
if err != nil {
t.Fatalf("load config: %v", err)
}
if cfg.Server.EdgeWireListen != wantEdgeWireListen {
t.Fatalf("edge_wire_listen: got %q want %q", cfg.Server.EdgeWireListen, wantEdgeWireListen)
}
if cfg.Server.WireListen != "0.0.0.0:19080" {
t.Fatalf("wire_listen should stay from YAML: got %q", cfg.Server.WireListen)
}
}
func TestEdgeRegistryHTTPHandlersListAndGetEdge(t *testing.T) {
registry := wire.NewEdgeRegistry()
at := time.Unix(1780142200, 123).UTC()
registry.MarkConnected(&iop.EdgeHelloRequest{
EdgeId: "edge-b",
EdgeName: "Edge B",
Version: "1.2.3",
Capabilities: []string{"run", "node-registry"},
}, at)
registry.MarkConnected(&iop.EdgeHelloRequest{
EdgeId: "edge-a",
EdgeName: "Edge A",
Version: "1.0.0",
}, at.Add(time.Second))
mux := http.NewServeMux()
registerEdgeRegistryHandlers(mux, registry, nil)
listResp := httptest.NewRecorder()
mux.ServeHTTP(listResp, httptest.NewRequest(http.MethodGet, "/edges", nil))
if listResp.Code != http.StatusOK {
t.Fatalf("GET /edges status=%d body=%s", listResp.Code, listResp.Body.String())
}
var list edgeRegistryResponse
if err := json.Unmarshal(listResp.Body.Bytes(), &list); err != nil {
t.Fatalf("decode list response: %v", err)
}
if len(list.Edges) != 2 {
t.Fatalf("expected 2 edges, got %d", len(list.Edges))
}
if list.Edges[0].EdgeID != "edge-a" || list.Edges[1].EdgeID != "edge-b" {
t.Fatalf("expected stable edge_id ordering, got %+v", list.Edges)
}
if list.Edges[1].Protocol != wire.Protocol {
t.Fatalf("protocol: got %q want %q", list.Edges[1].Protocol, wire.Protocol)
}
if !list.Edges[1].Connected {
t.Fatal("expected edge-b to be connected")
}
getResp := httptest.NewRecorder()
mux.ServeHTTP(getResp, httptest.NewRequest(http.MethodGet, "/edges/edge-b", nil))
if getResp.Code != http.StatusOK {
t.Fatalf("GET /edges/edge-b status=%d body=%s", getResp.Code, getResp.Body.String())
}
var got edgeRegistryView
if err := json.Unmarshal(getResp.Body.Bytes(), &got); err != nil {
t.Fatalf("decode get response: %v", err)
}
if got.EdgeName != "Edge B" || got.Version != "1.2.3" {
t.Fatalf("unexpected edge view: %+v", got)
}
if len(got.Capabilities) != 2 || got.Capabilities[0] != "run" || got.Capabilities[1] != "node-registry" {
t.Fatalf("capabilities: got %+v", got.Capabilities)
}
if !got.LastSeen.Equal(at) {
t.Fatalf("last_seen: got %s want %s", got.LastSeen, at)
}
}
func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) {
registry := wire.NewEdgeRegistry()
connectedAt := time.Unix(1780142200, 0).UTC()
registry.MarkConnected(&iop.EdgeHelloRequest{
EdgeId: "edge-a",
EdgeName: "Edge A",
Version: "1.0.0",
}, connectedAt)
registry.MarkConnected(&iop.EdgeHelloRequest{
EdgeId: "edge-b",
EdgeName: "Edge B",
Version: "1.0.0",
}, connectedAt)
eventAt := time.Unix(1780142300, 0).UTC()
receivedAt := eventAt.Add(time.Second)
registry.RecordNodeEvent("edge-a", &iop.EdgeNodeEvent{
EventId: "evt-node-1",
Type: "node.connected",
Source: "edge",
NodeId: "node-1",
Alias: "alpha",
Reason: "registered",
Timestamp: eventAt.UnixNano(),
Metadata: map[string]string{"rack": "r1"},
}, receivedAt)
registry.RecordNodeEvent("edge-b", &iop.EdgeNodeEvent{
EventId: "evt-node-2",
Type: "node.connected",
NodeId: "node-2",
Timestamp: eventAt.UnixNano(),
}, receivedAt)
mux := http.NewServeMux()
registerEdgeRegistryHandlers(mux, registry, nil)
resp := httptest.NewRecorder()
mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events", nil))
if resp.Code != http.StatusOK {
t.Fatalf("GET /edges/edge-a/events status=%d body=%s", resp.Code, resp.Body.String())
}
var got edgeNodeEventsResponse
if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
t.Fatalf("decode events response: %v", err)
}
if got.EdgeID != "edge-a" {
t.Fatalf("edge_id: got %q want edge-a", got.EdgeID)
}
if len(got.Events) != 1 {
t.Fatalf("events len: got %d want 1", len(got.Events))
}
event := got.Events[0]
if event.EventID != "evt-node-1" || event.Type != "node.connected" {
t.Fatalf("unexpected event identity: %+v", event)
}
if event.NodeID != "node-1" || event.Alias != "alpha" || event.Reason != "registered" {
t.Fatalf("unexpected node event view: %+v", event)
}
if !event.Timestamp.Equal(eventAt) {
t.Fatalf("timestamp: got %s want %s", event.Timestamp, eventAt)
}
if !event.ReceivedAt.Equal(receivedAt) {
t.Fatalf("received_at: got %s want %s", event.ReceivedAt, receivedAt)
}
if event.Metadata["rack"] != "r1" {
t.Fatalf("metadata rack: got %q", event.Metadata["rack"])
}
filtered := httptest.NewRecorder()
mux.ServeHTTP(filtered, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events?node_id=missing", nil))
if filtered.Code != http.StatusOK {
t.Fatalf("filtered events status=%d body=%s", filtered.Code, filtered.Body.String())
}
var filteredGot edgeNodeEventsResponse
if err := json.Unmarshal(filtered.Body.Bytes(), &filteredGot); err != nil {
t.Fatalf("decode filtered events response: %v", err)
}
if len(filteredGot.Events) != 0 {
t.Fatalf("expected empty filtered events, got %+v", filteredGot.Events)
}
extraResp := httptest.NewRecorder()
mux.ServeHTTP(extraResp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events/extra", nil))
if extraResp.Code != http.StatusNotFound {
t.Fatalf("GET /edges/edge-a/events/extra status=%d body=%s", extraResp.Code, extraResp.Body.String())
}
missingResp := httptest.NewRecorder()
mux.ServeHTTP(missingResp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/events", nil))
if missingResp.Code != http.StatusNotFound {
t.Fatalf("GET /edges/missing-edge/events status=%d body=%s", missingResp.Code, missingResp.Body.String())
}
}
func TestEdgeRegistryHTTPHandlersRejectUnsupportedCases(t *testing.T) {
registry := wire.NewEdgeRegistry()
mux := http.NewServeMux()
registerEdgeRegistryHandlers(mux, registry, nil)
postResp := httptest.NewRecorder()
mux.ServeHTTP(postResp, httptest.NewRequest(http.MethodPost, "/edges", nil))
if postResp.Code != http.StatusMethodNotAllowed {
t.Fatalf("POST /edges status=%d body=%s", postResp.Code, postResp.Body.String())
}
missingResp := httptest.NewRecorder()
mux.ServeHTTP(missingResp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge", nil))
if missingResp.Code != http.StatusNotFound {
t.Fatalf("GET /edges/missing-edge status=%d body=%s", missingResp.Code, missingResp.Body.String())
}
}
func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) {
registry := wire.NewEdgeRegistry()
connectedAt := time.Unix(1780142200, 0).UTC()
registry.MarkConnected(&iop.EdgeHelloRequest{
EdgeId: "edge-a",
EdgeName: "Edge A",
Version: "1.0.0",
}, connectedAt)
t.Run("Success", func(t *testing.T) {
mux := http.NewServeMux()
// Mock config payload containing secret settings to verify secrecy
dummySettings, err := structpb.NewStruct(map[string]any{
"secret_key": "super-secret-token",
})
if err != nil {
t.Fatalf("create dummy settings: %v", err)
}
requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
if edgeID != "edge-a" {
return nil, fmt.Errorf("unexpected edge: %s", edgeID)
}
return &iop.EdgeStatusResponse{
RequestId: "req-123",
EdgeId: "edge-a",
EdgeName: "Edge A",
ObservedTimeUnixNano: connectedAt.UnixNano(),
Nodes: []*iop.EdgeNodeSnapshot{
{
NodeId: "node-1",
Alias: "alpha",
Label: "node0",
Connected: true,
Config: &iop.NodeConfigPayload{
Adapters: []*iop.AdapterConfig{
{
Type: "ollama",
Enabled: true,
Settings: dummySettings,
},
},
Runtime: &iop.NodeRuntimeConfig{
Concurrency: 4,
},
},
},
},
}, nil
}
registerEdgeRegistryHandlers(mux, registry, requestStatus)
resp := httptest.NewRecorder()
mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil))
if resp.Code != http.StatusOK {
t.Fatalf("GET /edges/edge-a/status status=%d body=%s", resp.Code, resp.Body.String())
}
var view edgeStatusResponseView
if err := json.Unmarshal(resp.Body.Bytes(), &view); err != nil {
t.Fatalf("decode status view: %v", err)
}
if view.EdgeID != "edge-a" || len(view.Nodes) != 1 {
t.Fatalf("unexpected view contents: %+v", view)
}
node := view.Nodes[0]
if node.NodeID != "node-1" || node.Alias != "alpha" || !node.Connected {
t.Fatalf("unexpected node state: %+v", node)
}
if node.Config == nil {
t.Fatal("expected config summary to be populated")
}
if node.Config.Concurrency != 4 || len(node.Config.Adapters) != 1 {
t.Fatalf("unexpected config summary: %+v", node.Config)
}
adapter := node.Config.Adapters[0]
if adapter.Type != "ollama" || !adapter.Enabled {
t.Fatalf("unexpected adapter summary: %+v", adapter)
}
// Ensure raw settings/secret_key is NOT leaked anywhere in the JSON body
bodyStr := resp.Body.String()
if strings.Contains(bodyStr, "super-secret-token") || strings.Contains(bodyStr, "secret_key") || strings.Contains(bodyStr, "secret_settings") {
t.Fatalf("leaked secret settings in response: %s", bodyStr)
}
})
t.Run("MissingEdge", func(t *testing.T) {
mux := http.NewServeMux()
requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
return nil, fmt.Errorf("should not be called")
}
registerEdgeRegistryHandlers(mux, registry, requestStatus)
resp := httptest.NewRecorder()
mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/status", nil))
if resp.Code != http.StatusNotFound {
t.Fatalf("expected 404 for missing edge, got status=%d", resp.Code)
}
})
t.Run("UpstreamErrorOrTimeout", func(t *testing.T) {
mux := http.NewServeMux()
requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
return nil, fmt.Errorf("upstream timeout")
}
registerEdgeRegistryHandlers(mux, registry, requestStatus)
resp := httptest.NewRecorder()
mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil))
if resp.Code != http.StatusBadGateway {
t.Fatalf("expected 502 for upstream error, got status=%d body=%s", resp.Code, resp.Body.String())
}
var errorResp map[string]string
if err := json.Unmarshal(resp.Body.Bytes(), &errorResp); err != nil {
t.Fatalf("decode error response: %v", err)
}
if errorResp["error"] != "upstream timeout" {
t.Fatalf("unexpected error response: %+v", errorResp)
}
})
}