260 lines
10 KiB
Python
260 lines
10 KiB
Python
|
"""
|
||
|
|
Tests for cognition launcher, news sources, and monitor.
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import tempfile
|
||
|
|
import time
|
||
|
|
import pytest
|
||
|
|
from malkhut.cognition_launcher import CognitionLauncher, CognitionConfig
|
||
|
|
from malkhut.training.news_sources import NewsSourceRepository, SourceMetadata
|
||
|
|
from malkhut.training.monitor import CognitionMonitor, CognitionMetrics
|
||
|
|
|
||
|
|
|
||
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
|
# COGNITION LAUNCHER (8 tests)
|
||
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
|
|
||
|
|
class TestCognitionLauncher:
|
||
|
|
def test_launcher_creates(self):
|
||
|
|
launcher = CognitionLauncher()
|
||
|
|
assert launcher._running is False
|
||
|
|
assert launcher._total_fetched == 0
|
||
|
|
|
||
|
|
def test_launcher_config(self):
|
||
|
|
cfg = CognitionConfig(rate_limit_rpm=10, fetch_interval_s=5)
|
||
|
|
launcher = CognitionLauncher(config=cfg)
|
||
|
|
assert launcher.config.rate_limit_rpm == 10
|
||
|
|
|
||
|
|
def test_launcher_loads_regimes(self):
|
||
|
|
path = '/tmp/test_regimes_e2e.json'
|
||
|
|
with open(path, 'w') as f:
|
||
|
|
json.dump({"flash_crash": {"source": "test"}}, f)
|
||
|
|
try:
|
||
|
|
launcher = CognitionLauncher(CognitionConfig(regime_db_path=path))
|
||
|
|
assert launcher._total_regimes == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_launcher_saves_regimes(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
launcher = CognitionLauncher(CognitionConfig(regime_db_path=path))
|
||
|
|
launcher._discovered_regimes["test"] = {"source": "test"}
|
||
|
|
launcher._save_regimes()
|
||
|
|
with open(path) as f:
|
||
|
|
data = json.load(f)
|
||
|
|
assert "test" in data
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
|
||
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
|
# NEWS SOURCE REPOSITORY (15 tests)
|
||
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
|
|
||
|
|
class TestNewsSourceRepository:
|
||
|
|
def test_register_source(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo = NewsSourceRepository(path)
|
||
|
|
repo.register("s1", "Test", "https://example.com", "news", 0.8, 0.9, 1.0)
|
||
|
|
assert repo.source_count == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_rank_by_relevance(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo = NewsSourceRepository(path)
|
||
|
|
repo.register("s1", "Low", "https://a.com", "news", 0.3, 0.8, 1.0)
|
||
|
|
repo.register("s2", "High", "https://b.com", "news", 0.9, 0.9, 1.0)
|
||
|
|
ranked = repo.rank_by_relevance()
|
||
|
|
assert ranked[0].source_id == "s2"
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_rank_by_health(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo = NewsSourceRepository(path)
|
||
|
|
repo.register("s1", "Good", "https://a.com", reliability=0.95)
|
||
|
|
repo.register("s2", "Bad", "https://b.com", reliability=0.5)
|
||
|
|
ranked = repo.rank_by_health()
|
||
|
|
assert ranked[0].source_id == "s1"
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_get_by_type(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo = NewsSourceRepository(path)
|
||
|
|
repo.register("s1", "News", "https://news.com", "news")
|
||
|
|
repo.register("s2", "Data", "https://data.com", "data")
|
||
|
|
assert len(repo.get_by_type("news")) == 1
|
||
|
|
assert len(repo.get_by_type("data")) == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_get_by_tag(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo = NewsSourceRepository(path)
|
||
|
|
repo.register("s1", "Test", "https://example.com", tags=("crypto", "news"))
|
||
|
|
assert len(repo.get_by_tag("crypto")) == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_disable_enable(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo = NewsSourceRepository(path)
|
||
|
|
repo.register("s1", "Test", "https://example.com")
|
||
|
|
assert repo.enabled_count == 1
|
||
|
|
repo.disable("s1")
|
||
|
|
assert repo.enabled_count == 0
|
||
|
|
repo.enable("s1")
|
||
|
|
assert repo.enabled_count == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_persistence(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo1 = NewsSourceRepository(path)
|
||
|
|
repo1.register("s1", "Test", "https://example.com")
|
||
|
|
repo1.record_fetch("s1", success=True)
|
||
|
|
|
||
|
|
repo2 = NewsSourceRepository(path)
|
||
|
|
assert repo2.source_count == 1
|
||
|
|
assert repo2.get_enabled()[0].fetch_count == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_seed_defaults(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
repo = NewsSourceRepository(path)
|
||
|
|
repo.seed_defaults()
|
||
|
|
assert repo.source_count >= 10
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_health_score(self):
|
||
|
|
s = SourceMetadata(source_id="s1", name="Test", url="https://x.com",
|
||
|
|
source_type="news", regime_relevance=0.5, reliability=0.9,
|
||
|
|
freshness_hours=1.0, fetch_count=10, error_count=1)
|
||
|
|
assert s.health_score > 0
|
||
|
|
assert s.health_score < 1.0
|
||
|
|
|
||
|
|
|
||
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
|
# COGNITION MONITOR (10 tests)
|
||
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
|
|
||
|
|
class TestCognitionMonitor:
|
||
|
|
def test_record_fetch(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
mon.record_fetch("s1", success=True)
|
||
|
|
assert mon.total_fetched == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_record_error(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
mon.record_fetch("s1", success=False)
|
||
|
|
assert mon.total_errors == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_record_regime(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
mon.record_regime(3)
|
||
|
|
assert mon.discovered_regimes == 3
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_snapshot(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
mon.record_fetch("s1", success=True)
|
||
|
|
metrics = mon.snapshot(total_sources=5, enabled_sources=4)
|
||
|
|
assert isinstance(metrics, CognitionMetrics)
|
||
|
|
assert metrics.total_fetched == 1
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_metrics_logged(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
mon.snapshot(total_sources=5, enabled_sources=4)
|
||
|
|
with open(path) as f:
|
||
|
|
lines = f.readlines()
|
||
|
|
assert len(lines) == 1
|
||
|
|
record = json.loads(lines[0])
|
||
|
|
assert "sources" in record
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_check_alerts_high_error(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
for _ in range(20):
|
||
|
|
mon.record_fetch("s1", success=False)
|
||
|
|
mon.record_fetch("s1", success=True)
|
||
|
|
mon.snapshot(total_sources=1, enabled_sources=1)
|
||
|
|
alerts = mon.check_alerts()
|
||
|
|
assert len(alerts) > 0
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_check_alerts_low_fetch_rate(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
mon._start_time = time.time() - 600 # 10 min ago
|
||
|
|
mon.record_fetch("s1", success=True) # only 1 fetch in 10 min
|
||
|
|
mon.snapshot(total_sources=1, enabled_sources=1)
|
||
|
|
alerts = mon.check_alerts()
|
||
|
|
assert any("LOW_FETCH_RATE" in a for a in alerts)
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|
||
|
|
|
||
|
|
def test_check_alerts_clean(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as f:
|
||
|
|
path = f.name
|
||
|
|
try:
|
||
|
|
mon = CognitionMonitor(path)
|
||
|
|
for _ in range(100):
|
||
|
|
mon.record_fetch("s1", success=True)
|
||
|
|
mon.snapshot(total_sources=1, enabled_sources=1)
|
||
|
|
alerts = mon.check_alerts()
|
||
|
|
assert len(alerts) == 0
|
||
|
|
finally:
|
||
|
|
os.unlink(path)
|