From 8326e1316f45c09774047c5e50391e1133d88702 Mon Sep 17 00:00:00 2001 From: rohansharma0509 Date: Sat, 25 Jul 2026 00:00:37 +0000 Subject: [PATCH] test: cover run transcript projection, job/config render, process control, package metadata Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- tests/test_package_integration.py | 136 +++++++ tests/test_render_config.py | 116 ++++++ tests/test_render_jobs.py | 437 +++++++++++++++++++++ tests/test_run_process.py | 217 +++++++++++ tests/test_run_transcript_projection.py | 484 ++++++++++++++++++++++++ tests/test_setup_wizard_terminal.py | 80 ++++ 6 files changed, 1470 insertions(+) create mode 100644 tests/test_package_integration.py create mode 100644 tests/test_render_config.py create mode 100644 tests/test_render_jobs.py create mode 100644 tests/test_run_transcript_projection.py create mode 100644 tests/test_setup_wizard_terminal.py diff --git a/tests/test_package_integration.py b/tests/test_package_integration.py new file mode 100644 index 00000000..3f366c32 --- /dev/null +++ b/tests/test_package_integration.py @@ -0,0 +1,136 @@ +import subprocess +import sys +from importlib.metadata import PackageNotFoundError + +import pytest + +from codealmanac.integrations.updates import package as package_module +from codealmanac.integrations.updates.package import ( + InstalledPackageMetadataProvider, + SubprocessPackageCommandRunner, + clean_optional_text, + read_direct_url, +) + + +class FakeDistribution: + def __init__(self, version: str, files: dict[str, str]): + self.version = version + self.files = files + + def read_text(self, filename: str) -> str | None: + return self.files.get(filename) + + +def install_distribution(monkeypatch, distribution: FakeDistribution | None) -> None: + def read(name: str) -> FakeDistribution: + if distribution is None: + raise PackageNotFoundError(name) + return distribution + + monkeypatch.setattr(package_module, "distribution", read) + + +def test_metadata_provider_reports_unknown_when_package_is_missing(monkeypatch): + install_distribution(monkeypatch, None) + + metadata = InstalledPackageMetadataProvider().read() + + assert metadata.version == "unknown" + assert metadata.installer is None + assert metadata.editable is False + assert metadata.source_url is None + + +def test_metadata_provider_reads_installer_and_editable_direct_url(monkeypatch): + install_distribution( + monkeypatch, + FakeDistribution( + version="0.4.7", + files={ + "INSTALLER": " uv \n", + "direct_url.json": ( + '{"url": "file:///repo", "dir_info": {"editable": true}}' + ), + }, + ), + ) + + metadata = InstalledPackageMetadataProvider().read() + + assert metadata.version == "0.4.7" + assert metadata.installer == "uv" + assert metadata.editable is True + assert metadata.source_url == "file:///repo" + + +def test_metadata_provider_ignores_blank_installer_and_absent_direct_url(monkeypatch): + install_distribution( + monkeypatch, + FakeDistribution(version="0.4.7", files={"INSTALLER": " "}), + ) + + metadata = InstalledPackageMetadataProvider().read() + + assert metadata.installer is None + assert metadata.editable is False + assert metadata.source_url is None + + +@pytest.mark.parametrize( + "raw", + [ + None, + "not json", + '["pip"]', + '{"dir_info": "editable"}', + '{"url": " "}', + '{"url": 7}', + ], +) +def test_read_direct_url_falls_back_to_defaults_for_unusable_payloads(raw): + metadata = read_direct_url(raw) + + assert metadata.editable is False + assert metadata.source_url is None + + +def test_read_direct_url_requires_explicit_editable_flag(): + metadata = read_direct_url('{"url": "file:///repo", "dir_info": {"editable": 1}}') + + assert metadata.editable is False + assert metadata.source_url == "file:///repo" + + +@pytest.mark.parametrize( + ("value", "expected"), + [(None, None), ("", None), (" ", None), (" pip ", "pip")], +) +def test_clean_optional_text(value, expected): + assert clean_optional_text(value) == expected + + +def test_command_runner_captures_output_streams_and_exit_code(): + result = SubprocessPackageCommandRunner().run( + ( + sys.executable, + "-c", + "import sys; print('out'); print('err', file=sys.stderr); sys.exit(3)", + ) + ) + + assert result.exit_code == 3 + assert result.stdout.strip() == "out" + assert result.stderr.strip() == "err" + + +def test_command_runner_reports_os_error_as_failed_result(monkeypatch): + def explode(*args, **kwargs): + raise FileNotFoundError("no such tool") + + monkeypatch.setattr(subprocess, "run", explode) + + result = SubprocessPackageCommandRunner().run(("missing-tool", "install")) + + assert result.exit_code == 1 + assert result.stderr == "FileNotFoundError: no such tool" diff --git a/tests/test_render_config.py b/tests/test_render_config.py new file mode 100644 index 00000000..69e741a7 --- /dev/null +++ b/tests/test_render_config.py @@ -0,0 +1,116 @@ +import json +from datetime import timedelta +from pathlib import Path + +from codealmanac.cli.render.config import ( + render_config_apply, + render_config_entry, + render_config_set, + render_config_values, +) +from codealmanac.services.automation.models import ( + AutomationTask, + AutomationTaskApplyResult, +) +from codealmanac.services.config.models import ( + ConfigApplyResult, + ConfigEntry, + ConfigKey, + ConfigSetResult, +) + + +def apply_result(task: AutomationTask, enabled: bool) -> AutomationTaskApplyResult: + return AutomationTaskApplyResult( + task=task, + enabled=enabled, + interval=timedelta(hours=1), + plist_path=Path(f"/tmp/{task.value}.plist"), + changed=True, + ) + + +def test_render_config_values_prints_a_key_value_table(capsys): + render_config_values( + ( + ConfigEntry(key=ConfigKey.AUTO_COMMIT, value="true"), + ConfigEntry(key=ConfigKey.HARNESS_DEFAULT, value="claude"), + ), + json_output=False, + ) + + lines = capsys.readouterr().out.splitlines() + assert lines[0].split() == ["KEY", "VALUE"] + assert lines[1].split() == ["auto_commit", "true"] + assert lines[2].split() == ["harness.default", "claude"] + + +def test_render_config_values_json_emits_rows(capsys): + render_config_values( + (ConfigEntry(key=ConfigKey.AUTO_COMMIT, value="true"),), + json_output=True, + ) + + assert json.loads(capsys.readouterr().out) == [ + {"key": "auto_commit", "value": "true"} + ] + + +def test_render_config_entry_prints_only_the_value(capsys): + render_config_entry( + ConfigEntry(key=ConfigKey.HARNESS_MODEL, value="sonnet"), + json_output=False, + ) + + assert capsys.readouterr().out == "sonnet\n" + + +def test_render_config_entry_json_emits_the_entry(capsys): + render_config_entry( + ConfigEntry(key=ConfigKey.HARNESS_MODEL, value="sonnet"), + json_output=True, + ) + + assert json.loads(capsys.readouterr().out)["key"] == "harness.model" + + +def test_render_config_set_echoes_the_assignment(capsys): + result = ConfigSetResult( + path="/tmp/config.toml", + key=ConfigKey.AUTO_COMMIT, + value="false", + ) + + render_config_set(result, json_output=False) + render_config_set(result, json_output=True) + + out = capsys.readouterr().out.splitlines() + assert out[0] == "config: auto_commit = false" + assert json.loads("\n".join(out[1:]))["value"] == "false" + + +def test_render_config_apply_lists_automation_task_states(capsys): + result = ConfigApplyResult( + path="/tmp/config.toml", + automation=( + apply_result(AutomationTask.SYNC, True), + apply_result(AutomationTask.GARDEN, False), + ), + ) + + render_config_apply(result, json_output=False) + + assert capsys.readouterr().out.splitlines() == [ + "config applied", + " sync: enabled", + " garden: disabled", + ] + + +def test_render_config_apply_json_emits_the_result(capsys): + render_config_apply( + ConfigApplyResult(path="/tmp/config.toml", automation=()), + json_output=True, + ) + + assert json.loads(capsys.readouterr().out)["path"] == "/tmp/config.toml" diff --git a/tests/test_render_jobs.py b/tests/test_render_jobs.py new file mode 100644 index 00000000..6f8f779d --- /dev/null +++ b/tests/test_render_jobs.py @@ -0,0 +1,437 @@ +import json +from datetime import UTC, datetime + +import pytest + +from codealmanac.cli.render.job_logs import ( + detail_extras, + render_run_attach, + render_run_attach_detached, + render_run_attach_stream, + render_run_log, + truncate_middle, +) +from codealmanac.cli.render.jobs import ( + render_run, + render_run_cancel, + render_runs, +) +from codealmanac.services.harnesses.models import ( + HarnessEvent, + HarnessEventKind, + HarnessFailure, + HarnessKind, + HarnessToolDisplay, + HarnessToolDisplayKind, + HarnessToolStatus, + HarnessTranscriptRef, +) +from codealmanac.services.runs.models import ( + PageChangeSet, + RunAttachSnapshot, + RunAttachUpdate, + RunCancelResult, + RunEventKind, + RunKind, + RunLogEvent, + RunRecord, + RunStatus, +) + +CREATED_AT = datetime(2026, 7, 7, 12, 0, tzinfo=UTC) + + +def record( + *, + status: RunStatus = RunStatus.DONE, + title: str | None = "Garden the wiki", + summary: str | None = None, + error: str | None = None, + page_changes: PageChangeSet | None = None, + harness_transcript: HarnessTranscriptRef | None = None, +) -> RunRecord: + return RunRecord( + run_id="run-1", + repository_id="repo-1", + kind=RunKind.GARDEN, + status=status, + title=title, + summary=summary, + error=error, + created_at=CREATED_AT, + updated_at=CREATED_AT, + started_at=CREATED_AT, + finished_at=CREATED_AT if status == RunStatus.DONE else None, + page_changes=page_changes, + harness_transcript=harness_transcript, + ) + + +def log_event( + sequence: int, + *, + kind: RunEventKind = RunEventKind.STATUS, + message: str = "queued", + harness_event: HarnessEvent | None = None, +) -> RunLogEvent: + return RunLogEvent( + run_id="run-1", + sequence=sequence, + timestamp=CREATED_AT, + kind=kind, + message=message, + harness_event=harness_event, + ) + + +def tool_event(sequence: int, status: HarnessToolStatus) -> RunLogEvent: + return log_event( + sequence, + kind=RunEventKind.TOOL, + message="reading", + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_USE, + message="reading", + tool_id="tool-1", + tool_display=HarnessToolDisplay( + kind=HarnessToolDisplayKind.READ, + title="Read", + path="src/a.py", + status=status, + duration_ms=12, + ), + ), + ) + + +def test_render_runs_lists_a_table_of_jobs(capsys): + render_runs((record(),), json_output=False) + + lines = capsys.readouterr().out.splitlines() + assert lines[0].split() == ["ID", "KIND", "STATUS", "ELAPSED", "TITLE"] + assert lines[1].startswith("run-1 garden done") + assert lines[1].endswith("Garden the wiki") + + +def test_render_runs_reports_empty_result_on_stderr(capsys): + render_runs((), json_output=False) + + captured = capsys.readouterr() + assert captured.out == "" + assert captured.err.strip() == "# 0 jobs" + + +def test_render_runs_json_emits_rows(capsys): + render_runs((record(),), json_output=True) + + rows = json.loads(capsys.readouterr().out) + assert [row["run_id"] for row in rows] == ["run-1"] + + +def test_render_runs_shows_queued_jobs_without_elapsed_time(capsys): + render_runs((record(status=RunStatus.QUEUED, title=None),), json_output=False) + + assert "—" in capsys.readouterr().out + + +def test_render_runs_labels_every_status(capsys): + render_runs( + tuple(record(status=status) for status in RunStatus), + json_output=False, + ) + + rows = capsys.readouterr().out.splitlines()[1:] + assert [row.split()[2] for row in rows] == [status.value for status in RunStatus] + + +def test_render_run_prints_detail_lines(capsys): + render_run( + record( + summary="tidied 3 pages", + page_changes=PageChangeSet( + created=("alpha",), + updated=("beta", "gamma"), + ), + harness_transcript=HarnessTranscriptRef( + kind=HarnessKind.CLAUDE, + session_id="session-1", + transcript_path="/tmp/transcript.jsonl", + ), + ), + json_output=False, + ) + + out = capsys.readouterr().out + assert "job: run-1" in out + assert "kind: garden" in out + assert "status: done" in out + assert "title: Garden the wiki" in out + assert "summary: tidied 3 pages" in out + assert "changes: 1 created, 2 updated, 0 deleted" in out + assert "created: alpha" in out + assert "updated: beta, gamma" in out + assert "deleted:" not in out + assert "session: claude session-1" in out + assert "transcript: /tmp/transcript.jsonl" in out + assert "logs: codealmanac jobs logs run-1" in out + + +def test_render_run_prints_error_and_no_change_summary(capsys): + render_run( + record( + status=RunStatus.FAILED, + title=None, + error="provider timed out", + page_changes=PageChangeSet(), + ), + json_output=False, + ) + + out = capsys.readouterr().out + assert "changes: none" in out + assert "error: provider timed out" in out + assert "title:" not in out + assert "summary:" not in out + + +def test_render_run_omits_changes_for_a_job_that_reported_none(capsys): + render_run(record(), json_output=False) + + assert "changes:" not in capsys.readouterr().out + + +def test_render_run_json_emits_the_record(capsys): + render_run(record(), json_output=True) + + assert json.loads(capsys.readouterr().out)["run_id"] == "run-1" + + +def test_render_run_cancel_reports_changed_and_unchanged_jobs(capsys): + render_run_cancel( + RunCancelResult(record=record(status=RunStatus.CANCELLED), changed=True), + json_output=False, + ) + render_run_cancel( + RunCancelResult(record=record(), changed=False), + json_output=False, + ) + render_run_cancel( + RunCancelResult(record=record(), changed=True), + json_output=True, + ) + + out = capsys.readouterr().out.splitlines() + assert out[0] == "cancelled run-1" + assert out[1] == "job already done: run-1" + assert json.loads("\n".join(out[2:]))["changed"] is True + + +def test_render_run_log_prints_assistant_tool_and_note_steps(capsys): + render_run_log( + ( + log_event(1, message="worker_started"), + log_event( + 2, + kind=RunEventKind.MESSAGE, + message="thinking", + harness_event=HarnessEvent( + kind=HarnessEventKind.TEXT, + message="Reviewing the wiki\nfor stale pages", + ), + ), + tool_event(3, HarnessToolStatus.COMPLETED), + log_event(4, kind=RunEventKind.ERROR, message="boom"), + ), + json_output=False, + ) + + out = capsys.readouterr().out + assert "▌ assistant" in out + assert " Reviewing the wiki" in out + assert " for stale pages" in out + assert "● Read src/a.py" in out + assert "read · 12ms" in out + assert "✕ Error" in out + assert "· Worker Started" in out + + +def test_render_run_log_marks_pending_and_failed_tools(capsys): + render_run_log((tool_event(1, HarnessToolStatus.STARTED),), json_output=False) + started = capsys.readouterr().out + + render_run_log((tool_event(1, HarnessToolStatus.FAILED),), json_output=False) + failed = capsys.readouterr().out + + assert "◐ Read" in started + assert "✕ Read" in failed + assert "read · failed" in failed + + +def test_render_run_log_prints_agent_steps_without_a_target(capsys): + render_run_log( + ( + log_event( + 1, + kind=RunEventKind.TOOL, + message="agent", + harness_event=HarnessEvent( + kind=HarnessEventKind.AGENT_COMPLETED, + message="reviewer finished", + ), + ), + ), + json_output=False, + ) + + assert capsys.readouterr().out.strip() == "● Agent finished" + + +def test_render_run_log_prints_error_fix_below_the_message(capsys): + render_run_log( + ( + log_event( + 1, + kind=RunEventKind.TOOL, + message="error", + harness_event=HarnessEvent( + kind=HarnessEventKind.ERROR, + message="provider error", + failure=HarnessFailure( + provider=HarnessKind.CLAUDE, + message="rate limited", + fix="wait and retry", + ), + ), + ), + ), + json_output=False, + ) + + out = capsys.readouterr().out + assert "✕ Error" in out + assert " rate limited" in out + assert " wait and retry" in out + + +def test_render_run_log_json_emits_projected_steps(capsys): + render_run_log((log_event(1, message="worker_started"),), json_output=True) + + steps = json.loads(capsys.readouterr().out) + assert [step["title"] for step in steps] == ["Worker Started"] + assert "detail" not in steps[0] + + +def test_render_run_attach_prints_events_then_terminal_status(capsys): + render_run_attach( + RunAttachSnapshot( + record=record(summary="tidied 3 pages"), + events=(log_event(1, message="worker_started"),), + terminal=True, + ), + json_output=False, + ) + + out = capsys.readouterr().out + assert "· Worker Started" in out + assert "status: done" in out + assert "summary: tidied 3 pages" in out + + +def test_render_run_attach_reports_an_empty_log(capsys): + render_run_attach( + RunAttachSnapshot( + record=record(status=RunStatus.FAILED, error="provider timed out"), + events=(), + terminal=True, + ), + json_output=False, + ) + + out = capsys.readouterr().out + assert "no log events" in out + assert "status: failed" in out + assert "error: provider timed out" in out + + +def test_render_run_attach_json_emits_the_snapshot(capsys): + render_run_attach( + RunAttachSnapshot(record=record(), events=(), terminal=True), + json_output=True, + ) + + assert json.loads(capsys.readouterr().out)["record"]["run_id"] == "run-1" + + +def test_render_run_attach_stream_prints_steps_and_final_status(capsys): + render_run_attach_stream( + ( + RunAttachUpdate( + record=record(status=RunStatus.RUNNING), + events=(log_event(1, message="worker_started"),), + terminal=False, + ), + RunAttachUpdate(record=record(), events=(), terminal=True), + ), + json_output=False, + ) + + out = capsys.readouterr().out + assert "· Worker Started" in out + assert "no log events" not in out + assert out.strip().endswith("status: done") + + +def test_render_run_attach_stream_reports_a_run_without_events(capsys): + render_run_attach_stream( + (RunAttachUpdate(record=record(), events=(), terminal=True),), + json_output=False, + ) + + assert "no log events" in capsys.readouterr().out + + +def test_render_run_attach_stream_json_emits_one_update_per_line(capsys): + render_run_attach_stream( + ( + RunAttachUpdate(record=record(), events=(), terminal=False), + RunAttachUpdate(record=record(), events=(), terminal=True), + ), + json_output=True, + ) + + lines = capsys.readouterr().out.splitlines() + assert [json.loads(line)["terminal"] for line in lines] == [False, True] + + +def test_render_run_attach_detached_points_at_the_cancel_command(capsys): + render_run_attach_detached("run-1", json_output=False) + + out = capsys.readouterr().out + assert "detached — job continues in background" in out + assert "cancel: codealmanac jobs cancel run-1" in out + + +def test_render_run_attach_detached_stays_silent_for_json(capsys): + render_run_attach_detached("run-1", json_output=True) + + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize( + ("detail", "target", "expected"), + [ + (None, "src/a.py", None), + ("src/a.py", "src/a.py", None), + ("src/a.py · 12ms", "src/a.py", "12ms"), + ("exit 0 · 12ms", None, "exit 0 · 12ms"), + ], +) +def test_detail_extras_drops_paths_and_repeated_targets(detail, target, expected): + assert detail_extras(detail, target) == expected + + +def test_truncate_middle_keeps_head_and_tail(): + value = "a" * 40 + "/" + "b" * 40 + + assert truncate_middle(value, 20) == "aaaaaaaaaaa…bbbbbbbb" + assert truncate_middle("short", 20) == "short" diff --git a/tests/test_run_process.py b/tests/test_run_process.py index f4d67b36..6e96b680 100644 --- a/tests/test_run_process.py +++ b/tests/test_run_process.py @@ -11,15 +11,24 @@ from codealmanac.core.errors import ExecutionFailed from codealmanac.integrations.runs.process import ( PsutilRunProcessController, + SubprocessRunExecutorSpawner, + SubprocessRunWorkerSpawner, + executor_command, freeze_process_tree, + kill_process, + resume_processes, + resume_then_terminate, + worker_command, ) from codealmanac.services.runs.models import RunExecutionRef, RunKind, RunStatus from codealmanac.services.runs.requests import ( CancelRunRequest, MarkRunRunningRequest, + SpawnRunWorkerRequest, StartRunRequest, ) from codealmanac.settings import AppConfig +from codealmanac.workflows.run_queue.requests import SpawnRunExecutorRequest def test_process_controller_terminates_executor_and_separate_session_child(): @@ -160,6 +169,214 @@ def test_real_cancellation_stops_process_tree_before_sqlite_terminal_status( kill_if_alive(child_pid) +def test_freeze_process_tree_suspends_each_process_once(): + child = FakeProcess(pid=2) + root = FakeProcess(pid=1, children=[child, child]) + + frozen = freeze_process_tree(root) + + assert [process.pid for process in frozen] == [1, 2] + assert child.suspended == 1 + + +def test_freeze_process_tree_resumes_what_it_froze_when_suspension_fails(): + child = FakeProcess(pid=2, suspend_error=psutil.AccessDenied(2)) + root = FakeProcess(pid=1, children=[child]) + + with pytest.raises(ExecutionFailed, match="could not freeze run executor tree"): + freeze_process_tree(root) + + assert root.resumed == 1 + + +def test_freeze_process_tree_skips_children_that_exit_mid_walk(): + child = FakeProcess(pid=2, suspend_error=psutil.NoSuchProcess(2)) + root = FakeProcess(pid=1, children=[child]) + + frozen = freeze_process_tree(root) + + assert [process.pid for process in frozen] == [1] + + +def test_resume_then_terminate_ignores_a_process_that_already_exited(): + resume_then_terminate(FakeProcess(pid=1, resume_error=psutil.NoSuchProcess(1))) + + +def test_process_controller_resumes_the_tree_when_termination_is_denied(monkeypatch): + child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + + def deny(*args, **kwargs): + raise psutil.AccessDenied(child.pid) + + monkeypatch.setattr(psutil, "wait_procs", deny) + + try: + with pytest.raises(ExecutionFailed, match="could not terminate run executor"): + PsutilRunProcessController().terminate(execution_for(child.pid)) + assert psutil.Process(child.pid).status() != psutil.STATUS_STOPPED + finally: + kill_if_alive(child.pid) + child.wait(timeout=5) + + +def test_process_controller_reports_processes_that_survive_the_kill(monkeypatch): + child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + survivor = psutil.Process(child.pid) + + monkeypatch.setattr(psutil, "wait_procs", lambda procs, timeout: ([], [survivor])) + + try: + with pytest.raises(ExecutionFailed, match="did not stop"): + PsutilRunProcessController().terminate(execution_for(child.pid)) + finally: + kill_if_alive(child.pid) + child.wait(timeout=5) + + +def test_process_controller_reports_the_current_process_as_execution(): + execution = PsutilRunProcessController().current_execution() + + process = psutil.Process() + assert execution.pid == process.pid + assert execution.execution_id != "" + assert execution.process_started_at == datetime.fromtimestamp( + process.create_time(), UTC + ) + + +def test_process_controller_rejects_an_executor_that_already_exited(): + execution = RunExecutionRef( + execution_id="process-test", + pid=unused_pid(), + process_started_at=datetime.now(UTC), + ) + + with pytest.raises(ExecutionFailed, match="is gone before termination"): + PsutilRunProcessController().terminate(execution) + + +def test_worker_command_runs_the_worker_entrypoint_for_a_repository(tmp_path): + command = worker_command(SpawnRunWorkerRequest(cwd=tmp_path)) + + assert command == [ + sys.executable, + "-m", + "codealmanac.cli.main", + "__run-worker", + "--cwd", + str(tmp_path), + ] + + +def test_executor_command_runs_the_executor_entrypoint_for_a_run(): + command = executor_command(SpawnRunExecutorRequest(run_id="run-1")) + + assert command == [ + sys.executable, + "-m", + "codealmanac.cli.main", + "__run-executor", + "run-1", + ] + + +def test_executor_spawner_detaches_the_child_and_exposes_its_pid(monkeypatch): + calls = record_popen_calls(monkeypatch) + + process = SubprocessRunExecutorSpawner().spawn( + SpawnRunExecutorRequest(run_id="run-1") + ) + + assert process.pid == 4242 + assert process.wait() == 0 + assert calls[0]["command"][-2:] == ["__run-executor", "run-1"] + assert calls[0]["start_new_session"] is True + assert calls[0]["stdin"] == subprocess.DEVNULL + + +def test_worker_spawner_reports_the_command_it_launched(monkeypatch, tmp_path): + calls = record_popen_calls(monkeypatch) + + result = SubprocessRunWorkerSpawner().spawn(SpawnRunWorkerRequest(cwd=tmp_path)) + + assert result.child_pid == 4242 + assert result.command == tuple(worker_command(SpawnRunWorkerRequest(cwd=tmp_path))) + assert calls[0]["cwd"] == tmp_path + assert calls[0]["start_new_session"] is True + + +def test_resume_and_kill_helpers_tolerate_dead_processes(): + child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + process = psutil.Process(child.pid) + process.kill() + child.wait(timeout=5) + + resume_processes([process]) + kill_process(process) + + +class FakeProcess: + """Minimal psutil.Process stand-in for process-tree walk edge cases.""" + + def __init__( + self, + pid: int, + children: list["FakeProcess"] | None = None, + suspend_error: Exception | None = None, + resume_error: Exception | None = None, + ): + self.pid = pid + self._children = children or [] + self.suspend_error = suspend_error + self.resume_error = resume_error + self.suspended = 0 + self.resumed = 0 + self.terminated = 0 + + def create_time(self) -> float: + return float(self.pid) + + def children(self) -> list["FakeProcess"]: + return self._children + + def suspend(self) -> None: + if self.suspend_error is not None: + raise self.suspend_error + self.suspended += 1 + + def resume(self) -> None: + if self.resume_error is not None: + raise self.resume_error + self.resumed += 1 + + def terminate(self) -> None: + self.terminated += 1 + + +def record_popen_calls(monkeypatch) -> list[dict]: + calls: list[dict] = [] + + class FakeChild: + pid = 4242 + + def wait(self) -> int: + return 0 + + def fake_popen(command, **kwargs): + calls.append({"command": command, **kwargs}) + return FakeChild() + + monkeypatch.setattr(subprocess, "Popen", fake_popen) + return calls + + +def unused_pid() -> int: + candidate = 4242 + while psutil.pid_exists(candidate): + candidate += 1 + return candidate + + def execution_for(pid: int) -> RunExecutionRef: process = psutil.Process(pid) return RunExecutionRef( diff --git a/tests/test_run_transcript_projection.py b/tests/test_run_transcript_projection.py new file mode 100644 index 00000000..67692ae1 --- /dev/null +++ b/tests/test_run_transcript_projection.py @@ -0,0 +1,484 @@ +from datetime import UTC, datetime + +import pytest + +from codealmanac.services.harnesses.models import ( + HarnessActorConfidence, + HarnessActorRole, + HarnessAgentTrace, + HarnessEvent, + HarnessEventKind, + HarnessFailure, + HarnessKind, + HarnessRunActor, + HarnessRunStatus, + HarnessToolDisplay, + HarnessToolDisplayKind, + HarnessToolStatus, +) +from codealmanac.services.runs.models import RunEventKind, RunLogEvent +from codealmanac.services.runs.transcript import ( + RunStepKind, + join_text, + project_run_steps, + status_title, + tool_kind_label, + value_preview, +) + + +def log_event( + sequence: int, + *, + kind: RunEventKind = RunEventKind.TOOL, + message: str = "event", + harness_event: HarnessEvent | None = None, +) -> RunLogEvent: + return RunLogEvent( + run_id="run-1", + sequence=sequence, + timestamp=datetime(2026, 7, 7, 12, 0, tzinfo=UTC), + kind=kind, + message=message, + harness_event=harness_event, + ) + + +def actor(label: str) -> HarnessRunActor: + return HarnessRunActor( + thread_id="thread-1", + role=HarnessActorRole.ROOT, + label=label, + confidence=HarnessActorConfidence.PROVIDER, + ) + + +def test_run_events_without_harness_payload_become_status_and_error_steps(): + steps = project_run_steps( + ( + log_event(1, kind=RunEventKind.STATUS, message="worker_started"), + log_event(2, kind=RunEventKind.ERROR, message="boom"), + ) + ) + + assert [step.kind for step in steps] == [RunStepKind.STATUS, RunStepKind.ERROR] + assert steps[0].title == "Worker Started" + assert steps[0].error is False + assert steps[1].title == "Error" + assert steps[1].body == "boom" + assert steps[1].error is True + + +def test_consecutive_text_from_same_actor_merges_into_one_assistant_step(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TEXT, + message="first", + actor=actor("writer"), + ), + ), + log_event( + 2, + harness_event=HarnessEvent( + kind=HarnessEventKind.TEXT_DELTA, + message="second", + actor=actor("writer"), + ), + ), + ) + ) + + assert len(steps) == 1 + assert steps[0].kind == RunStepKind.ASSISTANT + assert steps[0].actor == "writer" + assert steps[0].body == "first\nsecond" + + +def test_text_from_a_different_actor_starts_a_new_assistant_step(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TEXT, + message="from writer", + actor=actor("writer"), + ), + ), + log_event( + 2, + harness_event=HarnessEvent( + kind=HarnessEventKind.TEXT, + message="from reviewer", + actor=actor("reviewer"), + ), + ), + ) + ) + + assert [step.actor for step in steps] == ["writer", "reviewer"] + assert steps[1].title == "reviewer" + + +def test_text_without_actor_titles_the_step_assistant(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TEXT, + message="hello", + ), + ), + ) + ) + + assert steps[0].title == "Assistant" + assert steps[0].actor is None + + +def test_tool_result_updates_the_matching_started_tool_step(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_USE, + message="reading file", + tool_id="tool-1", + tool_name="Read", + tool_input='{"path": "a.py"}', + tool_display=HarnessToolDisplay( + kind=HarnessToolDisplayKind.READ, + title="Read a.py", + path="src/a.py", + status=HarnessToolStatus.STARTED, + ), + ), + ), + log_event( + 2, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_RESULT, + message="done", + tool_id="tool-1", + tool_result="file body", + tool_display=HarnessToolDisplay( + kind=HarnessToolDisplayKind.READ, + path="src/a.py", + status=HarnessToolStatus.COMPLETED, + duration_ms=12, + summary="read 40 lines", + ), + ), + ), + ) + ) + + assert len(steps) == 1 + step = steps[0] + assert step.sequence == 2 + assert step.status == "completed" + assert step.body == "read 40 lines" + assert step.detail == "src/a.py · 12ms" + assert step.output == "file body" + assert step.tool == "read" + assert step.target == "src/a.py" + assert step.input == '{"path": "a.py"}' + assert step.error is False + + +def test_orphan_tool_result_creates_its_own_completed_step(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_SUMMARY, + message="ran a command", + tool_name="Shell", + tool_result="exit 0", + tool_is_error=False, + ), + ), + ) + ) + + assert len(steps) == 1 + assert steps[0].kind == RunStepKind.TOOL + assert steps[0].title == "Shell" + assert steps[0].status == "completed" + assert steps[0].output == "exit 0" + + +def test_failed_tool_result_marks_the_step_as_errored(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_USE, + message="running", + tool_id="tool-1", + tool_name="Shell", + ), + ), + log_event( + 2, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_RESULT, + message="failed", + tool_id="tool-1", + tool_is_error=True, + ), + ), + ) + ) + + assert steps[0].status == "failed" + assert steps[0].error is True + + +@pytest.mark.parametrize( + ("kind", "title", "status"), + [ + (HarnessEventKind.AGENT_SPAWNED, "Agent started", "started"), + (HarnessEventKind.AGENT_WAIT_STARTED, "Waiting for agent", "waiting"), + (HarnessEventKind.AGENT_COMPLETED, "Agent finished", "completed"), + ], +) +def test_agent_events_project_titles_and_statuses(kind, title, status): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=kind, + message="agent event", + agent_trace=HarnessAgentTrace(prompt="review", result="looks good"), + ), + ), + ) + ) + + assert steps[0].kind == RunStepKind.AGENT + assert steps[0].title == title + assert steps[0].status == status + assert steps[0].body == "looks good" + assert steps[0].detail == "review" + + +def test_agent_event_without_trace_uses_the_event_message(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.AGENT_SPAWNED, + message="spawned reviewer", + ), + ), + ) + ) + + assert steps[0].body == "spawned reviewer" + assert steps[0].detail is None + + +def test_error_event_carries_failure_message_and_fix(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.ERROR, + message="provider error", + status=HarnessRunStatus.FAILED, + failure=HarnessFailure( + provider=HarnessKind.CLAUDE, + message="rate limited", + fix="wait and retry", + ), + ), + ), + ) + ) + + assert steps[0].kind == RunStepKind.ERROR + assert steps[0].body == "rate limited" + assert steps[0].detail == "wait and retry" + assert steps[0].status == HarnessRunStatus.FAILED.value + assert steps[0].error is True + + +def test_done_event_with_failure_projects_an_error_step(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.DONE, + message="finished", + failure=HarnessFailure( + provider=HarnessKind.CLAUDE, + message="agent aborted", + ), + ), + ), + ) + ) + + assert steps[0].kind == RunStepKind.ERROR + assert steps[0].body == "agent aborted" + + +def test_warning_done_and_session_events_project_status_steps(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.WARNING, + message="slow provider", + ), + ), + log_event( + 2, + harness_event=HarnessEvent( + kind=HarnessEventKind.DONE, + message="all done", + ), + ), + log_event( + 3, + harness_event=HarnessEvent( + kind=HarnessEventKind.PROVIDER_SESSION, + message="session", + provider_session_id="session-1", + ), + ), + ) + ) + + assert [(step.title, step.body) for step in steps] == [ + ("Warning", "slow provider"), + ("Done", "all done"), + ("Session", "session-1"), + ] + + +def test_noise_events_are_dropped_from_the_transcript(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.CONTEXT_USAGE, + message="12000 tokens", + ), + ), + log_event( + 2, + harness_event=HarnessEvent( + kind=HarnessEventKind.PROVIDER_SESSION, + message="session", + ), + ), + log_event( + 3, + harness_event=HarnessEvent( + kind=HarnessEventKind.HOOK, + message="hook fired", + ), + ), + ) + ) + + assert steps == () + + +def test_tool_result_summary_falls_back_to_truncated_result_preview(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_RESULT, + message="done", + tool_name="Read", + tool_result="x" * 400, + ), + ), + ) + ) + + assert steps[0].body == "x" * 277 + "..." + + +def test_tool_without_display_or_name_is_titled_tool(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_USE, + message="doing something", + ), + ), + ) + ) + + assert steps[0].title == "Tool" + assert steps[0].tool is None + assert steps[0].target is None + assert steps[0].status == "started" + + +def test_tool_target_prefers_path_then_command_then_thread(): + steps = project_run_steps( + ( + log_event( + 1, + harness_event=HarnessEvent( + kind=HarnessEventKind.TOOL_USE, + message="shell", + tool_display=HarnessToolDisplay( + kind=HarnessToolDisplayKind.SHELL, + command="ls -la", + exit_code=0, + ), + ), + ), + ) + ) + + assert steps[0].target == "ls -la" + assert steps[0].detail == "ls -la · exit 0" + + +@pytest.mark.parametrize( + ("first", "second", "expected"), + [ + (None, "a", "a"), + ("a", "b", "a\nb"), + ("a\n", "b", "a\nb"), + ("a", "\nb", "a\nb"), + ], +) +def test_join_text(first, second, expected): + assert join_text(first, second) == expected + + +def test_status_title_normalizes_separators(): + assert status_title("worker-started_now") == "Worker Started Now" + + +def test_tool_kind_label_reads_as_prose(): + assert tool_kind_label(HarnessToolDisplayKind.WEB) == "web" + + +def test_value_preview_stringifies_non_text_values(): + assert value_preview({"ok": True}) == "{'ok': True}" diff --git a/tests/test_setup_wizard_terminal.py b/tests/test_setup_wizard_terminal.py new file mode 100644 index 00000000..c3c517cd --- /dev/null +++ b/tests/test_setup_wizard_terminal.py @@ -0,0 +1,80 @@ +import os +import pty +import sys +import termios + +import pytest + +from codealmanac.cli.dispatch.setup_wizard.terminal import ( + read_setup_key, + supports_interactive_setup, + wizard_terminal, +) + + +@pytest.fixture +def terminal(monkeypatch): + """A real pty wired to sys.stdin so the wizard sees a terminal.""" + controller, follower = pty.openpty() + previous = termios.tcgetattr(follower) + stdin = os.fdopen(follower, "rb", buffering=0) + monkeypatch.setattr(sys, "stdin", stdin) + try: + yield controller + finally: + termios.tcsetattr(follower, termios.TCSADRAIN, previous) + stdin.close() + os.close(controller) + + +def test_interactive_setup_needs_a_terminal_on_both_streams(terminal, monkeypatch): + assert supports_interactive_setup() is False + + monkeypatch.setattr(sys.stdout, "isatty", lambda: True) + + assert supports_interactive_setup() is True + + +def test_wizard_terminal_switches_to_the_alternate_screen_and_restores( + terminal, + capsys, +): + fd = sys.stdin.fileno() + before = termios.tcgetattr(fd) + + with wizard_terminal(): + during = termios.tcgetattr(fd) + + out = capsys.readouterr().out + assert out.startswith("\x1b[?1049h\x1b[?25l") + assert out.endswith("\x1b[?25h\x1b[?1049l") + assert during != before + assert termios.tcgetattr(fd) == before + + +def test_wizard_terminal_is_a_no_op_without_a_terminal(capsys): + with wizard_terminal(): + pass + + assert capsys.readouterr().out == "" + + +def test_read_setup_key_returns_a_single_character(terminal): + os.write(terminal, b"q") + + with wizard_terminal(): + assert read_setup_key() == "q" + + +def test_read_setup_key_reads_a_whole_arrow_sequence(terminal): + os.write(terminal, b"\x1b[B") + + with wizard_terminal(): + assert read_setup_key() == "\x1b[B" + + +def test_read_setup_key_returns_a_lone_escape_when_nothing_follows(terminal): + os.write(terminal, b"\x1b") + + with wizard_terminal(): + assert read_setup_key() == "\x1b"