forked from Rockachopa/Timmy-time-dashboard
134 lines
4.1 KiB
Python
134 lines
4.1 KiB
Python
"""Tests for the three-phase loop scaffold.
|
|
|
|
Validates the acceptance criteria from issue #324:
|
|
1. Loop accepts context payload as input to Phase 1
|
|
2. Phase 1 output feeds into Phase 2 without manual intervention
|
|
3. Phase 2 output feeds into Phase 3 without manual intervention
|
|
4. Phase 3 output feeds back into Phase 1
|
|
5. Full cycle completes without crash
|
|
6. No state leaks between cycles
|
|
7. Each phase logs what it received and what it produced
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
from loop.phase1_gather import gather
|
|
from loop.phase2_reason import reason
|
|
from loop.phase3_act import act
|
|
from loop.runner import run_cycle
|
|
from loop.schema import ContextPayload
|
|
|
|
|
|
def _make_payload(source: str = "test", content: str = "hello") -> ContextPayload:
|
|
return ContextPayload(source=source, content=content, token_count=5)
|
|
|
|
|
|
# --- Schema ---
|
|
|
|
|
|
def test_context_payload_defaults():
|
|
p = ContextPayload(source="user", content="hi")
|
|
assert p.source == "user"
|
|
assert p.content == "hi"
|
|
assert p.token_count == -1
|
|
assert p.metadata == {}
|
|
assert isinstance(p.timestamp, datetime)
|
|
|
|
|
|
def test_with_metadata_returns_new_payload():
|
|
p = _make_payload()
|
|
p2 = p.with_metadata(foo="bar")
|
|
assert p2.metadata == {"foo": "bar"}
|
|
assert p.metadata == {} # original unchanged
|
|
|
|
|
|
def test_with_metadata_merges():
|
|
p = _make_payload().with_metadata(a=1)
|
|
p2 = p.with_metadata(b=2)
|
|
assert p2.metadata == {"a": 1, "b": 2}
|
|
|
|
|
|
# --- Individual phases ---
|
|
|
|
|
|
def test_gather_marks_phase():
|
|
result = gather(_make_payload())
|
|
assert result.metadata["phase"] == "gather"
|
|
assert result.metadata["gathered"] is True
|
|
|
|
|
|
def test_reason_marks_phase():
|
|
gathered = gather(_make_payload())
|
|
result = reason(gathered)
|
|
assert result.metadata["phase"] == "reason"
|
|
assert result.metadata["reasoned"] is True
|
|
|
|
|
|
def test_act_marks_phase():
|
|
gathered = gather(_make_payload())
|
|
reasoned = reason(gathered)
|
|
result = act(reasoned)
|
|
assert result.metadata["phase"] == "act"
|
|
assert result.metadata["acted"] is True
|
|
|
|
|
|
# --- Full cycle ---
|
|
|
|
|
|
def test_full_cycle_completes():
|
|
"""Acceptance criterion 5: full cycle completes without crash."""
|
|
payload = _make_payload(source="user", content="What is sovereignty?")
|
|
result = run_cycle(payload)
|
|
assert result.metadata["gathered"] is True
|
|
assert result.metadata["reasoned"] is True
|
|
assert result.metadata["acted"] is True
|
|
|
|
|
|
def test_full_cycle_preserves_source():
|
|
"""Source field survives the full pipeline."""
|
|
result = run_cycle(_make_payload(source="timer"))
|
|
assert result.source == "timer"
|
|
|
|
|
|
def test_full_cycle_preserves_content():
|
|
"""Content field survives the full pipeline."""
|
|
result = run_cycle(_make_payload(content="test data"))
|
|
assert result.content == "test data"
|
|
|
|
|
|
def test_no_state_leaks_between_cycles():
|
|
"""Acceptance criterion 6: no state leaks between cycles."""
|
|
r1 = run_cycle(_make_payload(source="cycle1", content="first"))
|
|
r2 = run_cycle(_make_payload(source="cycle2", content="second"))
|
|
assert r1.source == "cycle1"
|
|
assert r2.source == "cycle2"
|
|
assert r1.content == "first"
|
|
assert r2.content == "second"
|
|
|
|
|
|
def test_cycle_output_feeds_back_as_input():
|
|
"""Acceptance criterion 4: Phase 3 output feeds back into Phase 1."""
|
|
first = run_cycle(_make_payload(source="initial"))
|
|
second = run_cycle(first)
|
|
# Second cycle should still work — no crash, metadata accumulates
|
|
assert second.metadata["gathered"] is True
|
|
assert second.metadata["acted"] is True
|
|
|
|
|
|
def test_phases_log(caplog):
|
|
"""Acceptance criterion 7: each phase logs what it received and produced."""
|
|
import logging
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
run_cycle(_make_payload())
|
|
|
|
messages = caplog.text
|
|
assert "Phase 1 (Gather) received" in messages
|
|
assert "Phase 1 (Gather) produced" in messages
|
|
assert "Phase 2 (Reason) received" in messages
|
|
assert "Phase 2 (Reason) produced" in messages
|
|
assert "Phase 3 (Act) received" in messages
|
|
assert "Phase 3 (Act) produced" in messages
|
|
assert "Loop cycle start" in messages
|
|
assert "Loop cycle complete" in messages
|