497 lines
14 KiB
Go
497 lines
14 KiB
Go
package api
|
|
|
|
import (
|
|
"bytes"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"reanimator/internal/ingest"
|
|
"reanimator/internal/repository"
|
|
)
|
|
|
|
func TestIngestHardwareIdempotent(t *testing.T) {
|
|
dsn := os.Getenv("DATABASE_DSN")
|
|
if dsn == "" {
|
|
t.Skip("DATABASE_DSN not set")
|
|
}
|
|
|
|
db, err := repository.Open(dsn)
|
|
if err != nil {
|
|
t.Fatalf("open db: %v", err)
|
|
}
|
|
defer db.Close()
|
|
|
|
if err := applyMigrations(db); err != nil {
|
|
t.Fatalf("apply migrations: %v", err)
|
|
}
|
|
if err := cleanupRegistry(db); err != nil {
|
|
t.Fatalf("cleanup: %v", err)
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
RegisterIngestRoutes(mux, IngestDependencies{Service: ingest.NewService(db)})
|
|
server := httptest.NewServer(mux)
|
|
defer server.Close()
|
|
|
|
payload := map[string]any{
|
|
"target_host": "hwserver-1",
|
|
"collected_at": time.Now().UTC().Format(time.RFC3339),
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": "BOARD123"},
|
|
"cpus": []map[string]any{{"socket": 0, "status": "OK"}},
|
|
},
|
|
}
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
t.Fatalf("marshal payload: %v", err)
|
|
}
|
|
|
|
resp, err := http.Post(server.URL+"/ingest/hardware", "application/json", bytes.NewReader(body))
|
|
if err != nil {
|
|
t.Fatalf("post: %v", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("expected 201, got %d", resp.StatusCode)
|
|
}
|
|
var createdResp struct {
|
|
Summary struct {
|
|
TimelineEventsCreated int `json:"timeline_events_created"`
|
|
} `json:"summary"`
|
|
}
|
|
createdBody, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
t.Fatalf("read body: %v", err)
|
|
}
|
|
if err := json.Unmarshal(createdBody, &createdResp); err != nil {
|
|
t.Fatalf("decode body: %v", err)
|
|
}
|
|
if createdResp.Summary.TimelineEventsCreated != 5 {
|
|
t.Fatalf("expected timeline_events_created=5, got %d", createdResp.Summary.TimelineEventsCreated)
|
|
}
|
|
|
|
resp, err = http.Post(server.URL+"/ingest/hardware", "application/json", bytes.NewReader(body))
|
|
if err != nil {
|
|
t.Fatalf("post duplicate: %v", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusOK {
|
|
t.Fatalf("expected 200 on duplicate, got %d", resp.StatusCode)
|
|
}
|
|
|
|
assertCount(t, db, "log_bundles", 1)
|
|
assertCount(t, db, "observations", 1)
|
|
assertCount(t, db, "installations", 1)
|
|
assertCount(t, db, "parts", 1)
|
|
}
|
|
|
|
func TestIngestHardwareStatusEvents(t *testing.T) {
|
|
dsn := os.Getenv("DATABASE_DSN")
|
|
if dsn == "" {
|
|
t.Skip("DATABASE_DSN not set")
|
|
}
|
|
|
|
db, err := repository.Open(dsn)
|
|
if err != nil {
|
|
t.Fatalf("open db: %v", err)
|
|
}
|
|
defer db.Close()
|
|
|
|
if err := applyMigrations(db); err != nil {
|
|
t.Fatalf("apply migrations: %v", err)
|
|
}
|
|
if err := cleanupRegistry(db); err != nil {
|
|
t.Fatalf("cleanup: %v", err)
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
RegisterIngestRoutes(mux, IngestDependencies{Service: ingest.NewService(db)})
|
|
server := httptest.NewServer(mux)
|
|
defer server.Close()
|
|
|
|
collectedAt := "2026-02-10T16:00:00Z"
|
|
warningAt := "2026-02-10T15:10:00Z"
|
|
failedAt := "2026-02-10T15:12:00Z"
|
|
|
|
payload := map[string]any{
|
|
"target_host": "status-server",
|
|
"collected_at": collectedAt,
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": "STATUS-123"},
|
|
"storage": []map[string]any{
|
|
{
|
|
"slot": "Storage-0",
|
|
"serial_number": "WARN-001",
|
|
"present": true,
|
|
"status": "Warning",
|
|
"status_changed_at": warningAt,
|
|
},
|
|
{
|
|
"slot": "Storage-1",
|
|
"serial_number": "CRIT-001",
|
|
"present": true,
|
|
"status": "Critical",
|
|
"status_changed_at": failedAt,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
t.Fatalf("marshal payload: %v", err)
|
|
}
|
|
|
|
resp, err := http.Post(server.URL+"/ingest/hardware", "application/json", bytes.NewReader(body))
|
|
if err != nil {
|
|
t.Fatalf("post: %v", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("expected 201, got %d", resp.StatusCode)
|
|
}
|
|
|
|
assertCountQuery(t, db, "SELECT COUNT(*) FROM failure_events", 1)
|
|
assertCountQuery(t, db, "SELECT COUNT(*) FROM timeline_events WHERE event_type = 'COMPONENT_WARNING'", 2)
|
|
assertCountQuery(t, db, "SELECT COUNT(*) FROM timeline_events WHERE event_type = 'COMPONENT_FAILED'", 2)
|
|
|
|
wantWarningAt, _ := time.Parse(time.RFC3339, warningAt)
|
|
wantFailedAt, _ := time.Parse(time.RFC3339, failedAt)
|
|
|
|
var gotWarningAt time.Time
|
|
if err := db.QueryRow(`SELECT event_time FROM timeline_events WHERE event_type = 'COMPONENT_WARNING' ORDER BY id LIMIT 1`).Scan(&gotWarningAt); err != nil {
|
|
t.Fatalf("query warning event_time: %v", err)
|
|
}
|
|
if !gotWarningAt.UTC().Equal(wantWarningAt.UTC()) {
|
|
t.Fatalf("warning event_time = %s, want %s", gotWarningAt.UTC().Format(time.RFC3339), wantWarningAt.UTC().Format(time.RFC3339))
|
|
}
|
|
|
|
var gotFailedAt time.Time
|
|
if err := db.QueryRow(`SELECT event_time FROM timeline_events WHERE event_type = 'COMPONENT_FAILED' ORDER BY id LIMIT 1`).Scan(&gotFailedAt); err != nil {
|
|
t.Fatalf("query failed event_time: %v", err)
|
|
}
|
|
if !gotFailedAt.UTC().Equal(wantFailedAt.UTC()) {
|
|
t.Fatalf("failed event_time = %s, want %s", gotFailedAt.UTC().Format(time.RFC3339), wantFailedAt.UTC().Format(time.RFC3339))
|
|
}
|
|
}
|
|
|
|
func TestIngestHardwareComponentRecoversToOK(t *testing.T) {
|
|
dsn := os.Getenv("DATABASE_DSN")
|
|
if dsn == "" {
|
|
t.Skip("DATABASE_DSN not set")
|
|
}
|
|
|
|
db, err := repository.Open(dsn)
|
|
if err != nil {
|
|
t.Fatalf("open db: %v", err)
|
|
}
|
|
defer db.Close()
|
|
|
|
if err := applyMigrations(db); err != nil {
|
|
t.Fatalf("apply migrations: %v", err)
|
|
}
|
|
if err := cleanupRegistry(db); err != nil {
|
|
t.Fatalf("cleanup: %v", err)
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
RegisterIngestRoutes(mux, IngestDependencies{Service: ingest.NewService(db)})
|
|
server := httptest.NewServer(mux)
|
|
defer server.Close()
|
|
|
|
postPayload := func(payload map[string]any) {
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
t.Fatalf("marshal payload: %v", err)
|
|
}
|
|
resp, err := http.Post(server.URL+"/ingest/hardware", "application/json", bytes.NewReader(body))
|
|
if err != nil {
|
|
t.Fatalf("post: %v", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("expected 201, got %d", resp.StatusCode)
|
|
}
|
|
}
|
|
|
|
failedAt := "2026-02-10T15:10:00Z"
|
|
okAt := "2026-02-10T15:22:00Z"
|
|
|
|
postPayload(map[string]any{
|
|
"target_host": "recover-server",
|
|
"collected_at": "2026-02-10T15:10:10Z",
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": "RECOVER-001"},
|
|
"storage": []map[string]any{
|
|
{
|
|
"slot": "Disk.Bay.0",
|
|
"serial_number": "RECOVER-DISK-001",
|
|
"present": true,
|
|
"status": "Critical",
|
|
"status_changed_at": failedAt,
|
|
},
|
|
},
|
|
},
|
|
})
|
|
|
|
postPayload(map[string]any{
|
|
"target_host": "recover-server",
|
|
"collected_at": "2026-02-10T15:22:10Z",
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": "RECOVER-001"},
|
|
"storage": []map[string]any{
|
|
{
|
|
"slot": "Disk.Bay.0",
|
|
"serial_number": "RECOVER-DISK-001",
|
|
"present": true,
|
|
"status": "OK",
|
|
"status_changed_at": okAt,
|
|
},
|
|
},
|
|
},
|
|
})
|
|
|
|
assertCountQuery(t, db, "SELECT COUNT(*) FROM timeline_events WHERE event_type = 'COMPONENT_FAILED'", 2)
|
|
assertCountQuery(t, db, "SELECT COUNT(*) FROM timeline_events WHERE event_type = 'COMPONENT_OK'", 2)
|
|
|
|
var latestEventType string
|
|
if err := db.QueryRow(`
|
|
SELECT event_type
|
|
FROM timeline_events
|
|
WHERE subject_type = 'component'
|
|
AND subject_id = (SELECT id FROM parts WHERE vendor_serial = 'RECOVER-DISK-001' LIMIT 1)
|
|
ORDER BY event_time DESC, id DESC
|
|
LIMIT 1
|
|
`).Scan(&latestEventType); err != nil {
|
|
t.Fatalf("latest component status query: %v", err)
|
|
}
|
|
if latestEventType != "COMPONENT_OK" {
|
|
t.Fatalf("expected latest component event COMPONENT_OK, got %q", latestEventType)
|
|
}
|
|
}
|
|
|
|
func TestIngestHardwareFirmwareChange(t *testing.T) {
|
|
dsn := os.Getenv("DATABASE_DSN")
|
|
if dsn == "" {
|
|
t.Skip("DATABASE_DSN not set")
|
|
}
|
|
|
|
db, err := repository.Open(dsn)
|
|
if err != nil {
|
|
t.Fatalf("open db: %v", err)
|
|
}
|
|
defer db.Close()
|
|
|
|
if err := applyMigrations(db); err != nil {
|
|
t.Fatalf("apply migrations: %v", err)
|
|
}
|
|
if err := cleanupRegistry(db); err != nil {
|
|
t.Fatalf("cleanup: %v", err)
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
RegisterIngestRoutes(mux, IngestDependencies{Service: ingest.NewService(db)})
|
|
server := httptest.NewServer(mux)
|
|
defer server.Close()
|
|
|
|
boardSerial := "FIRMWARE-123"
|
|
firstPayload := map[string]any{
|
|
"target_host": "firmware-server",
|
|
"collected_at": time.Now().UTC().Format(time.RFC3339),
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": boardSerial},
|
|
"firmware": []map[string]any{
|
|
{"device_name": "BIOS", "version": "1.0.0"},
|
|
},
|
|
},
|
|
}
|
|
|
|
secondPayload := map[string]any{
|
|
"target_host": "firmware-server",
|
|
"collected_at": time.Now().UTC().Add(5 * time.Minute).Format(time.RFC3339),
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": boardSerial},
|
|
"firmware": []map[string]any{
|
|
{"device_name": "BIOS", "version": "1.1.0"},
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, payload := range []map[string]any{firstPayload, secondPayload} {
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
t.Fatalf("marshal payload: %v", err)
|
|
}
|
|
resp, err := http.Post(server.URL+"/ingest/hardware", "application/json", bytes.NewReader(body))
|
|
if err != nil {
|
|
t.Fatalf("post: %v", err)
|
|
}
|
|
resp.Body.Close()
|
|
if resp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("expected 201, got %d", resp.StatusCode)
|
|
}
|
|
}
|
|
|
|
var version string
|
|
row := db.QueryRow(`
|
|
SELECT firmware_version
|
|
FROM machine_firmware_states
|
|
WHERE machine_id = (SELECT id FROM machines WHERE vendor_serial = ? LIMIT 1) AND device_name = ?
|
|
`, boardSerial, "BIOS")
|
|
if err := row.Scan(&version); err != nil {
|
|
t.Fatalf("firmware state query: %v", err)
|
|
}
|
|
if version != "1.1.0" {
|
|
t.Fatalf("expected firmware_version 1.1.0, got %s", version)
|
|
}
|
|
|
|
assertCountQuery(t, db, "SELECT COUNT(*) FROM timeline_events WHERE subject_type = 'asset' AND event_type = 'FIRMWARE_CHANGED'", 2)
|
|
}
|
|
|
|
func TestIngestHardwareCreatesMachineInStock(t *testing.T) {
|
|
dsn := os.Getenv("DATABASE_DSN")
|
|
if dsn == "" {
|
|
t.Skip("DATABASE_DSN not set")
|
|
}
|
|
|
|
db, err := repository.Open(dsn)
|
|
if err != nil {
|
|
t.Fatalf("open db: %v", err)
|
|
}
|
|
defer db.Close()
|
|
|
|
if err := applyMigrations(db); err != nil {
|
|
t.Fatalf("apply migrations: %v", err)
|
|
}
|
|
if err := cleanupRegistry(db); err != nil {
|
|
t.Fatalf("cleanup: %v", err)
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
RegisterIngestRoutes(mux, IngestDependencies{Service: ingest.NewService(db)})
|
|
server := httptest.NewServer(mux)
|
|
defer server.Close()
|
|
|
|
payload := map[string]any{
|
|
"target_host": "stock-server",
|
|
"collected_at": time.Now().UTC().Format(time.RFC3339),
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": "STOCK-001"},
|
|
},
|
|
}
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
t.Fatalf("marshal payload: %v", err)
|
|
}
|
|
|
|
resp, err := http.Post(server.URL+"/ingest/hardware", "application/json", bytes.NewReader(body))
|
|
if err != nil {
|
|
t.Fatalf("post: %v", err)
|
|
}
|
|
resp.Body.Close()
|
|
if resp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("expected 201, got %d", resp.StatusCode)
|
|
}
|
|
|
|
var projectID sql.NullString
|
|
if err := db.QueryRow(`SELECT project_id FROM machines WHERE vendor_serial = ?`, "STOCK-001").
|
|
Scan(&projectID); err != nil {
|
|
t.Fatalf("query machine ownership: %v", err)
|
|
}
|
|
if projectID.Valid {
|
|
t.Fatalf("expected stock machine with null project, got project=%v", projectID.String)
|
|
}
|
|
}
|
|
|
|
func TestIngestHardwareStoresStatusHistoryInObservationDetails(t *testing.T) {
|
|
dsn := os.Getenv("DATABASE_DSN")
|
|
if dsn == "" {
|
|
t.Skip("DATABASE_DSN not set")
|
|
}
|
|
|
|
db, err := repository.Open(dsn)
|
|
if err != nil {
|
|
t.Fatalf("open db: %v", err)
|
|
}
|
|
defer db.Close()
|
|
|
|
if err := applyMigrations(db); err != nil {
|
|
t.Fatalf("apply migrations: %v", err)
|
|
}
|
|
if err := cleanupRegistry(db); err != nil {
|
|
t.Fatalf("cleanup: %v", err)
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
RegisterIngestRoutes(mux, IngestDependencies{Service: ingest.NewService(db)})
|
|
server := httptest.NewServer(mux)
|
|
defer server.Close()
|
|
|
|
payload := map[string]any{
|
|
"target_host": "history-server",
|
|
"collected_at": "2026-02-10T15:30:00Z",
|
|
"hardware": map[string]any{
|
|
"board": map[string]any{"serial_number": "HISTORY-001"},
|
|
"storage": []map[string]any{
|
|
{
|
|
"slot": "Disk.Bay.0",
|
|
"serial_number": "HISTORY-DISK-001",
|
|
"present": true,
|
|
"status": "OK",
|
|
"status_changed_at": "2026-02-10T15:22:00Z",
|
|
"status_history": []map[string]any{
|
|
{
|
|
"status": "Critical",
|
|
"changed_at": "2026-02-10T15:10:00Z",
|
|
},
|
|
{
|
|
"status": "OK",
|
|
"changed_at": "2026-02-10T15:22:00Z",
|
|
"details": "Recovered after controller reset",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
t.Fatalf("marshal payload: %v", err)
|
|
}
|
|
|
|
resp, err := http.Post(server.URL+"/ingest/hardware", "application/json", bytes.NewReader(body))
|
|
if err != nil {
|
|
t.Fatalf("post: %v", err)
|
|
}
|
|
resp.Body.Close()
|
|
if resp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("expected 201, got %d", resp.StatusCode)
|
|
}
|
|
|
|
var historyStatus, statusChangedAt string
|
|
row := db.QueryRow(`
|
|
SELECT
|
|
JSON_UNQUOTE(JSON_EXTRACT(details, '$.status_history[0].status')),
|
|
JSON_UNQUOTE(JSON_EXTRACT(details, '$.status_changed_at'))
|
|
FROM observations
|
|
ORDER BY observed_at DESC, id DESC
|
|
LIMIT 1
|
|
`)
|
|
if err := row.Scan(&historyStatus, &statusChangedAt); err != nil {
|
|
t.Fatalf("details query: %v", err)
|
|
}
|
|
if historyStatus != "CRITICAL" {
|
|
t.Fatalf("expected first status_history status CRITICAL, got %q", historyStatus)
|
|
}
|
|
if statusChangedAt != "2026-02-10T15:22:00Z" {
|
|
t.Fatalf("expected status_changed_at=2026-02-10T15:22:00Z, got %q", statusChangedAt)
|
|
}
|
|
|
|
assertCountQuery(t, db, "SELECT COUNT(*) FROM timeline_events WHERE event_type = 'COMPONENT_OK'", 2)
|
|
}
|