69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
|
"""Tests for PersistentLearningStore."""
|
||
|
|
|
||
|
|
import tempfile
|
||
|
|
|
||
|
|
from fusionagi.governance.adaptive_ethics import AdaptiveEthics
|
||
|
|
from fusionagi.governance.consequence_engine import ConsequenceEngine
|
||
|
|
from fusionagi.memory.persistent_learning import PersistentLearningStore
|
||
|
|
|
||
|
|
|
||
|
|
def test_save_and_load_consequences() -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
engine = ConsequenceEngine()
|
||
|
|
engine.record_choice(
|
||
|
|
choice_id="c1",
|
||
|
|
actor="test",
|
||
|
|
action_taken="act1",
|
||
|
|
estimated_risk=0.3,
|
||
|
|
estimated_reward=0.7,
|
||
|
|
)
|
||
|
|
engine.record_consequence("c1", outcome_positive=True, actual_risk_realized=0.1, actual_reward_gained=0.8)
|
||
|
|
|
||
|
|
store = PersistentLearningStore(data_dir=tmpdir)
|
||
|
|
path = store.save_consequences(engine)
|
||
|
|
assert path.endswith("consequences.json")
|
||
|
|
|
||
|
|
engine2 = ConsequenceEngine()
|
||
|
|
loaded = store.load_consequences(engine2)
|
||
|
|
assert loaded == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_save_and_load_ethics() -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
ethics = AdaptiveEthics()
|
||
|
|
ethics.record_experience(
|
||
|
|
action_type="file_read",
|
||
|
|
context_summary="reading file outside scope",
|
||
|
|
advisory_reason="out of scope",
|
||
|
|
proceeded=True,
|
||
|
|
outcome_positive=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
store = PersistentLearningStore(data_dir=tmpdir)
|
||
|
|
path = store.save_ethics(ethics)
|
||
|
|
assert path.endswith("ethics.json")
|
||
|
|
|
||
|
|
ethics2 = AdaptiveEthics()
|
||
|
|
loaded = store.load_ethics(ethics2)
|
||
|
|
assert loaded == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_save_risk_histories() -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
engine = ConsequenceEngine()
|
||
|
|
engine.record_choice("c1", actor="t", action_taken="act1", estimated_risk=0.5, estimated_reward=0.5)
|
||
|
|
engine.record_consequence("c1", outcome_positive=True, actual_risk_realized=0.2, actual_reward_gained=0.8)
|
||
|
|
|
||
|
|
store = PersistentLearningStore(data_dir=tmpdir)
|
||
|
|
path = store.save_risk_histories(engine)
|
||
|
|
assert path.endswith("risk_histories.json")
|
||
|
|
|
||
|
|
|
||
|
|
def test_load_nonexistent_returns_zero() -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
store = PersistentLearningStore(data_dir=tmpdir)
|
||
|
|
engine = ConsequenceEngine()
|
||
|
|
assert store.load_consequences(engine) == 0
|
||
|
|
ethics = AdaptiveEthics()
|
||
|
|
assert store.load_ethics(ethics) == 0
|