96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from canonical_layer.risk import RiskService
|
|
from canonical_layer.store import CanonicalStore
|
|
from config.settings import OneCSettings
|
|
|
|
|
|
def _build_settings(db_url: str) -> OneCSettings:
|
|
return OneCSettings(
|
|
base_url="http://localhost",
|
|
infobase="buh_test",
|
|
username="",
|
|
password="",
|
|
odata_path="/odata/standard.odata/",
|
|
timeout=30,
|
|
verify_tls=False,
|
|
probe_top=5,
|
|
probe_entity_sets=(),
|
|
canonical_db_url=db_url,
|
|
refresh_default_limit_per_set=50,
|
|
refresh_default_entity_keywords=("document", "posting"),
|
|
feature_default_baseline_window_hours=24,
|
|
anomaly_stale_refresh_threshold_hours=6,
|
|
feature_entity_scan_limit=200000,
|
|
risk_medium_threshold=0.45,
|
|
risk_high_threshold=0.75,
|
|
risk_anomaly_scan_limit=5000,
|
|
)
|
|
|
|
|
|
def test_risk_engine_builds_patterns_from_feature_anomalies(tmp_path: Path) -> None:
|
|
db_url = f"sqlite:///{(tmp_path / 'risk_engine.db').as_posix()}"
|
|
settings = _build_settings(db_url)
|
|
store = CanonicalStore(settings.canonical_db_url)
|
|
store.ensure_created()
|
|
|
|
feature_run_id = store.start_feature_run(baseline_window_hours=24)
|
|
store.replace_feature_results(
|
|
run_id=feature_run_id,
|
|
metrics=[],
|
|
anomalies=[
|
|
{
|
|
"signal_type": "high_link_degree",
|
|
"severity": "high",
|
|
"scope": "DocumentSales",
|
|
"scope_id": "doc-001",
|
|
"score": 1.8,
|
|
"details": {"link_count": 50},
|
|
},
|
|
{
|
|
"signal_type": "entity_count_drift",
|
|
"severity": "medium",
|
|
"scope": "source_entity",
|
|
"scope_id": "DocumentSales",
|
|
"score": 0.4,
|
|
"details": {"drift_ratio": 0.4},
|
|
},
|
|
],
|
|
)
|
|
store.finish_feature_run(
|
|
run_id=feature_run_id,
|
|
status="success",
|
|
entities_total=10,
|
|
metrics_written=0,
|
|
anomalies_written=2,
|
|
details={},
|
|
)
|
|
|
|
service = RiskService(settings=settings, store=store)
|
|
result = service.run_risk_engine()
|
|
assert result.status == "success"
|
|
assert result.patterns_written >= 3
|
|
|
|
patterns = service.list_patterns(limit=100, active_only=True)
|
|
pattern_keys = {item["pattern_key"] for item in patterns}
|
|
assert "global_risk_summary" in pattern_keys
|
|
assert "suspicious_link_hub_risk" in pattern_keys
|
|
assert "structural_drift_risk" in pattern_keys
|
|
|
|
|
|
def test_risk_engine_handles_missing_feature_baseline(tmp_path: Path) -> None:
|
|
db_url = f"sqlite:///{(tmp_path / 'risk_engine_missing_feature.db').as_posix()}"
|
|
settings = _build_settings(db_url)
|
|
store = CanonicalStore(settings.canonical_db_url)
|
|
store.ensure_created()
|
|
service = RiskService(settings=settings, store=store)
|
|
|
|
result = service.run_risk_engine()
|
|
assert result.status == "success"
|
|
patterns = service.list_patterns(limit=100, active_only=True)
|
|
pattern_keys = {item["pattern_key"] for item in patterns}
|
|
assert "global_risk_summary" in pattern_keys
|
|
assert "operational_freshness_risk" in pattern_keys
|