Claudia
c5e5ce9dc0
Some checks failed
Tests / test (push) Failing after 5s
- Fixed bare 'from governance.' imports in source + tests - Fixed bare 'from intelligence.' imports in tests - Fixed mock.patch targets to use full cortex.xxx paths - All 405 tests passing
119 lines
4.6 KiB
Python
119 lines
4.6 KiB
Python
"""Tests for intelligence/loop.py — Darkplex Loop state machine and helpers."""
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
sys.path.insert(0, str(Path.home() / "repos" / "darkplex-core" / "intelligence"))
|
|
|
|
import cortex.intelligence.loop as darkplex_loop
|
|
|
|
|
|
class TestImportance:
|
|
def test_empty(self):
|
|
assert darkplex_loop._importance("") == 0.0
|
|
|
|
def test_heartbeat_low(self):
|
|
assert darkplex_loop._importance("HEARTBEAT_OK all systems nominal") < 0.2
|
|
|
|
def test_business_content_high(self):
|
|
score = darkplex_loop._importance("Meeting about the project deadline and budget milestone")
|
|
assert score > 0.4
|
|
|
|
def test_clamped(self):
|
|
for text in ["", "x" * 1000, "meeting project company contract decision strategy"]:
|
|
s = darkplex_loop._importance(text)
|
|
assert 0.0 <= s <= 1.0
|
|
|
|
|
|
class TestLoopState:
|
|
def test_init(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
assert state.status == "INIT"
|
|
assert state.cycle_count == 0
|
|
|
|
def test_save_and_load(self, tmp_path):
|
|
sf = tmp_path / "state.json"
|
|
with patch.object(darkplex_loop, 'STATE_FILE', sf):
|
|
state = darkplex_loop.LoopState()
|
|
state.status = "RUNNING"
|
|
state.cycle_count = 5
|
|
state.save()
|
|
|
|
state2 = darkplex_loop.LoopState()
|
|
assert state2.status == "RUNNING"
|
|
assert state2.cycle_count == 5
|
|
|
|
def test_record_success(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
state.record_success({"test": "ok"})
|
|
assert state.status == "RUNNING"
|
|
assert state.consecutive_failures == 0
|
|
assert state.cycle_count == 1
|
|
|
|
def test_record_failure_degraded(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
state.record_failure("ingest", "timeout")
|
|
assert state.status == "DEGRADED"
|
|
assert state.consecutive_failures == 1
|
|
|
|
def test_record_failure_emergency(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
for i in range(3):
|
|
state.record_failure("ingest", "timeout")
|
|
assert state.status == "EMERGENCY"
|
|
|
|
def test_can_alert(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
assert state.can_alert()
|
|
state.mark_alerted()
|
|
assert not state.can_alert()
|
|
|
|
def test_record_perf(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
state.record_perf({"total_ms": 1000, "ingest_ms": 200})
|
|
assert state.perf["total_ms"] == 1000
|
|
assert len(state.perf_history) == 1
|
|
|
|
def test_perf_averages(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
state.record_perf({"total_ms": 1000})
|
|
state.record_perf({"total_ms": 2000})
|
|
avgs = state.perf_averages()
|
|
assert avgs["total_ms"] == 1500
|
|
|
|
def test_perf_history_capped(self, tmp_path):
|
|
with patch.object(darkplex_loop, 'STATE_FILE', tmp_path / "state.json"):
|
|
state = darkplex_loop.LoopState()
|
|
for i in range(15):
|
|
state.record_perf({"total_ms": i * 100})
|
|
assert len(state.perf_history) == 10
|
|
|
|
|
|
class TestCheckNewEvents:
|
|
@patch("cortex.intelligence.loop.subprocess.run")
|
|
def test_returns_pending(self, mock_run):
|
|
mock_run.return_value = MagicMock(
|
|
returncode=0, stdout=json.dumps({"num_pending": 42})
|
|
)
|
|
assert darkplex_loop.check_new_events() == 42
|
|
|
|
@patch("cortex.intelligence.loop.subprocess.run")
|
|
def test_returns_negative_on_failure(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=1, stdout="")
|
|
assert darkplex_loop.check_new_events() == -1
|
|
|
|
@patch("cortex.intelligence.loop.subprocess.run")
|
|
def test_handles_exception(self, mock_run):
|
|
mock_run.side_effect = Exception("nats not found")
|
|
assert darkplex_loop.check_new_events() == -1
|