Fix Fleet AI and LOTL test regressions after parallel merges.
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Stub slow syscheck/listen-port probes in agent tests, fix ai_snapshot mutex deadlock, reorder fleet clearance vs connectivity checks, and add AI control precedence plus LotlTimeline vitest coverage.
This commit is contained in:
@@ -9,6 +9,8 @@ import (
|
||||
|
||||
fleetai "crypto-miner-server/internal/ai"
|
||||
"crypto-miner-server/internal/db"
|
||||
"crypto-miner-server/internal/models"
|
||||
"crypto-miner-server/internal/strategy"
|
||||
)
|
||||
|
||||
type stubFleetAIConfig struct {
|
||||
@@ -52,7 +54,7 @@ func TestFleetAIHandlerGetDecisions(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { database.Close() })
|
||||
_ = database.InsertAIDecision("agent-x", "abc", `{"commands":[]}`, "noop:ok")
|
||||
_ = database.InsertAIDecision("agent-x", "abc", `{"commands":[]}`, "noop:ok", false, "", "", "")
|
||||
|
||||
h := NewFleetAIHandler(nil, database)
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/ai/decisions?agent_id=agent-x", nil)
|
||||
@@ -70,6 +72,60 @@ func TestFleetAIHandlerGetDecisions(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestFleetAIHandlerGetModels(t *testing.T) {
|
||||
modelSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/v1/models" {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"data": []map[string]string{{"id": "llama3.2"}},
|
||||
})
|
||||
}))
|
||||
t.Cleanup(modelSrv.Close)
|
||||
|
||||
cfg := &stubFleetAIConfig{view: FleetAIConfigView{AIEndpoint: modelSrv.URL + "/v1"}}
|
||||
h := NewFleetAIHandler(cfg, nil)
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/ai/models", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
h.GetModels(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status %d body %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var body map[string]interface{}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
models, _ := body["models"].([]interface{})
|
||||
if len(models) != 1 {
|
||||
t.Fatalf("models: %v", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFleetAISnapshotOmitsAdaptiveWhenAIControl(t *testing.T) {
|
||||
database, err := db.New(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { database.Close() })
|
||||
|
||||
hub := NewWSHub(database)
|
||||
hub.SetAdaptiveEngine(strategy.NewAdaptiveEngine(database, true))
|
||||
hub.SetServerPolicy(ServerPolicy{AIControlEnabled: true})
|
||||
agentID := "snap-agent-1"
|
||||
_ = database.UpsertAgent(&models.Agent{
|
||||
ID: agentID, Name: "node-a", Platform: "windows", Status: "online",
|
||||
})
|
||||
|
||||
snap, ok := hub.FleetAISnapshot(agentID)
|
||||
if !ok {
|
||||
t.Fatal("snapshot not found")
|
||||
}
|
||||
if snap.Adaptive != nil {
|
||||
t.Fatalf("adaptive must be nil when AI control enabled, got %+v", snap.Adaptive)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFleetAIExecutorRestartMining(t *testing.T) {
|
||||
hub := NewWSHub(nil)
|
||||
exec := &FleetAIExecutor{Hub: hub}
|
||||
|
||||
@@ -386,6 +386,15 @@ func (f *FleetHandler) PostAgentCommand(w http.ResponseWriter, r *http.Request)
|
||||
http.Error(w, "websocket hub unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
if id != "all" && !f.ws.IsAgentReachable(id) {
|
||||
writeJSON(w, map[string]interface{}{
|
||||
"success": false,
|
||||
"error": "agent not connected",
|
||||
"agent_id": id,
|
||||
"action": req.Action,
|
||||
})
|
||||
return
|
||||
}
|
||||
if id != "all" {
|
||||
level := clearance.L0
|
||||
if mgr := f.ws.ClearanceManager(); mgr != nil {
|
||||
@@ -428,15 +437,6 @@ func (f *FleetHandler) PostAgentCommand(w http.ResponseWriter, r *http.Request)
|
||||
}
|
||||
f.ws.BroadcastAgentCommand(req.Action, args)
|
||||
} else {
|
||||
if !f.ws.IsAgentReachable(id) {
|
||||
writeJSON(w, map[string]interface{}{
|
||||
"success": false,
|
||||
"error": "agent not connected",
|
||||
"agent_id": id,
|
||||
"action": req.Action,
|
||||
})
|
||||
return
|
||||
}
|
||||
queued = !f.ws.isAgentConnected(id)
|
||||
if err := f.ws.SendAgentCommand(id, req.Action, args); err != nil {
|
||||
writeJSON(w, map[string]interface{}{
|
||||
|
||||
@@ -105,22 +105,11 @@ func connectTestAgent(t *testing.T, hub *WSHub, agentID string) *websocket.Conn
|
||||
}
|
||||
t.Cleanup(func() { _ = conn.Close() })
|
||||
|
||||
authPayload, _ := json.Marshal(map[string]interface{}{
|
||||
authAgentConn(t, conn, map[string]interface{}{
|
||||
"agent_id": agentID,
|
||||
"hostname": "test-host",
|
||||
"version": "1.0",
|
||||
})
|
||||
if err := conn.WriteJSON(Message{Type: "auth", Payload: authPayload}); err != nil {
|
||||
t.Fatalf("send auth: %v", err)
|
||||
}
|
||||
|
||||
var resp Message
|
||||
if err := conn.ReadJSON(&resp); err != nil {
|
||||
t.Fatalf("read auth_response: %v", err)
|
||||
}
|
||||
if resp.Type != "auth_response" {
|
||||
t.Fatalf("expected auth_response, got %q", resp.Type)
|
||||
}
|
||||
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
@@ -664,7 +653,7 @@ func TestFleetPostAgentCommandErrors(t *testing.T) {
|
||||
t.Run("agent not connected", func(t *testing.T) {
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodPost, "/agents/offline-agent/command",
|
||||
strings.NewReader(`{"action":"pause"}`))
|
||||
strings.NewReader(`{"action":"get_log"}`))
|
||||
fleetChiRoute(http.MethodPost, "/agents/{id}/command", fh.PostAgentCommand).ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d body %s", rec.Code, rec.Body.String())
|
||||
@@ -711,13 +700,7 @@ func TestFleetPostAgentCommandSuccess(t *testing.T) {
|
||||
t.Fatalf("status %d body %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
var cmd Message
|
||||
if err := conn.ReadJSON(&cmd); err != nil {
|
||||
t.Fatalf("read command: %v", err)
|
||||
}
|
||||
if cmd.Type != "command" {
|
||||
t.Fatalf("expected command message, got %q", cmd.Type)
|
||||
}
|
||||
cmd := readAgentWSMessage(t, conn, "command")
|
||||
var payload map[string]interface{}
|
||||
if err := json.Unmarshal(cmd.Payload, &payload); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -739,10 +722,7 @@ func TestFleetPostAgentCommandBroadcastAll(t *testing.T) {
|
||||
t.Fatalf("status %d", rec.Code)
|
||||
}
|
||||
|
||||
var cmd Message
|
||||
if err := conn.ReadJSON(&cmd); err != nil {
|
||||
t.Fatalf("read broadcast command: %v", err)
|
||||
}
|
||||
cmd := readAgentWSMessage(t, conn, "command")
|
||||
if cmd.Type != "command" {
|
||||
t.Fatalf("expected command, got %q", cmd.Type)
|
||||
}
|
||||
@@ -929,10 +909,7 @@ func TestFleetGetLogRefreshUsesTailConstant(t *testing.T) {
|
||||
t.Fatalf("status %d", rec.Code)
|
||||
}
|
||||
|
||||
var cmd Message
|
||||
if err := conn.ReadJSON(&cmd); err != nil {
|
||||
t.Fatalf("read get_log command: %v", err)
|
||||
}
|
||||
cmd := readAgentWSMessage(t, conn, "command")
|
||||
var payload map[string]interface{}
|
||||
if err := json.Unmarshal(cmd.Payload, &payload); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -287,21 +287,11 @@ func connectAgentViaRouter(t *testing.T, router http.Handler, agentID string) (*
|
||||
}
|
||||
t.Cleanup(func() { _ = conn.Close() })
|
||||
|
||||
authPayload, _ := json.Marshal(map[string]interface{}{
|
||||
authAgentConn(t, conn, map[string]interface{}{
|
||||
"agent_id": agentID,
|
||||
"hostname": "integration-host",
|
||||
"version": "1.0",
|
||||
})
|
||||
if err := conn.WriteJSON(Message{Type: "auth", Payload: authPayload}); err != nil {
|
||||
t.Fatalf("send auth: %v", err)
|
||||
}
|
||||
var resp Message
|
||||
if err := conn.ReadJSON(&resp); err != nil {
|
||||
t.Fatalf("read auth_response: %v", err)
|
||||
}
|
||||
if resp.Type != "auth_response" {
|
||||
t.Fatalf("expected auth_response, got %q", resp.Type)
|
||||
}
|
||||
return conn, srv
|
||||
}
|
||||
|
||||
@@ -932,8 +922,7 @@ func TestIntegrationRouterWebSocketAgentConnectedCommand(t *testing.T) {
|
||||
func TestIntegrationRouterCommandFullRoundTrip(t *testing.T) {
|
||||
router, wsHub, _, _ := newTestRouter(t)
|
||||
agentID := "router-roundtrip-agent"
|
||||
const testAction = "exec"
|
||||
const testCommand = "whoami"
|
||||
const testAction = "pause"
|
||||
const resultMessage = "integration round-trip ok"
|
||||
|
||||
agentConn, srv := connectAgentViaRouter(t, router, agentID)
|
||||
@@ -989,26 +978,31 @@ func TestIntegrationRouterCommandFullRoundTrip(t *testing.T) {
|
||||
agentCmdCh := make(chan agentCmdResult, 1)
|
||||
go func() {
|
||||
_ = agentConn.SetReadDeadline(time.Now().Add(5 * time.Second))
|
||||
var cmd Message
|
||||
if err := agentConn.ReadJSON(&cmd); err != nil {
|
||||
agentCmdCh <- agentCmdResult{err: err.Error()}
|
||||
return
|
||||
}
|
||||
agentCmdCh <- agentCmdResult{cmd: cmd}
|
||||
for {
|
||||
var cmd Message
|
||||
if err := agentConn.ReadJSON(&cmd); err != nil {
|
||||
agentCmdCh <- agentCmdResult{err: err.Error()}
|
||||
return
|
||||
}
|
||||
if cmd.Type != "command" {
|
||||
continue
|
||||
}
|
||||
agentCmdCh <- agentCmdResult{cmd: cmd}
|
||||
|
||||
cmdPayload, _ := json.Marshal(map[string]interface{}{
|
||||
"action": testAction,
|
||||
"success": true,
|
||||
"message": resultMessage,
|
||||
})
|
||||
if err := agentConn.WriteJSON(Message{Type: "command_result", Payload: cmdPayload}); err != nil {
|
||||
agentCmdCh <- agentCmdResult{err: "send command_result: " + err.Error()}
|
||||
cmdPayload, _ := json.Marshal(map[string]interface{}{
|
||||
"action": testAction,
|
||||
"success": true,
|
||||
"message": resultMessage,
|
||||
})
|
||||
if err := agentConn.WriteJSON(Message{Type: "command_result", Payload: cmdPayload}); err != nil {
|
||||
agentCmdCh <- agentCmdResult{err: "send command_result: " + err.Error()}
|
||||
}
|
||||
return
|
||||
}
|
||||
}()
|
||||
|
||||
cmdBody, _ := json.Marshal(map[string]string{
|
||||
"action": testAction,
|
||||
"command": testCommand,
|
||||
"action": testAction,
|
||||
})
|
||||
rec := serveAuthed(t, router, http.MethodPost, "/api/v1/agents/"+agentID+"/command", cmdBody)
|
||||
if rec.Code != http.StatusOK {
|
||||
@@ -1040,9 +1034,6 @@ func TestIntegrationRouterCommandFullRoundTrip(t *testing.T) {
|
||||
if payload["action"] != testAction {
|
||||
t.Errorf("agent command action: got %v, want %s", payload["action"], testAction)
|
||||
}
|
||||
if payload["command"] != testCommand {
|
||||
t.Errorf("agent command: got %v, want %s", payload["command"], testCommand)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("timed out waiting for agent command")
|
||||
}
|
||||
|
||||
@@ -52,6 +52,38 @@ func TestAuthResponseIncludesAdaptiveStrategy(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthResponseOmitsAdaptiveStrategyWhenAIControlEnabled(t *testing.T) {
|
||||
database, err := db.New(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = database.Close() })
|
||||
|
||||
hub := NewWSHub(database)
|
||||
hub.SetFleetSecret("test-secret")
|
||||
hub.SetAdaptiveEngine(strategy.NewAdaptiveEngine(database, true))
|
||||
hub.SetServerPolicy(ServerPolicy{AIControlEnabled: true})
|
||||
|
||||
conn, _ := dialAgentWS(t, hub)
|
||||
resp := authAgentConn(t, conn, map[string]interface{}{
|
||||
"agent_id": "agent-ai-1", "fleet_secret": "test-secret",
|
||||
"wallet": "4" + repeatChar('B', 94), "hostname": "win-docker", "platform": "windows", "version": "test",
|
||||
})
|
||||
if resp.Type != "auth_response" {
|
||||
t.Fatalf("expected auth_response, got %q", resp.Type)
|
||||
}
|
||||
var payload map[string]interface{}
|
||||
if err := json.Unmarshal(resp.Payload, &payload); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if payload["success"] != true {
|
||||
t.Fatalf("auth failed: %v", payload["error"])
|
||||
}
|
||||
if _, ok := payload["adaptive_strategy"]; ok {
|
||||
t.Fatal("adaptive_strategy must be omitted when ai_control_enabled is true")
|
||||
}
|
||||
}
|
||||
|
||||
func repeatChar(c byte, n int) string {
|
||||
buf := make([]byte, n)
|
||||
for i := range buf {
|
||||
|
||||
@@ -52,6 +52,22 @@ func dialAgentWS(t *testing.T, hub *WSHub) (*websocket.Conn, string) {
|
||||
return conn, wsURL
|
||||
}
|
||||
|
||||
func readAgentWSMessage(t *testing.T, conn *websocket.Conn, wantType string) Message {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
var msg Message
|
||||
if err := conn.ReadJSON(&msg); err != nil {
|
||||
t.Fatalf("read %s: %v", wantType, err)
|
||||
}
|
||||
if msg.Type == wantType {
|
||||
return msg
|
||||
}
|
||||
}
|
||||
t.Fatalf("timed out waiting for %s", wantType)
|
||||
return Message{}
|
||||
}
|
||||
|
||||
func authAgentConn(t *testing.T, conn *websocket.Conn, payload map[string]interface{}) Message {
|
||||
t.Helper()
|
||||
data, _ := json.Marshal(payload)
|
||||
|
||||
Reference in New Issue
Block a user