Claudia
fd7d75c0ed
Some checks failed
Tests / test (push) Failing after 2s
- Merged all unique darkplex-core modules into cortex: - intelligence/ subfolder (anticipator, collective, shared_memory, knowledge_cleanup, temporal, llm_extractor, loop) - governance/ subfolder (policy engine, risk scorer, evidence, enforcer, report generator) - entity_manager.py, knowledge_extractor.py - Fixed bare 'from intelligence.' imports to 'from cortex.intelligence.' - Added 'darkplex' CLI alias alongside 'cortex' - Package renamed to darkplex-core v0.2.0 - 405 tests passing (was 234) - 14 new test files covering all merged modules
112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
"""Tests for intelligence/collective module."""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
from cortex.intelligence.shared_memory import Insight, SharedMemory, ALLOWED_AGENTS
|
|
from cortex.intelligence.collective import AggregatedPattern, CollectiveLearning
|
|
|
|
|
|
class TestCollectiveLearningInit:
|
|
def test_init(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
assert cl._patterns == []
|
|
assert len(cl._insights_by_topic) == 0
|
|
|
|
|
|
class TestPatternDetection:
|
|
def test_no_patterns_with_single_agent(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
# Add insights from same agent
|
|
agent = list(ALLOWED_AGENTS)[0]
|
|
for i in range(5):
|
|
cl._insights_by_topic["infra"].append(
|
|
Insight(agent=agent, topic="infra", content=f"test {i}")
|
|
)
|
|
cl._detect_patterns()
|
|
assert len(cl._patterns) == 0
|
|
|
|
def test_pattern_with_multiple_agents(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
agents = list(ALLOWED_AGENTS)[:2]
|
|
cl._insights_by_topic["infra"].append(
|
|
Insight(agent=agents[0], topic="infra", content="observation 1")
|
|
)
|
|
cl._insights_by_topic["infra"].append(
|
|
Insight(agent=agents[1], topic="infra", content="observation 2")
|
|
)
|
|
cl._detect_patterns()
|
|
assert len(cl._patterns) == 1
|
|
assert cl._patterns[0].topic == "infra"
|
|
|
|
|
|
class TestGetPatterns:
|
|
def test_filter_by_topic(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
agents = list(ALLOWED_AGENTS)[:2]
|
|
for topic in ["infra", "security"]:
|
|
for agent in agents:
|
|
cl._insights_by_topic[topic].append(
|
|
Insight(agent=agent, topic=topic, content="test")
|
|
)
|
|
cl._detect_patterns()
|
|
assert len(cl.get_patterns(topic="infra")) == 1
|
|
|
|
def test_filter_by_confidence(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
agents = list(ALLOWED_AGENTS)[:2]
|
|
cl._insights_by_topic["low"].append(
|
|
Insight(agent=agents[0], topic="low", content="x", confidence=0.1)
|
|
)
|
|
cl._insights_by_topic["low"].append(
|
|
Insight(agent=agents[1], topic="low", content="y", confidence=0.1)
|
|
)
|
|
cl._detect_patterns()
|
|
assert len(cl.get_patterns(min_confidence=0.5)) == 0
|
|
|
|
|
|
class TestTopicSummary:
|
|
def test_empty(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
assert cl.get_topic_summary() == {}
|
|
|
|
|
|
class TestExportKnowledge:
|
|
def test_export_json(self):
|
|
import json
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
data = json.loads(cl.export_knowledge())
|
|
assert "patterns" in data
|
|
assert "topics" in data
|
|
assert "allowed_agents" in data
|
|
|
|
|
|
class TestHandleInsight:
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_non_allowed_agent(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
insight = Insight(agent="unauthorized_agent", topic="test", content="bad")
|
|
await cl._handle_insight(insight)
|
|
assert len(cl._insights_by_topic) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_accepts_allowed_agent(self):
|
|
sm = mock.AsyncMock(spec=SharedMemory)
|
|
cl = CollectiveLearning(sm)
|
|
agent = list(ALLOWED_AGENTS)[0]
|
|
insight = Insight(agent=agent, topic="test", content="good")
|
|
await cl._handle_insight(insight)
|
|
assert len(cl._insights_by_topic["test"]) == 1
|