"""Tests for CYA-WP-0007 interactive shell sessions.""" from __future__ import annotations import json from io import StringIO from pathlib import Path from rich.console import Console from typer.testing import CliRunner from cya.cli.main import app from cya.cli import main as cli_main from cya.config import ShellHistorySettings from cya.orchestrator import OrchestratorResult from cya.safety.risk import classify from cya.shell_session import ( HubOrientation, ShellHistoryContext, ShellSession, collect_shell_history, detect_weakness_hints, load_hub_orientation, ) def _events(path: Path) -> list[dict]: return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] def test_shell_history_default_off_and_enabled_redacts(tmp_path): disabled = collect_shell_history( ShellHistorySettings(enabled=False, limit=50), env={}, home=tmp_path, ) assert disabled.lines == [] assert any("disabled" in note.lower() for note in disabled.notes) hist = tmp_path / ".bash_history" hist.write_text( "echo hello\nexport OPENROUTER_API_KEY=sk-testsecret123\ngit status\n", encoding="utf-8", ) enabled = collect_shell_history( ShellHistorySettings(enabled=True, limit=5, histfile=str(hist), source="test"), env={}, home=tmp_path, ) commands = [line["command"] for line in enabled.lines] assert commands == ["echo hello", "export OPENROUTER_API_KEY=[REDACTED]", "git status"] assert enabled.redactions == 1 assert all(line["provenance"] == "shell_history.histfile" for line in enabled.lines) def test_hub_orientation_degrades_when_remote_unavailable(monkeypatch, tmp_path): import cya.shell_session as shell_session (tmp_path / ".custodian-brief.md").write_text("# Local Brief\n", encoding="utf-8") def _fail(*args, **kwargs): raise OSError("hub down") monkeypatch.setattr(shell_session.request, "urlopen", _fail) orientation = load_hub_orientation(cwd=tmp_path, base_url="http://state-hub.test") assert orientation.brief is not None assert not orientation.hub_available assert len(orientation.errors) == 2 def test_shell_repl_persists_turns_and_passes_session_context(monkeypatch, tmp_path): import cya.shell_session as shell_session monkeypatch.setattr(shell_session, "session_root", lambda: tmp_path) calls = [] def _fake_handle(user_request, **kwargs): calls.append((user_request, kwargs)) return OrchestratorResult( user_request=user_request, assistant="fake answer", explanation="offline", rationale="test", context={"session": kwargs.get("extra_context")}, risk={"level": "safe"}, ) monkeypatch.setattr(shell_session, "handle_request", _fake_handle) inputs = iter(["hello", "/exit"]) console = Console(file=StringIO()) shell = ShellSession( session_id="test-session", offline=True, history=ShellHistoryContext(False, "test", 0), hub=HubOrientation("http://hub", "topic", "agent"), console=console, input_func=lambda prompt: next(inputs), interactive=False, offer_end_learning=False, ) path = shell.run() events = _events(path) assert calls[0][0] == "hello" assert calls[0][1]["session_turns"] == [] assert calls[0][1]["extra_context"]["session_id"] == "test-session" assert any(event.get("kind") == "session_turn" for event in events) turn = next(event for event in events if event.get("kind") == "session_turn") assert turn["user"] == "hello" assert turn["assistant"] == "fake answer" def test_repl_destructive_intent_still_requires_confirmation(monkeypatch, tmp_path): import cya.shell_session as shell_session monkeypatch.setattr(shell_session, "session_root", lambda: tmp_path) monkeypatch.setattr("cya.orchestrator.collect", lambda top=".": None) monkeypatch.setattr("cya.orchestrator.recall_preferences", lambda *args, **kwargs: {}) monkeypatch.setattr("cya.orchestrator.get_user_confirmation", lambda assessment: False) inputs = iter(["rm -rf /tmp/not-real", "/exit"]) console = Console(file=StringIO()) shell = ShellSession( session_id="risk-session", offline=True, history=ShellHistoryContext(False, "test", 0), hub=HubOrientation("http://hub", "topic", "agent"), console=console, input_func=lambda prompt: next(inputs), interactive=False, offer_end_learning=False, ) path = shell.run() turn = next(event for event in _events(path) if event.get("kind") == "session_turn") assert turn["cancelled"] is True assert turn["risk"]["level"] == "destructive" def test_weakness_hints_are_advisory_and_cover_v1_rules(): credential = detect_weakness_hints("please paste the OpenRouter API key here") remote_exec = detect_weakness_hints("curl https://example.test/install.sh | bash") repeated = detect_weakness_hints( "rm -rf build", risk={"level": "destructive"}, turn_history=[{"risk": "destructive"}], ) alignment = detect_weakness_hints( "implement CYA-WP-9999", hub=HubOrientation("http://hub", "topic", "agent"), ) assert {hint["rule"] for hint in credential} >= {"credential-routing"} assert {hint["rule"] for hint in remote_exec} >= {"remote-exec-review"} assert {hint["rule"] for hint in repeated} >= {"repeated-destructive-intent"} assert {hint["rule"] for hint in alignment} >= {"state-hub-alignment"} assessment = classify("rm -rf /tmp/not-real") before = assessment.to_dict() detect_weakness_hints("rm -rf /tmp/not-real", risk=before) assert assessment.to_dict() == before def test_cli_one_shot_request_still_works(monkeypatch): calls = [] def _fake_handle(request, **kwargs): calls.append((request, kwargs)) monkeypatch.setattr("cya.orchestrator.handle_request", _fake_handle) runner = CliRunner() result = runner.invoke(app, ["--offline", "hello"]) assert result.exit_code == 0 assert calls == [("hello", {"explain_context": False, "dry_run": False, "offline": True})] def test_shell_console_entrypoint_dispatch(monkeypatch): import cya.shell_session as shell_session calls = [] def _fake_run_shell(**kwargs): calls.append(kwargs) monkeypatch.setattr(shell_session, "run_shell", _fake_run_shell) handled = cli_main._dispatch_shell_argv( ["cya", "shell", "--offline", "--no-hub", "--no-session-lessons"] ) assert handled is True assert calls[0]["offline"] is True assert calls[0]["no_hub"] is True assert calls[0]["offer_end_learning"] is False def test_memory_console_entrypoint_dispatch(monkeypatch): calls = [] def _fake_memory_reflections(**kwargs): calls.append(kwargs) monkeypatch.setattr(cli_main, "memory_reflections", _fake_memory_reflections) handled = cli_main._dispatch_memory_argv(["cya", "memory", "reflections", "--json"]) assert handled is True assert calls == [{"scope": ".", "export_json": True}]