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
126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
"""Tests for governance/policy.py — Policy Engine.
|
|
|
|
NOTE: This module exists only in darkplex-core. Tests written against the module API.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
# We need yaml for creating test fixtures
|
|
import yaml
|
|
|
|
|
|
def _write_policy(tmpdir, filename, data):
|
|
path = Path(tmpdir) / filename
|
|
path.write_text(yaml.dump(data))
|
|
return path
|
|
|
|
|
|
class TestRule:
|
|
def setup_method(self):
|
|
import sys
|
|
sys.path.insert(0, str(Path.home() / "repos" / "darkplex-core"))
|
|
from governance.policy import Rule
|
|
self.Rule = Rule
|
|
|
|
def test_matches_simple(self):
|
|
r = self.Rule(name="r1", conditions={"agent": "claudia"}, effect="allow")
|
|
assert r.matches({"agent": "claudia"})
|
|
|
|
def test_no_match(self):
|
|
r = self.Rule(name="r1", conditions={"agent": "claudia"}, effect="allow")
|
|
assert not r.matches({"agent": "other"})
|
|
|
|
def test_missing_key(self):
|
|
r = self.Rule(name="r1", conditions={"agent": "claudia"}, effect="allow")
|
|
assert not r.matches({})
|
|
|
|
def test_list_condition(self):
|
|
r = self.Rule(name="r1", conditions={"action": ["read", "write"]}, effect="allow")
|
|
assert r.matches({"action": "read"})
|
|
assert not r.matches({"action": "delete"})
|
|
|
|
def test_multiple_conditions(self):
|
|
r = self.Rule(name="r1", conditions={"agent": "claudia", "action": "send"}, effect="deny")
|
|
assert r.matches({"agent": "claudia", "action": "send"})
|
|
assert not r.matches({"agent": "claudia", "action": "read"})
|
|
|
|
|
|
class TestPolicyEngine:
|
|
def setup_method(self):
|
|
import sys
|
|
sys.path.insert(0, str(Path.home() / "repos" / "darkplex-core"))
|
|
from governance.policy import PolicyEngine
|
|
self.PolicyEngine = PolicyEngine
|
|
|
|
def test_empty_dir(self, tmp_path):
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path))
|
|
assert engine.policies == []
|
|
|
|
def test_nonexistent_dir(self, tmp_path):
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path / "nope"))
|
|
assert engine.policies == []
|
|
|
|
def test_load_policy(self, tmp_path):
|
|
_write_policy(tmp_path, "test.yaml", {
|
|
"name": "test-policy",
|
|
"description": "Test",
|
|
"version": "1.0.0",
|
|
"rules": [
|
|
{"name": "deny-external", "conditions": {"target": "external"}, "effect": "deny", "priority": 10},
|
|
],
|
|
})
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path))
|
|
assert len(engine.policies) == 1
|
|
assert engine.policies[0].name == "test-policy"
|
|
assert len(engine.policies[0].rules) == 1
|
|
|
|
def test_evaluate_no_match(self, tmp_path):
|
|
_write_policy(tmp_path, "test.yaml", {
|
|
"name": "p", "description": "", "version": "1",
|
|
"rules": [{"name": "r1", "conditions": {"agent": "x"}, "effect": "deny"}],
|
|
})
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path))
|
|
result = engine.evaluate({"agent": "y"})
|
|
assert result["verdict"] == "allow"
|
|
|
|
def test_evaluate_match_deny(self, tmp_path):
|
|
_write_policy(tmp_path, "test.yaml", {
|
|
"name": "p", "description": "", "version": "1",
|
|
"rules": [{"name": "r1", "conditions": {"target": "external"}, "effect": "deny", "priority": 5}],
|
|
})
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path))
|
|
result = engine.evaluate({"target": "external"})
|
|
assert result["verdict"] == "deny"
|
|
|
|
def test_priority_ordering(self, tmp_path):
|
|
_write_policy(tmp_path, "test.yaml", {
|
|
"name": "p", "description": "", "version": "1",
|
|
"rules": [
|
|
{"name": "allow-all", "conditions": {"agent": "claudia"}, "effect": "allow", "priority": 1},
|
|
{"name": "deny-ext", "conditions": {"agent": "claudia"}, "effect": "deny", "priority": 10},
|
|
],
|
|
})
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path))
|
|
result = engine.evaluate({"agent": "claudia"})
|
|
assert result["verdict"] == "deny" # higher priority wins
|
|
|
|
def test_reload(self, tmp_path):
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path))
|
|
assert len(engine.policies) == 0
|
|
_write_policy(tmp_path, "new.yaml", {
|
|
"name": "new", "description": "", "version": "1", "rules": [],
|
|
})
|
|
engine.reload()
|
|
assert len(engine.policies) == 1
|
|
|
|
def test_skips_schema_yaml(self, tmp_path):
|
|
_write_policy(tmp_path, "schema.yaml", {"name": "schema"})
|
|
_write_policy(tmp_path, "real.yaml", {
|
|
"name": "real", "description": "", "version": "1", "rules": [],
|
|
})
|
|
engine = self.PolicyEngine(policies_dir=str(tmp_path))
|
|
assert len(engine.policies) == 1
|
|
assert engine.policies[0].name == "real"
|