Skip to content

test(workbench): share setup for workspaces and saved scans #1400

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mldangelo-oai merged 1 commit into from
Oct 7, 2026
Merged

test(workbench): share setup for workspaces and saved scans #1400

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 2 additions & 6 deletions plugins/codex-security/tests/test_scan_contract_examples.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)

Expand Down
156 changes: 35 additions & 121 deletions plugins/codex-security/tests/test_workbench_cancellation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand All @@ -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"


Expand All @@ -58,84 +46,46 @@ 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"

restarted = start_delivered_scan(state_dir, "--workspace-id", str(saved["id"]))
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"])

Expand All @@ -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"])
Expand All @@ -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"
Expand Down
Loading
Loading