tests/test_memory.py

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from models import ActionType, OpsguardAction
from server.opsguard_environment import OpsguardEnvironment


def test_query_history_returns_episode_memory():
    env = OpsguardEnvironment()
    env.reset(scenario_id="E1_typosquat_wave", seed=0)
    seen_authors = set()
    for _ in range(15):
        obs = env.step(OpsguardAction(action_type=ActionType.LABEL, label="bug"))
        if obs.current_issue:
            seen_authors.add(obs.current_issue.author_login)
        if obs.done:
            break
    obs = env.step(OpsguardAction(action_type=ActionType.QUERY_HISTORY, query=list(seen_authors)[0]))
    assert obs.memory_hits, "expected memory hits from prior steps"
    assert any(h.get("source") == "episode_memory" for h in obs.memory_hits)


def test_memory_persists_across_long_horizon():
    env = OpsguardEnvironment()
    env.reset(scenario_id="E2_social_eng_buildup", seed=0)
    for _ in range(80):
        obs = env.step(OpsguardAction(action_type=ActionType.LABEL, label="bug"))
        if obs.done:
            break
    assert env._episode is not None
    assert env._episode.memory.stats()["episodic"] >= 30


if __name__ == "__main__":
    test_query_history_returns_episode_memory()
    print("PASS test_query_history_returns_episode_memory")
    test_memory_persists_across_long_horizon()
    print("PASS test_memory_persists_across_long_horizon")
    print("ALL GREEN")