From a997fc145cd17d2fbea93d0f8539eca0988943eb Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 7 Oct 2026 03:10:35 +0000 Subject: [PATCH] test(workbench): reuse workspace and lifecycle command fixtures --- .../tests/test_scan_contract_examples.py | 8 +- .../tests/test_workbench_cancellation.py | 156 ++++------------ .../codex-security/tests/test_workbench_db.py | 161 ++++------------ .../tests/test_workbench_handoff.py | 72 ++----- .../tests/test_workbench_progress.py | 105 +++-------- .../tests/workbench_test_support.py | 176 ++++++++++++++---- 6 files changed, 254 insertions(+), 424 deletions(-) diff --git a/plugins/codex-security/tests/test_scan_contract_examples.py b/plugins/codex-security/tests/test_scan_contract_examples.py index f3e27451b0..1d44d68b20 100644 --- a/plugins/codex-security/tests/test_scan_contract_examples.py +++ b/plugins/codex-security/tests/test_scan_contract_examples.py @@ -9,6 +9,7 @@ from jsonschema import Draft202012Validator, FormatChecker from referencing import Registry, Resource +from workbench_test_support import saved_coverage PLUGIN_DIR = Path(__file__).resolve().parent.parent EXAMPLE_DIR = PLUGIN_DIR / "examples" / "completed-scan" @@ -352,12 +353,7 @@ def test_findings_accept_code_evidence_call_stack_role(self) -> None: draft = { "scanId": "7fc17317-9594-49e0-b06a-d72fd7e14bba", "findings": [draft_finding], - "coverage": { - "completeness": "complete", - "surfaces": [], - "explicitExclusions": [], - "deferred": [], - }, + "coverage": saved_coverage(), } draft_validator.validate(draft) diff --git a/plugins/codex-security/tests/test_workbench_cancellation.py b/plugins/codex-security/tests/test_workbench_cancellation.py index 952b66a47c..4ab6a5dfef 100644 --- a/plugins/codex-security/tests/test_workbench_cancellation.py +++ b/plugins/codex-security/tests/test_workbench_cancellation.py @@ -4,9 +4,15 @@ from pathlib import Path from workbench_test_support import ( + cancel_scan, create_saved_workspace, - run_workbench, + fail_scan, + get_scan, + mark_handoff_delivered, + scan_claim_command, + scan_command, start_delivered_scan, + start_scan_command, write_completed_contract, ) @@ -16,39 +22,21 @@ def test_workbench_records_scan_failure(tmp_path: Path) -> None: target = tmp_path / "target" target.mkdir() saved = create_saved_workspace(state_dir, target) - started = run_workbench( - state_dir, - "start-scan", - "--workspace-id", - str(saved["id"]), - "--scan-root", - str(tmp_path / "scans"), + started = start_scan_command( + state_dir, str(saved["id"]), "--scan-root", str(tmp_path / "scans") ) scan_id = str(started["results"]["scanId"]) claim_token = str(uuid.uuid4()) - run_workbench( - state_dir, "claim-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) - failed = run_workbench( - state_dir, - "fail-scan", - "--scan-id", - scan_id, - "--message", - "Repository checkout became unavailable.", - "--claim-token", - claim_token, + scan_claim_command(state_dir, "claim-handoff-delivery", scan_id, claim_token) + failed = fail_scan( + state_dir, scan_id, "Repository checkout became unavailable.", "--claim-token", claim_token ) assert failed["scan"]["progress"]["status"] == "failed" assert failed["scan"]["failureMessage"] == "Repository checkout became unavailable." - delivered = run_workbench( - state_dir, "mark-handoff-delivered", "--scan-id", scan_id, "--claim-token", claim_token - ) + delivered = mark_handoff_delivered(state_dir, scan_id, claim_token) assert delivered["results"]["handoffStatus"] == "delivered" - replayed = run_workbench( - state_dir, "mark-handoff-delivered", "--scan-id", scan_id, "--claim-token", claim_token - ) + replayed = mark_handoff_delivered(state_dir, scan_id, claim_token) assert replayed["results"]["handoffStatus"] == "delivered" @@ -58,62 +46,32 @@ def test_workbench_cancels_running_scan_and_rejects_late_updates(tmp_path: Path) target.mkdir() thread_id = "thread-cancel-owner" saved = create_saved_workspace(state_dir, target, thread_id=thread_id) - started = run_workbench(state_dir, "start-scan", "--workspace-id", str(saved["id"])) + started = start_scan_command(state_dir, str(saved["id"])) scan_id = str(started["results"]["scanId"]) claim_token = str(uuid.uuid4()) - claimed = run_workbench( - state_dir, "claim-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) + claimed = scan_claim_command(state_dir, "claim-handoff-delivery", scan_id, claim_token) assert claimed["results"]["handoffClaimToken"] == claim_token - wrong_thread = run_workbench( - state_dir, - "cancel-scan", - "--scan-id", - scan_id, - "--thread-id", - "thread-cancel-other", - check=False, - ) + wrong_thread = cancel_scan(state_dir, scan_id, "thread-cancel-other", check=False) assert wrong_thread["returncode"] != 0 assert "owning Codex thread" in str(wrong_thread["stderr"]) - canceled = run_workbench( - state_dir, "cancel-scan", "--scan-id", scan_id, "--thread-id", thread_id - ) + canceled = cancel_scan(state_dir, scan_id, thread_id) assert canceled["results"]["progress"]["status"] == "canceled" assert canceled["results"]["canceledAt"] assert canceled["results"]["handoffClaimToken"] == claim_token - replayed = run_workbench( - state_dir, "cancel-scan", "--scan-id", scan_id, "--thread-id", thread_id - ) + replayed = cancel_scan(state_dir, scan_id, thread_id) assert replayed["results"]["canceledAt"] == canceled["results"]["canceledAt"] for command in ( ("update-progress", "--phase", "discovery"), ("complete-scan",), ): - rejected = run_workbench( - state_dir, - command[0], - "--scan-id", - scan_id, - *command[1:], - check=False, - ) + rejected = scan_command(state_dir, command[0], scan_id, *command[1:], check=False) assert rejected["returncode"] != 0 - delivered = run_workbench( - state_dir, - "mark-handoff-delivered", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - thread_id, - ) + delivered = mark_handoff_delivered(state_dir, scan_id, claim_token, "--thread-id", thread_id) assert delivered["results"]["progress"]["status"] == "canceled" assert delivered["results"]["handoffStatus"] == "delivered" @@ -121,21 +79,13 @@ def test_workbench_cancels_running_scan_and_rejects_late_updates(tmp_path: Path) restarted_scan_id = str(restarted["results"]["scanId"]) assert restarted_scan_id != scan_id assert restarted["results"]["progress"]["status"] == "running" - previous_scan = run_workbench(state_dir, "get-scan", "--scan-id", scan_id) + previous_scan = get_scan(state_dir, scan_id) assert previous_scan["scan"]["scanId"] == scan_id assert previous_scan["workspace"]["results"]["scanId"] == scan_id assert previous_scan["workspace"]["results"]["progress"]["status"] == "canceled" write_completed_contract(Path(str(restarted["results"]["scanDir"])), restarted_scan_id, target) - run_workbench(state_dir, "complete-scan", "--scan-id", restarted_scan_id) - rejected = run_workbench( - state_dir, - "cancel-scan", - "--scan-id", - restarted_scan_id, - "--thread-id", - thread_id, - check=False, - ) + scan_command(state_dir, "complete-scan", restarted_scan_id) + rejected = cancel_scan(state_dir, restarted_scan_id, thread_id, check=False) assert rejected["returncode"] != 0 assert "Only a running scan can be canceled" in str(rejected["stderr"]) @@ -146,50 +96,23 @@ def test_workbench_rejects_unconfirmed_cross_thread_handoff_delivery(tmp_path: P target.mkdir() thread_id = "thread-handoff-owner" saved = create_saved_workspace(state_dir, target, thread_id=thread_id) - started = run_workbench(state_dir, "start-scan", "--workspace-id", str(saved["id"])) + started = start_scan_command(state_dir, str(saved["id"])) scan_id = str(started["results"]["scanId"]) claim_token = str(uuid.uuid4()) - run_workbench( - state_dir, "claim-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) + scan_claim_command(state_dir, "claim-handoff-delivery", scan_id, claim_token) - wrong_thread = run_workbench( - state_dir, - "mark-handoff-delivered", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - "thread-handoff-other", - check=False, + wrong_thread = mark_handoff_delivered( + state_dir, scan_id, claim_token, "--thread-id", "thread-handoff-other", check=False ) assert wrong_thread["returncode"] != 0 assert "owning Codex thread" in str(wrong_thread["stderr"]) - delivered = run_workbench( - state_dir, - "mark-handoff-delivered", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - thread_id, - ) + delivered = mark_handoff_delivered(state_dir, scan_id, claim_token, "--thread-id", thread_id) assert delivered["results"]["handoffStatus"] == "delivered" assert delivered["results"]["handoffClaimToken"] == claim_token - wrong_replay = run_workbench( - state_dir, - "mark-handoff-delivered", - "--scan-id", - scan_id, - "--claim-token", - str(uuid.uuid4()), - "--thread-id", - thread_id, - check=False, + wrong_replay = mark_handoff_delivered( + state_dir, scan_id, str(uuid.uuid4()), "--thread-id", thread_id, check=False ) assert wrong_replay["returncode"] != 0 assert "owned by another continuation" in str(wrong_replay["stderr"]) @@ -200,22 +123,13 @@ def test_workbench_allows_user_confirmed_recovery_in_another_thread(tmp_path: Pa target = tmp_path / "target" target.mkdir() saved = create_saved_workspace(state_dir, target, thread_id="thread-recovery-owner") - started = run_workbench(state_dir, "start-scan", "--workspace-id", str(saved["id"])) + started = start_scan_command(state_dir, str(saved["id"])) scan_id = str(started["results"]["scanId"]) claim_token = f"recovery_{uuid.uuid4()}" - run_workbench( - state_dir, "claim-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) + scan_claim_command(state_dir, "claim-handoff-delivery", scan_id, claim_token) - delivered = run_workbench( - state_dir, - "mark-handoff-delivered", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - "thread-recovery-confirmed", + delivered = mark_handoff_delivered( + state_dir, scan_id, claim_token, "--thread-id", "thread-recovery-confirmed" ) assert delivered["results"]["handoffStatus"] == "delivered" diff --git a/plugins/codex-security/tests/test_workbench_db.py b/plugins/codex-security/tests/test_workbench_db.py index 2395caa7f6..4e5004c57d 100644 --- a/plugins/codex-security/tests/test_workbench_db.py +++ b/plugins/codex-security/tests/test_workbench_db.py @@ -18,13 +18,23 @@ SCRIPT, create_saved_git_workspace, create_saved_workspace, + create_workspace, empty_target_scan, + fail_scan, + get_scan, initialize_git_repository, + mark_handoff_delivered, run_workbench, + save_workspace, + scan_claim_command, + set_triage, stable_target_id, start_delivered_scan, start_saved_scan, + start_scan_command, start_workspace_scan, + update_progress, + workspace_command, write_completed_contract, ) @@ -803,10 +813,8 @@ def test_workbench_persists_scan_model_and_updates_it_from_progress(tmp_path: Pa assert listed[0]["model"] == "gpt-5.6-sol" assert listed[0]["reasoningEffort"] == "high" - updated = run_workbench( + updated = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase", "discovery", @@ -818,14 +826,7 @@ def test_workbench_persists_scan_model_and_updates_it_from_progress(tmp_path: Pa assert updated["model"] == "gpt-5.6-terra" assert updated["reasoningEffort"] == "low" - preserved = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase", - "discovery", - )["scan"] + preserved = update_progress(state_dir, scan_id, "--phase", "discovery")["scan"] assert preserved["model"] == "gpt-5.6-terra" assert preserved["reasoningEffort"] == "low" @@ -838,14 +839,7 @@ def test_workbench_persists_progress_and_indexes_completed_findings(tmp_path: Pa workspace_id = str(saved["id"]) assert saved["userContext"] == "Pay attention to uploaded archives." - started = run_workbench( - state_dir, - "start-scan", - "--workspace-id", - workspace_id, - "--scan-root", - str(tmp_path / "scans"), - ) + started = start_scan_command(state_dir, workspace_id, "--scan-root", str(tmp_path / "scans")) results = started["results"] assert isinstance(results, dict) assert results["userContext"] == "Pay attention to uploaded archives." @@ -860,29 +854,19 @@ def test_workbench_persists_progress_and_indexes_completed_findings(tmp_path: Pa assert not (scan_dir / "events.jsonl").exists() claim_token = str(uuid.uuid4()) - claimed = run_workbench( - state_dir, "claim-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) + claimed = scan_claim_command(state_dir, "claim-handoff-delivery", scan_id, claim_token) assert claimed["results"]["handoffClaimedAt"] is not None assert claimed["results"]["handoffClaimToken"] == claim_token - released = run_workbench( - state_dir, "release-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) + released = scan_claim_command(state_dir, "release-handoff-delivery", scan_id, claim_token) assert released["results"]["handoffClaimedAt"] is None - claimed_again = run_workbench( - state_dir, "claim-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) + claimed_again = scan_claim_command(state_dir, "claim-handoff-delivery", scan_id, claim_token) assert claimed_again["results"]["handoffClaimedAt"] is not None - delivered = run_workbench( - state_dir, "mark-handoff-delivered", "--scan-id", scan_id, "--claim-token", claim_token - ) + delivered = mark_handoff_delivered(state_dir, scan_id, claim_token) assert delivered["results"]["handoffStatus"] == "delivered" assert delivered["results"]["handoffClaimedAt"] is None - updated = run_workbench( + updated = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase", "validation", @@ -903,9 +887,7 @@ def test_workbench_persists_progress_and_indexes_completed_findings(tmp_path: Pa assert updated["scan"]["progress"]["candidates"] == {"reportable": 1} write_completed_contract(scan_dir, scan_id, target) - completed = run_workbench( - state_dir, "complete-scan", "--scan-id", scan_id, "--claim-token", claim_token - ) + completed = scan_claim_command(state_dir, "complete-scan", scan_id, claim_token) completed_scan = completed["scan"] assert completed_scan["progress"]["status"] == "complete" assert completed_scan["findingCount"] == 1 @@ -942,7 +924,7 @@ def test_workbench_persists_progress_and_indexes_completed_findings(tmp_path: Pa assert finding["remediationTests"] == ["Reject traversal entries during extraction."] assert finding["locations"][0]["absolutePath"] == str(target / "src" / "extract.py") - reopened = run_workbench(state_dir, "get-workspace", "--workspace-id", workspace_id) + reopened = workspace_command(state_dir, "get-workspace", workspace_id) assert reopened["results"]["findings"][0]["findingId"].startswith("csf_") assert reopened["results"]["findings"][0]["occurrenceId"].startswith("occ_") assert reopened["results"]["findings"][0]["attackPath"]["reachability"] == ( @@ -970,29 +952,20 @@ def test_workbench_persists_progress_and_indexes_completed_findings(tmp_path: Pa occurrence_id, ), ) - aliased = run_workbench(state_dir, "get-scan", "--scan-id", scan_id) + aliased = get_scan(state_dir, scan_id) assert aliased["scan"]["findings"][0]["rootCause"] == { "code": "destination.write_bytes(entry.read())", "evidenceRefs": ["archive-write"], "summary": "Legacy containment details remain visible.", } - run_workbench( - state_dir, - "set-finding-triage", - "--occurrence-id", - occurrence_id, - "--status", - "closed", - "--close-reason", - "already_fixed", - ) + set_triage(state_dir, occurrence_id, "closed", "--close-reason", "already_fixed") with sqlite3.connect(database) as connection: connection.execute( "UPDATE finding_occurrences SET details_json = '{}' WHERE id = ?", (occurrence_id,), ) - backfilled = run_workbench(state_dir, "get-workspace", "--workspace-id", workspace_id) + backfilled = workspace_command(state_dir, "get-workspace", workspace_id) assert backfilled["results"]["findings"][0]["attackPath"]["impact"]["level"] == "high" assert backfilled["results"]["findings"][0]["triage"]["status"] == "closed" @@ -1005,7 +978,7 @@ def test_workbench_persists_progress_and_indexes_completed_findings(tmp_path: Pa """, (occurrence_id,), ) - poisoned = run_workbench(state_dir, "get-scan", "--scan-id", scan_id) + poisoned = get_scan(state_dir, scan_id) assert "attackPath" not in poisoned["scan"]["findings"][0] assert poisoned["scan"]["findings"][0]["title"] == finding["title"] @@ -3628,10 +3601,8 @@ def test_workbench_rejects_progress_completed_above_total(tmp_path: Path) -> Non target.mkdir() saved = create_saved_workspace(state_dir, target) started = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"])) - failed = run_workbench( + failed = update_progress( state_dir, - "update-progress", - "--scan-id", str(started["results"]["scanId"]), "--review-items-total", "2", @@ -3650,10 +3621,8 @@ def test_workbench_rejects_regressive_progress(tmp_path: Path) -> None: saved = create_saved_workspace(state_dir, target) started = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"])) scan_id = str(started["results"]["scanId"]) - run_workbench( + update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase", "validation", @@ -3663,26 +3632,12 @@ def test_workbench_rejects_regressive_progress(tmp_path: Path) -> None: "6", ) - phase_failed = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase", - "discovery", - check=False, - ) + phase_failed = update_progress(state_dir, scan_id, "--phase", "discovery", check=False) assert phase_failed["returncode"] != 0 assert "earlier phase" in str(phase_failed["stderr"]) - coverage_failed = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--review-items-completed", - "5", - check=False, + coverage_failed = update_progress( + state_dir, scan_id, "--review-items-completed", "5", check=False ) assert coverage_failed["returncode"] != 0 assert "cannot decrease" in str(coverage_failed["stderr"]) @@ -3693,36 +3648,14 @@ def test_workbench_tracks_review_pass_for_deep_scan_only(tmp_path: Path) -> None target = tmp_path / "target" target.mkdir() workspace_id = str(uuid.uuid4()) - run_workbench( - state_dir, - "create-workspace", - "--workspace-id", - workspace_id, - "--target-path", - str(target), - "--mode", - "deep", - ) - saved = run_workbench( - state_dir, - "save-workspace", - "--workspace-id", - workspace_id, - "--target-path", - str(target), - "--scope", - ".", - "--mode", - "deep", - ) + create_workspace(state_dir, workspace_id, "--target-path", str(target), "--mode", "deep") + saved = save_workspace(state_dir, workspace_id, str(target), ".", "deep") started = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"])) scan_id = str(started["results"]["scanId"]) assert started["results"]["progress"]["reviewPass"] is None - run_workbench( + update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase", "discovery", @@ -3733,14 +3666,7 @@ def test_workbench_tracks_review_pass_for_deep_scan_only(tmp_path: Path) -> None "--review-items-completed", "0", ) - updated = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--review-items-completed", - "22", - ) + updated = update_progress(state_dir, scan_id, "--review-items-completed", "22") assert updated["scan"]["progress"]["reviewPass"] == 2 assert updated["scan"]["progress"]["coverage"] == { "closedRows": 22, @@ -3756,14 +3682,8 @@ def test_workbench_tracks_review_pass_for_deep_scan_only(tmp_path: Path) -> None "--workspace-id", str(standard["id"]), ) - failed = run_workbench( - state_dir, - "update-progress", - "--scan-id", - str(standard_scan["results"]["scanId"]), - "--deep-review-pass", - "1", - check=False, + failed = update_progress( + state_dir, str(standard_scan["results"]["scanId"]), "--deep-review-pass", "1", check=False ) assert failed["returncode"] != 0 assert "Only Deep Scan" in str(failed["stderr"]) @@ -3778,18 +3698,11 @@ def test_workbench_updates_progress_timestamp_for_phase_and_failure(tmp_path: Pa scan_id = str(started["results"]["scanId"]) started_at = str(started["results"]["progress"]["updatedAt"]) time.sleep(0.001) - updated = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase", - "validation", - ) + updated = update_progress(state_dir, scan_id, "--phase", "validation") updated_at = str(updated["scan"]["progress"]["updatedAt"]) assert updated_at > started_at time.sleep(0.001) - failed = run_workbench(state_dir, "fail-scan", "--scan-id", scan_id, "--message", "Stopped.") + failed = fail_scan(state_dir, scan_id, "Stopped.") assert str(failed["scan"]["progress"]["updatedAt"]) > updated_at diff --git a/plugins/codex-security/tests/test_workbench_handoff.py b/plugins/codex-security/tests/test_workbench_handoff.py index f641187f1e..db5e274ef2 100644 --- a/plugins/codex-security/tests/test_workbench_handoff.py +++ b/plugins/codex-security/tests/test_workbench_handoff.py @@ -6,7 +6,14 @@ from pathlib import Path import pytest -from workbench_test_support import create_saved_workspace, run_workbench +from workbench_test_support import ( + attach_continuation, + create_saved_workspace, + mark_handoff_delivered, + run_workbench, + scan_claim_command, + start_scan_command, +) @pytest.mark.parametrize("mode", ("standard", "deep")) @@ -399,72 +406,29 @@ def test_workbench_attaches_one_continuation_thread_to_claimed_scan( target = tmp_path / "target" target.mkdir() saved = create_saved_workspace(state_dir, target) - started = run_workbench(state_dir, "start-scan", "--workspace-id", str(saved["id"])) + started = start_scan_command(state_dir, str(saved["id"])) scan_id = str(started["results"]["scanId"]) claim_token = str(uuid.uuid4()) - run_workbench( - state_dir, "claim-handoff-delivery", "--scan-id", scan_id, "--claim-token", claim_token - ) + scan_claim_command(state_dir, "claim-handoff-delivery", scan_id, claim_token) - attached = run_workbench( - state_dir, - "attach-scan-continuation-thread", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - "continuation-thread", - ) + attached = attach_continuation(state_dir, scan_id, claim_token, "continuation-thread") assert attached["results"]["continuationThreadId"] == "continuation-thread" - delivered = run_workbench( - state_dir, - "mark-handoff-delivered", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - "continuation-thread", + delivered = mark_handoff_delivered( + state_dir, scan_id, claim_token, "--thread-id", "continuation-thread" ) assert delivered["results"]["handoffStatus"] == "delivered" - replayed = run_workbench( - state_dir, - "attach-scan-continuation-thread", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - "continuation-thread", - ) + replayed = attach_continuation(state_dir, scan_id, claim_token, "continuation-thread") assert replayed["results"]["continuationThreadId"] == "continuation-thread" - wrong_token = run_workbench( - state_dir, - "attach-scan-continuation-thread", - "--scan-id", - scan_id, - "--claim-token", - str(uuid.uuid4()), - "--thread-id", - "continuation-thread", - check=False, + wrong_token = attach_continuation( + state_dir, scan_id, str(uuid.uuid4()), "continuation-thread", check=False ) assert "claim token" in str(wrong_token["stderr"]) - different_thread = run_workbench( - state_dir, - "attach-scan-continuation-thread", - "--scan-id", - scan_id, - "--claim-token", - claim_token, - "--thread-id", - "different-thread", - check=False, + different_thread = attach_continuation( + state_dir, scan_id, claim_token, "different-thread", check=False ) assert "another continuation" in str(different_thread["stderr"]) diff --git a/plugins/codex-security/tests/test_workbench_progress.py b/plugins/codex-security/tests/test_workbench_progress.py index 111defa063..94e646714c 100644 --- a/plugins/codex-security/tests/test_workbench_progress.py +++ b/plugins/codex-security/tests/test_workbench_progress.py @@ -1,7 +1,11 @@ import json from pathlib import Path -from workbench_test_support import create_saved_workspace, run_workbench, start_delivered_scan +from workbench_test_support import ( + create_saved_workspace, + start_delivered_scan, + update_progress, +) def test_validation_clears_discovery_finding_count(tmp_path: Path) -> None: @@ -12,26 +16,12 @@ def test_validation_clears_discovery_finding_count(tmp_path: Path) -> None: started = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"])) scan_id = str(started["results"]["scanId"]) - discovery = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase", - "discovery", - "--reportable-findings-count", - "8", + discovery = update_progress( + state_dir, scan_id, "--phase", "discovery", "--reportable-findings-count", "8" ) assert discovery["scan"]["progress"]["candidates"] == {"reportable": 8} - validation = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase", - "validation", - ) + validation = update_progress(state_dir, scan_id, "--phase", "validation") assert validation["scan"]["progress"]["candidates"] == {"reportable": 0} @@ -43,10 +33,8 @@ def test_phase_progress_tracks_and_resets_phase_specific_receipts(tmp_path: Path started = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"])) scan_id = str(started["results"]["scanId"]) - discovery = run_workbench( + discovery = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase", "discovery", @@ -63,24 +51,15 @@ def test_phase_progress_tracks_and_resets_phase_specific_receipts(tmp_path: Path "unit": "review_receipts", } - validation = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase", - "validation", - ) + validation = update_progress(state_dir, scan_id, "--phase", "validation") assert validation["scan"]["progress"]["phaseProgress"] == { "completed": 0, "total": 0, "unit": None, } - validation_progress = run_workbench( + validation_progress = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase-items-total", "3", @@ -103,10 +82,8 @@ def test_phase_progress_rejects_regression_within_one_phase(tmp_path: Path) -> N saved = create_saved_workspace(state_dir, target) started = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"])) scan_id = str(started["results"]["scanId"]) - run_workbench( + update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase-items-total", "3", @@ -116,15 +93,7 @@ def test_phase_progress_rejects_regression_within_one_phase(tmp_path: Path) -> N "checks", ) - regressed = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase-items-completed", - "1", - check=False, - ) + regressed = update_progress(state_dir, scan_id, "--phase-items-completed", "1", check=False) assert regressed["returncode"] != 0 assert "Completed phase items cannot decrease" in str(regressed["stderr"]) @@ -142,10 +111,8 @@ def test_preflight_issues_replace_and_remain_visible_after_preflight(tmp_path: P "severity": "block", "status": "fail", } - blocked = run_workbench( + blocked = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase-items-total", "4", @@ -159,14 +126,7 @@ def test_preflight_issues_replace_and_remain_visible_after_preflight(tmp_path: P assert blocked["scan"]["progress"]["preflightIssues"] == [blocked_issue] assert blocked["scan"]["progress"]["preflightProgress"] == {"completed": 4, "total": 4} - clean = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--preflight-issues-json", - "[]", - ) + clean = update_progress(state_dir, scan_id, "--preflight-issues-json", "[]") assert clean["scan"]["progress"]["preflightIssues"] == [] warning_issue = { @@ -175,10 +135,8 @@ def test_preflight_issues_replace_and_remain_visible_after_preflight(tmp_path: P "severity": "warn", "status": "fail", } - ready = run_workbench( + ready = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase-items-total", "4", @@ -192,26 +150,11 @@ def test_preflight_issues_replace_and_remain_visible_after_preflight(tmp_path: P assert ready["scan"]["progress"]["preflightIssues"] == [warning_issue] assert ready["scan"]["progress"]["preflightProgress"] == {"completed": 4, "total": 4} - advanced = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--phase", - "threat_model", - ) + advanced = update_progress(state_dir, scan_id, "--phase", "threat_model") assert advanced["scan"]["progress"]["preflightIssues"] == [warning_issue] assert advanced["scan"]["progress"]["preflightProgress"] == {"completed": 4, "total": 4} - rejected = run_workbench( - state_dir, - "update-progress", - "--scan-id", - scan_id, - "--preflight-issues-json", - "[]", - check=False, - ) + rejected = update_progress(state_dir, scan_id, "--preflight-issues-json", "[]", check=False) assert rejected["returncode"] != 0 assert "only be updated during preflight" in str(rejected["stderr"]) @@ -230,10 +173,8 @@ def test_preflight_unknown_check_completes_only_after_clean_rerun(tmp_path: Path "status": "unknown", } - incomplete = run_workbench( + incomplete = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase-items-total", "4", @@ -250,10 +191,8 @@ def test_preflight_unknown_check_completes_only_after_clean_rerun(tmp_path: Path } assert incomplete["scan"]["progress"]["preflightIssues"] == [unknown_issue] - resolved = run_workbench( + resolved = update_progress( state_dir, - "update-progress", - "--scan-id", scan_id, "--phase-items-total", "4", @@ -277,10 +216,8 @@ def test_preflight_issues_reject_non_displayable_severity(tmp_path: Path) -> Non target.mkdir() saved = create_saved_workspace(state_dir, target) started = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"])) - rejected = run_workbench( + rejected = update_progress( state_dir, - "update-progress", - "--scan-id", str(started["results"]["scanId"]), "--preflight-issues-json", json.dumps( diff --git a/plugins/codex-security/tests/workbench_test_support.py b/plugins/codex-security/tests/workbench_test_support.py index 8bfb6cf56b..bbbb46bd76 100644 --- a/plugins/codex-security/tests/workbench_test_support.py +++ b/plugins/codex-security/tests/workbench_test_support.py @@ -47,6 +47,15 @@ def write_checkpoint(checkpoint_dir: Path, payload: Any) -> Path: return checkpoint_path +def saved_coverage(*, deferred=(), surfaces=(), completeness=None): + return { + "completeness": completeness or ("partial" if deferred else "complete"), + "surfaces": list(surfaces), + "explicitExclusions": [], + "deferred": list(deferred), + } + + def saved_draft( scan_id: str, *, @@ -62,10 +71,7 @@ def saved_draft( "complete": complete, "findings": list(findings), "coverage": { - "completeness": completeness or ("partial" if deferred else "complete"), - "surfaces": list(surfaces), - "explicitExclusions": [], - "deferred": list(deferred), + **saved_coverage(deferred=deferred, surfaces=surfaces, completeness=completeness), **({"resolvedDeferred": list(closures)} if closures else {}), }, } @@ -180,11 +186,136 @@ def run_workbench( return json.loads(completed.stdout) -def fail_deep_scan(state_dir, codex_home, scan_id, *, message="Worker stopped.", deep_status=None): +def set_triage( + state_dir: Path, occurrence_id: str, status: str, *extra: str, **options: Any +) -> dict[str, object]: return run_workbench( + state_dir, + "set-finding-triage", + "--occurrence-id", + occurrence_id, + "--status", + status, + *extra, + **options, + ) + + +def mark_handoff_delivered( + state_dir: Path, scan_id: str, claim_token: str, *extra: str, **options: Any +) -> dict[str, object]: + return scan_claim_command( + state_dir, "mark-handoff-delivered", scan_id, claim_token, *extra, **options + ) + + +def attach_continuation( + state_dir: Path, scan_id: str, claim_token: str, thread_id: str, **options: Any +) -> dict[str, object]: + return scan_claim_command( + state_dir, + "attach-scan-continuation-thread", + scan_id, + claim_token, + "--thread-id", + thread_id, + **options, + ) + + +def cancel_scan(state_dir: Path, scan_id: str, thread_id: str, **options: Any) -> dict[str, object]: + return scan_command(state_dir, "cancel-scan", scan_id, "--thread-id", thread_id, **options) + + +def start_scan_command( + state_dir: Path, workspace_id: str, *extra: str, **options: Any +) -> dict[str, object]: + return workspace_command(state_dir, "start-scan", workspace_id, *extra, **options) + + +def get_scan(state_dir: Path, scan_id: str, *extra: str, **options: Any) -> dict[str, object]: + return scan_command(state_dir, "get-scan", scan_id, *extra, **options) + + +def fail_scan( + state_dir: Path, scan_id: str, message: str, *extra: str, **options: Any +) -> dict[str, object]: + return scan_command(state_dir, "fail-scan", scan_id, "--message", message, *extra, **options) + + +def scan_command( + state_dir: Path, command: str, scan_id: str, *extra: str, **options: Any +) -> dict[str, object]: + return run_workbench(state_dir, command, "--scan-id", scan_id, *extra, **options) + + +def scan_claim_command( + state_dir: Path, command: str, scan_id: str, claim_token: str, *extra: str, **options: Any +) -> dict[str, object]: + return run_workbench( + state_dir, command, "--scan-id", scan_id, "--claim-token", claim_token, *extra, **options + ) + + +def workspace_command( + state_dir: Path, command: str, workspace_id: str, *extra: str, **options: Any +) -> dict[str, object]: + return run_workbench(state_dir, command, "--workspace-id", workspace_id, *extra, **options) + + +def save_workspace( + state_dir: Path, + workspace_id: str, + target_path: str, + scope: str, + mode: str, + *extra: str, + check: bool = True, + environment: dict[str, str] | None = None, +) -> dict[str, object]: + return workspace_command( + state_dir, + "save-workspace", + workspace_id, + "--target-path", + target_path, + "--scope", + scope, + "--mode", + mode, + *extra, + check=check, + environment=environment, + ) + + +def create_workspace( + state_dir: Path, + workspace_id: str, + *extra: str, + environment: dict[str, str] | None = None, +) -> dict[str, object]: + return workspace_command( + state_dir, "create-workspace", workspace_id, *extra, environment=environment + ) + + +def update_progress( + state_dir: Path, + scan_id: str, + *extra: str, + check: bool = True, + environment: dict[str, str] | None = None, +) -> dict[str, object]: + return scan_command( + state_dir, "update-progress", scan_id, *extra, check=check, environment=environment + ) + + +def fail_deep_scan(state_dir, codex_home, scan_id, *, message="Worker stopped.", deep_status=None): + return scan_command( state_dir, "fail-deep-scan", - "--scan-id", scan_id, "--message", message, @@ -263,10 +394,8 @@ def create_saved_workspace( state_dir: Path, target: Path, *, thread_id: str | None = None, mode: str = "standard" ) -> dict[str, object]: workspace_id = str(uuid.uuid4()) - created = run_workbench( + created = create_workspace( state_dir, - "create-workspace", - "--workspace-id", workspace_id, *(["--thread-id", thread_id] if thread_id else []), "--target-path", @@ -283,16 +412,11 @@ def create_saved_workspace( "isWorktree": False, "reviewChangesSupported": False, } - return run_workbench( + return save_workspace( state_dir, - "save-workspace", - "--workspace-id", workspace_id, - "--target-path", str(target), - "--scope", ".", - "--mode", mode, "--user-context", "Pay attention to uploaded archives.", @@ -303,26 +427,8 @@ def create_saved_git_workspace( state_dir: Path, target: Path, *, mode: str = "standard" ) -> dict[str, object]: workspace_id = str(uuid.uuid4()) - run_workbench( - state_dir, - "create-workspace", - "--workspace-id", - workspace_id, - "--target-path", - str(target), - ) - return run_workbench( - state_dir, - "save-workspace", - "--workspace-id", - workspace_id, - "--target-path", - str(target), - "--scope", - ".", - "--mode", - mode, - ) + create_workspace(state_dir, workspace_id, "--target-path", str(target)) + return save_workspace(state_dir, workspace_id, str(target), ".", mode) def worker_paths(scan_dir: Path, name: str) -> tuple[Path, Path, Path]: