"""Request preview, authorization, and workflow initiation tests (#643).
Covers each acceptance criterion:
* AC1 — preview shows authorize/deny with reasons.
* AC2 — apply creates an exclusive assignment or returns wait/blocked.
* AC3 — duplicate assign rejected.
* AC4 — preview / apply / deny / collision are all exercised.
* AC5 — the UI never renders a secret, and messaging stays brief.
Required tests named in the issue: allocator integration with fakes, and
gated-action tests. The allocator is injected as a fake throughout so no test
touches Gitea or reserves real work; one class asserts the *real* default
allocator refuses an incomplete inventory rather than ranking a partial set.
"""
from __future__ import annotations
import contextlib
import json
import os
import pathlib
import sys
import tempfile
import unittest
from typing import Any
from unittest import mock
from tests.webui_testclient import TestClient
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
import allocator_service # noqa: E402
from webui import console_audit, console_authz, request_service # noqa: E402
from webui.app import create_app # noqa: E402
from webui.console_redaction import scan_for_secrets # noqa: E402
from webui.request_views import render_requests_page # noqa: E402
EXEC_FLAG = "WEBUI_REQUESTS_EXECUTION"
SCOPE = {
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
}
def _principal(role: str) -> console_authz.Principal:
return console_authz.Principal(
subject=f"{role}@example.com",
role=role,
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
authenticated=True,
)
def _request(
*,
role: str = "author",
kind: str = "issue",
number: int = 643,
intent: str = "implement request preview and initiation",
head: str | None = None,
) -> request_service.WorkRequest:
parsed, error = request_service.parse_request(
{
"desired_role": role,
"work_kind": kind,
"work_number": number,
"intent_summary": intent,
"expected_head_sha": head,
**SCOPE,
}
)
assert error is None, error
assert parsed is not None
return parsed
def _selection(
*, kind: str = "issue", number: int = 643, head_sha: str | None = None
) -> dict[str, Any]:
return {
"kind": kind,
"number": number,
"title": "Web Console: Requests, intent preview, authorization",
"head_sha": head_sha,
"selected_action": "implement",
"expected_role_next": "author",
}
@contextlib.contextmanager
def _patched_control_plane(
*,
release_sink: list[tuple[str, str]] | None = None,
release_error: Exception | None = None,
leases=None,
):
"""Stand in for ``control_plane_db`` so compensation paths are observable.
The service imports the module inside the function, so patching
``sys.modules`` is what intercepts it. No real DB is opened.
"""
module = mock.MagicMock()
db = mock.MagicMock()
def _release(lease_id, *, session_id):
if release_error is not None:
raise release_error
if release_sink is not None:
release_sink.append((lease_id, session_id))
db.release_lease.side_effect = _release
db.list_leases.side_effect = leases or (lambda **_kwargs: [])
module.ControlPlaneDB.return_value = db
with mock.patch.dict(sys.modules, {"control_plane_db": module}):
yield db
def _drifting_allocator(*, lease_id: str | None = "lease-wrong"):
"""An allocator that previews the requested unit but assigns another.
This is the #643 B1 race in miniature: the CAS fingerprint hashes only
``{kind, number}``, so a lease taken on the requested unit inside the window
leaves the fingerprint identical while the selection moves on.
"""
assignment: dict[str, Any] = {"assignment_id": "asn-wrong"}
if lease_id:
assignment["lease_id"] = lease_id
def _drifting(
*, request, apply, expected_candidate_set_fingerprint=None, session_id=None
):
return {
"outcome": (
allocator_service.OUTCOME_ASSIGNED
if apply
else allocator_service.OUTCOME_PREVIEW
),
"selected": _selection(number=999 if apply else 643),
"assignment": dict(assignment) if apply else None,
"candidate_set_fingerprint": "fp-test",
"session_id": session_id,
}
return _drifting
def _fake_allocator(
*,
selection: dict[str, Any] | None = None,
preview_outcome: str = allocator_service.OUTCOME_PREVIEW,
apply_outcome: str = allocator_service.OUTCOME_ASSIGNED,
assignment: dict[str, Any] | None = None,
calls: list[dict[str, Any]] | None = None,
):
"""Build an allocator double that records how it was called."""
chosen = selection if selection is not None else _selection()
made = (
assignment
if assignment is not None
else {
"assignment_id": "asn-test-0001",
"lease_id": "lease-test-0001",
"session_id": "webui-request-test",
"expected_head_sha": chosen.get("head_sha"),
}
)
def _allocator(*, request, apply, expected_candidate_set_fingerprint=None):
if calls is not None:
calls.append(
{
"apply": apply,
"role": request.desired_role,
"fingerprint": expected_candidate_set_fingerprint,
}
)
return {
"outcome": apply_outcome if apply else preview_outcome,
"selected": dict(chosen),
"reasons": ["fake allocator"],
"candidate_set_fingerprint": "fp-test",
"candidate_count": 3,
"inventory_complete": True,
"selection_policy": allocator_service.SELECTION_POLICY,
"substrate": "control_plane_db",
"assignment": dict(made) if apply else None,
}
return _allocator
def _no_claims(_request):
return {}
def _claimed(role: str = "author"):
def _source(request):
return {
request.work_key: {
"lease_id": "lease-foreign-9999",
"session_id": "prgs-author-999-foreign",
"role": role,
"expires_at": "2026-07-25T09:10:37Z",
}
}
return _source
class TestRequestParsing(unittest.TestCase):
"""The request model rejects rather than guesses."""
def test_valid_request_round_trips(self):
req = _request()
self.assertEqual(req.work_key, ("issue", 643))
self.assertEqual(req.display_ref, "#643")
self.assertEqual(req.to_dict()["desired_role"], "author")
def test_unknown_role_rejected(self):
parsed, error = request_service.parse_request(
{
"desired_role": "admin",
"work_kind": "issue",
"work_number": 1,
"intent_summary": "x",
**SCOPE,
}
)
self.assertIsNone(parsed)
self.assertEqual(error.reason_code, "unknown_role")
self.assertEqual(error.field_name, "desired_role")
def test_unknown_work_kind_rejected(self):
parsed, error = request_service.parse_request(
{
"desired_role": "author",
"work_kind": "branch",
"work_number": 1,
"intent_summary": "x",
**SCOPE,
}
)
self.assertIsNone(parsed)
self.assertEqual(error.reason_code, "unknown_work_kind")
def test_non_positive_number_rejected(self):
for value in (0, -3):
with self.subTest(value=value):
parsed, error = request_service.parse_request(
{
"desired_role": "author",
"work_kind": "issue",
"work_number": value,
"intent_summary": "x",
**SCOPE,
}
)
self.assertIsNone(parsed)
self.assertEqual(error.reason_code, "invalid_work_number")
def test_missing_intent_rejected(self):
parsed, error = request_service.parse_request(
{
"desired_role": "author",
"work_kind": "issue",
"work_number": 1,
**SCOPE,
}
)
self.assertIsNone(parsed)
self.assertEqual(error.reason_code, "missing_intent")
def test_intent_is_bounded(self):
req = _request(intent="x" * 5000)
self.assertEqual(len(req.intent_summary), request_service.MAX_INTENT_CHARS)
def test_unresolved_scope_rejected(self):
parsed, error = request_service.parse_request(
{
"desired_role": "author",
"work_kind": "issue",
"work_number": 1,
"intent_summary": "x",
}
)
self.assertIsNone(parsed)
self.assertEqual(error.reason_code, "scope_unresolved")
def test_default_scope_fills_missing_fields(self):
parsed, error = request_service.parse_request(
{
"desired_role": "author",
"work_kind": "issue",
"work_number": 7,
"intent_summary": "x",
},
default_scope=SCOPE,
)
self.assertIsNone(error)
self.assertEqual(parsed.repo, "Gitea-Tools")
class TestPreviewAuthorizeDeny(unittest.TestCase):
"""AC1 — preview shows authorize/deny with reasons."""
def test_authorized_preview_names_every_check(self):
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
self.assertTrue(preview.authorized)
self.assertEqual(
{c.name for c in preview.checks},
{
request_service.CHECK_AUTHORIZATION,
request_service.CHECK_CAPABILITY,
request_service.CHECK_LEASE_AVAILABILITY,
request_service.CHECK_NEXT_SAFE_ACTION,
request_service.CHECK_HEAD_PIN,
},
)
self.assertEqual(preview.required_profile, "prgs-author")
self.assertEqual(preview.required_namespace, "gitea-author")
def test_every_check_carries_a_reason(self):
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
for check in preview.checks:
with self.subTest(check=check.name):
self.assertTrue(check.reason_code.strip())
self.assertTrue(check.detail.strip())
def test_anonymous_preview_denied_with_reason(self):
preview = request_service.preview_request(
_request(),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
self.assertFalse(preview.authorized)
self.assertEqual(preview.reason_code, console_authz.DENY_UNAUTHENTICATED)
def test_viewer_preview_denied_for_insufficient_role(self):
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.VIEWER),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
self.assertFalse(preview.authorized)
self.assertEqual(preview.reason_code, console_authz.DENY_INSUFFICIENT_ROLE)
def test_denied_preview_never_reaches_the_allocator(self):
"""A denial must not double as a queue oracle."""
calls: list[dict[str, Any]] = []
request_service.preview_request(
_request(),
principal=_principal(console_authz.VIEWER),
allocator=_fake_allocator(calls=calls),
claims_source=_no_claims,
audit=False,
)
self.assertEqual(calls, [])
def test_preview_lists_prohibited_actions(self):
preview = request_service.preview_request(
_request(role="author"),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
self.assertIn("merge", preview.prohibited_actions)
self.assertIn("approve", preview.prohibited_actions)
def test_preview_reports_next_safe_action(self):
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
self.assertIn("issue #643", preview.next_safe_action)
def test_preview_never_mutates(self):
calls: list[dict[str, Any]] = []
request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(calls=calls),
claims_source=_no_claims,
audit=False,
)
self.assertEqual([c["apply"] for c in calls], [False])
class TestPreviewFailClosed(unittest.TestCase):
"""Missing evidence denies; it never reads as an absence of obstacles."""
def test_unreadable_claim_inventory_denies(self):
def _boom(_request):
raise RuntimeError("db unavailable")
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_boom,
audit=False,
)
self.assertFalse(preview.authorized)
self.assertEqual(
preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
)
def test_allocator_failure_denies(self):
def _boom(**_kwargs):
raise RuntimeError("allocator exploded")
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_boom,
claims_source=_no_claims,
audit=False,
)
self.assertFalse(preview.authorized)
self.assertEqual(
preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
)
def test_allocator_selecting_other_work_denies(self):
preview = request_service.preview_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(selection=_selection(number=999)),
claims_source=_no_claims,
audit=False,
)
self.assertFalse(preview.authorized)
self.assertEqual(preview.reason_code, request_service.REASON_NOT_NEXT_SAFE)
self.assertIn("#999", preview.detail)
def test_pr_without_head_sha_denies(self):
preview = request_service.preview_request(
_request(role="reviewer", kind="pr", number=898),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(
selection=_selection(kind="pr", number=898, head_sha=None)
),
claims_source=_no_claims,
audit=False,
)
self.assertFalse(preview.authorized)
self.assertEqual(
preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
)
def test_pr_head_moved_denies(self):
preview = request_service.preview_request(
_request(role="reviewer", kind="pr", number=898, head="a" * 40),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(
selection=_selection(kind="pr", number=898, head_sha="b" * 40)
),
claims_source=_no_claims,
audit=False,
)
self.assertFalse(preview.authorized)
self.assertEqual(preview.reason_code, "head_moved")
def test_pr_head_matching_passes(self):
preview = request_service.preview_request(
_request(role="reviewer", kind="pr", number=898, head="b" * 40),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(
selection=_selection(kind="pr", number=898, head_sha="b" * 40)
),
claims_source=_no_claims,
audit=False,
)
self.assertTrue(preview.authorized)
class TestApplyExecutionGate(unittest.TestCase):
"""Execution stays wired off unless an operator opts in explicitly."""
def test_action_is_registered_and_unwired_by_default(self):
action = console_authz.get_action(request_service.ACTION_ID)
self.assertIsNotNone(action)
self.assertEqual(action.phase, 2)
self.assertEqual(action.minimum_role, console_authz.OPERATOR)
self.assertTrue(action.requires_confirmation)
self.assertFalse(console_authz.execution_wired(action, env={}))
def test_flag_named_but_unset_does_not_wire(self):
action = console_authz.get_action(request_service.ACTION_ID)
self.assertFalse(console_authz.execution_wired(action, env={EXEC_FLAG: "no"}))
self.assertTrue(console_authz.execution_wired(action, env={EXEC_FLAG: "1"}))
def test_opting_in_wires_only_this_action(self):
env = {EXEC_FLAG: "1"}
for action_id, action in console_authz.ACTIONS.items():
with self.subTest(action=action_id):
self.assertEqual(
console_authz.execution_wired(action, env=env),
action_id == request_service.ACTION_ID,
)
def test_apply_denied_while_unwired(self):
with mock.patch.dict(os.environ, {EXEC_FLAG: ""}):
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_fake_allocator(),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertEqual(result["outcome"], request_service.OUTCOME_DENIED)
self.assertEqual(result["reason_code"], request_service.REASON_UNAUTHORIZED)
self.assertFalse(result["mutation_performed"])
class TestApplyOutcomes(unittest.TestCase):
"""AC2/AC3/AC4 — assignment, wait, blocked, and duplicate rejection."""
def setUp(self):
patcher = mock.patch.dict(os.environ, {EXEC_FLAG: "1"})
patcher.start()
self.addCleanup(patcher.stop)
def test_apply_creates_exclusive_assignment(self):
calls: list[dict[str, Any]] = []
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_fake_allocator(calls=calls),
claims_source=_no_claims,
)
self.assertTrue(result["ok"])
self.assertEqual(result["outcome"], allocator_service.OUTCOME_ASSIGNED)
self.assertEqual(result["assignment"]["assignment_id"], "asn-test-0001")
self.assertTrue(result["mutation_performed"])
self.assertEqual(result["status_code"], 201)
# Dry-run first, then apply — never apply alone.
self.assertEqual([c["apply"] for c in calls], [False, True])
# The apply call carries the fingerprint the dry-run produced.
self.assertEqual(calls[1]["fingerprint"], "fp-test")
def test_assignment_returns_a_role_handoff(self):
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_fake_allocator(),
claims_source=_no_claims,
)
handoff = result["handoff"]
self.assertEqual(handoff["required_profile"], "prgs-author")
self.assertEqual(handoff["required_namespace"], "gitea-author")
self.assertEqual(handoff["assignment_id"], "asn-test-0001")
self.assertIn("merge", handoff["forbidden_actions"])
def test_unconfirmed_apply_refuses_before_the_allocator(self):
calls: list[dict[str, Any]] = []
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.OPERATOR),
confirm=False,
allocator=_fake_allocator(calls=calls),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertEqual(
result["reason_code"], request_service.REASON_CONFIRMATION_REQUIRED
)
self.assertEqual(calls, [])
def test_duplicate_assignment_rejected(self):
"""AC3 — an active lease on the work unit blocks a second assign."""
calls: list[dict[str, Any]] = []
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_fake_allocator(calls=calls),
claims_source=_claimed(),
)
self.assertFalse(result["ok"])
self.assertEqual(result["outcome"], request_service.OUTCOME_BLOCKED)
self.assertEqual(
result["reason_code"], request_service.REASON_DUPLICATE_ASSIGNMENT
)
self.assertFalse(result["mutation_performed"])
# The dry-run ran; the apply never did.
self.assertEqual([c["apply"] for c in calls], [False])
def test_not_next_safe_work_returns_wait_without_applying(self):
calls: list[dict[str, Any]] = []
result = request_service.apply_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_fake_allocator(
selection=_selection(number=999), calls=calls
),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertEqual(result["outcome"], request_service.OUTCOME_WAIT)
self.assertEqual(result["reason_code"], request_service.REASON_NOT_NEXT_SAFE)
self.assertEqual([c["apply"] for c in calls], [False])
def test_allocator_declining_on_apply_returns_blocked(self):
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_fake_allocator(
apply_outcome=allocator_service.OUTCOME_BLOCKED_LEASE,
assignment={},
),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertEqual(result["outcome"], request_service.OUTCOME_BLOCKED)
self.assertEqual(
result["reason_code"], request_service.REASON_ALLOCATOR_OUTCOME
)
self.assertFalse(result["mutation_performed"])
def test_allocator_drift_on_apply_is_not_read_as_an_assignment(self):
"""The apply call must return *this* work unit, not a substitute.
Drift is not merely refused: the allocator has already committed the
substitute assignment by the time egress rejects it, so the refusal must
also release it. Asserting only ``mutation_performed is False`` would
pass just as well against a leak.
"""
released: list[tuple[str, str]] = []
with _patched_control_plane(release_sink=released):
result = request_service.apply_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_drifting_allocator(),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertIsNone(result["assignment"])
# The substitute assignment was released, so nothing durable survives.
self.assertEqual(len(released), 1)
self.assertEqual(released[0][0], "lease-wrong")
compensation = result["compensation"]
self.assertTrue(compensation["released"])
self.assertEqual(compensation["lease_id"], "lease-wrong")
self.assertEqual(compensation["assignment_id"], "asn-wrong")
self.assertEqual(compensation["selected"]["number"], 999)
self.assertFalse(result["mutation_performed"])
self.assertNotIn("orphaned_assignment", result)
def test_drift_whose_release_fails_reports_the_orphan_and_a_reclaim(self):
"""A release that fails must surface the leak, never swallow it."""
with _patched_control_plane(release_error=RuntimeError("db is read-only")):
result = request_service.apply_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_drifting_allocator(),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertIsNone(result["assignment"])
# A lease really is out there; saying "nothing changed" would be a lie.
self.assertTrue(result["mutation_performed"])
orphan = result["orphaned_assignment"]
self.assertFalse(orphan["released"])
self.assertTrue(orphan["attempted"])
self.assertIn("db is read-only", orphan["error"])
reclaim = orphan["reclaim_action"]
self.assertEqual(reclaim["tool"], "gitea_release_workflow_lease")
self.assertEqual(reclaim["lease_id"], "lease-wrong")
def test_drift_without_a_lease_id_still_surfaces_a_reclaim(self):
"""An assignment with no lease id cannot be released — say so."""
result = request_service.apply_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_drifting_allocator(lease_id=None),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertTrue(result["mutation_performed"])
orphan = result["orphaned_assignment"]
self.assertFalse(orphan["attempted"])
self.assertEqual(orphan["assignment_id"], "asn-wrong")
self.assertIn("reclaim_action", orphan)
def test_exception_after_commit_releases_the_session_lease(self):
"""A post-commit exception surfaces as no result — with a live lease.
``allocate_next_work`` catches only three exception types, so anything
else raised after ``assign_and_lease`` committed reaches the caller as
``None`` while the lease is durable. The stable per-flow session id is
what makes that lease findable.
"""
released: list[tuple[str, str]] = []
seen: list[str | None] = []
def _explodes_after_commit(
*, request, apply, expected_candidate_set_fingerprint=None, session_id=None
):
seen.append(session_id)
if not apply:
return {
"outcome": allocator_service.OUTCOME_PREVIEW,
"selected": _selection(number=643),
"assignment": None,
"candidate_set_fingerprint": "fp-test",
}
raise KeyError("selection['required_profile']")
def _leases(**_kwargs):
return [
{
"lease_id": "lease-committed",
"session_id": seen[-1],
"work_kind": "issue",
"work_number": 643,
}
]
with _patched_control_plane(release_sink=released, leases=_leases):
result = request_service.apply_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_explodes_after_commit,
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertEqual(
result["reason_code"], request_service.REASON_EVIDENCE_UNAVAILABLE
)
self.assertEqual(len(released), 1)
self.assertEqual(released[0][0], "lease-committed")
self.assertEqual(
result["compensation"]["released"][0]["lease_id"], "lease-committed"
)
self.assertFalse(result["mutation_performed"])
def test_one_session_id_spans_the_dry_run_and_the_apply(self):
"""Both halves of an apply share one control-plane identity."""
seen: list[str | None] = []
def _recording(
*, request, apply, expected_candidate_set_fingerprint=None, session_id=None
):
seen.append(session_id)
return {
"outcome": (
allocator_service.OUTCOME_ASSIGNED
if apply
else allocator_service.OUTCOME_PREVIEW
),
"selected": _selection(number=643),
"assignment": (
{
"assignment_id": "asn-ok",
"lease_id": "lease-ok",
"expected_head_sha": None,
}
if apply
else None
),
"candidate_set_fingerprint": "fp-test",
"session_id": session_id,
}
result = request_service.apply_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_recording,
claims_source=_no_claims,
)
self.assertTrue(result["ok"])
self.assertEqual(len(seen), 2)
self.assertTrue(all(s for s in seen))
self.assertEqual(seen[0], seen[1], "dry-run and apply must share one id")
self.assertTrue(seen[0].startswith("webui-request-"))
def test_viewer_cannot_apply(self):
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.VIEWER),
confirm=True,
allocator=_fake_allocator(),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertEqual(result["reason_code"], request_service.REASON_UNAUTHORIZED)
def test_anonymous_cannot_apply(self):
result = request_service.apply_request(
_request(),
confirm=True,
allocator=_fake_allocator(),
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertFalse(result["mutation_performed"])
class TestDefaultAllocatorPastTheEarlyReturns(unittest.TestCase):
"""``default_allocator`` beyond its two fail-closed guards (#643).
Both prior tests returned before ``ControlPlaneDB`` was ever constructed, so
the session id, the ``side_effect_free`` routing and the CAS round-trip had
no coverage at all — which is how a preview that writes session rows shipped.
"""
def setUp(self):
from webui.queue_loader import QueueSnapshot
self.snapshot = QueueSnapshot(
project_id="p",
repo_label="r",
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
)
@contextlib.contextmanager
def _harness(self):
"""Run the real ``default_allocator`` against a recorded allocator call."""
calls: dict[str, Any] = {}
def _allocate(db, **kwargs):
calls.update(kwargs)
return {"outcome": allocator_service.OUTCOME_PREVIEW}
module = mock.MagicMock()
with mock.patch.dict(sys.modules, {"control_plane_db": module}), \
mock.patch(
"webui.queue_loader.load_queue_snapshot",
return_value=self.snapshot,
), \
mock.patch(
"webui.traffic_loader.candidates_from_queue_snapshot",
return_value=[],
), \
mock.patch.object(
allocator_service, "allocate_next_work", side_effect=_allocate
):
yield calls
def test_preview_runs_side_effect_free_and_never_applies(self):
with self._harness() as calls:
result = request_service.default_allocator(
request=_request(), apply=False
)
self.assertIsNotNone(result)
self.assertTrue(calls["side_effect_free"])
self.assertFalse(calls["apply"])
def test_apply_is_not_side_effect_free(self):
with self._harness() as calls:
request_service.default_allocator(request=_request(), apply=True)
self.assertFalse(calls["side_effect_free"])
self.assertTrue(calls["apply"])
def test_caller_session_id_is_passed_through_verbatim(self):
with self._harness() as calls:
request_service.default_allocator(
request=_request(), apply=True, session_id="webui-request-fixed"
)
self.assertEqual(calls["session_id"], "webui-request-fixed")
def test_absent_session_id_is_minted_with_the_expected_shape(self):
with self._harness() as calls:
request_service.default_allocator(request=_request(), apply=False)
self.assertTrue(str(calls["session_id"]).startswith("webui-request-"))
def test_scope_and_fingerprint_reach_the_allocator(self):
with self._harness() as calls:
request_service.default_allocator(
request=_request(number=664),
apply=False,
expected_candidate_set_fingerprint="fp-pinned",
)
self.assertEqual(calls["expected_candidate_set_fingerprint"], "fp-pinned")
self.assertEqual(calls["remote"], SCOPE["remote"])
self.assertEqual(calls["org"], SCOPE["org"])
self.assertEqual(calls["repo"], SCOPE["repo"])
self.assertEqual(calls["allocation_mode"], "role_scoped")
class TestDefaultClaimsSource(unittest.TestCase):
"""``default_claims_source`` had no test at all (#643)."""
def test_claims_are_read_for_the_request_scope(self):
db = mock.MagicMock()
db.list_active_claims.return_value = {("issue", 643): {"lease_id": "l1"}}
module = mock.MagicMock()
module.ControlPlaneDB.return_value = db
with mock.patch.dict(sys.modules, {"control_plane_db": module}):
claims = request_service.default_claims_source(_request())
self.assertEqual(claims, {("issue", 643): {"lease_id": "l1"}})
db.list_active_claims.assert_called_once_with(
remote=SCOPE["remote"], org=SCOPE["org"], repo=SCOPE["repo"]
)
def test_an_unreadable_substrate_denies_rather_than_returning_empty(self):
module = mock.MagicMock()
module.ControlPlaneDB.side_effect = RuntimeError("no db")
with mock.patch.dict(sys.modules, {"control_plane_db": module}):
# _load_claims converts the failure into None, which fails the
# lease check closed; an empty mapping would read as "nothing
# claimed" and wrongly authorize.
claims = request_service._load_claims(
_request(), request_service.default_claims_source
)
self.assertIsNone(claims)
class TestAllocatorIntegrationFakes(unittest.TestCase):
"""The real default allocator refuses a partial inventory (#758)."""
def test_incomplete_inventory_returns_none(self):
from webui.queue_loader import PaginationMeta, QueueSnapshot
snapshot = QueueSnapshot(
project_id="p",
repo_label="r",
prs=(),
issues=(),
pr_pagination=PaginationMeta(
page=1,
per_page=50,
returned_count=50,
has_more=True,
is_final_page=False,
inventory_complete=False,
pages_fetched=1,
),
issue_pagination=None,
)
with mock.patch(
"webui.queue_loader.load_queue_snapshot", return_value=snapshot
):
result = request_service.default_allocator(
request=_request(), apply=False
)
self.assertIsNone(result)
def test_fetch_error_returns_none(self):
from webui.queue_loader import QueueSnapshot
snapshot = QueueSnapshot(
project_id="p",
repo_label="r",
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
fetch_error="no credentials",
)
with mock.patch(
"webui.queue_loader.load_queue_snapshot", return_value=snapshot
):
result = request_service.default_allocator(
request=_request(), apply=True
)
self.assertIsNone(result)
class TestAuditRecords(unittest.TestCase):
"""Every preview and apply is auditable, correlated, and redacted."""
def setUp(self):
handle = tempfile.NamedTemporaryFile(
mode="w", suffix=".jsonl", delete=False
)
handle.close()
self.sink = handle.name
self.addCleanup(
lambda: os.path.exists(self.sink) and os.remove(self.sink)
)
patcher = mock.patch.dict(
os.environ, {console_audit.AUDIT_LOG_ENV: self.sink}
)
patcher.start()
self.addCleanup(patcher.stop)
def _records(self) -> list[dict[str, Any]]:
with open(self.sink, encoding="utf-8") as handle:
return [json.loads(line) for line in handle if line.strip()]
def test_preview_is_audited_with_a_correlation_id(self):
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
)
records = self._records()
self.assertEqual(len(records), 1)
record = records[0]
self.assertEqual(record["action"], request_service.ACTION_ID)
self.assertEqual(record["result"], console_audit.RESULT_PREVIEWED)
self.assertEqual(
record["correlation"]["request_id"], preview.correlation_id
)
self.assertEqual(record["target"]["ref"], "#643")
def test_denied_apply_is_audited(self):
with mock.patch.dict(os.environ, {EXEC_FLAG: ""}):
request_service.apply_request(
_request(),
principal=_principal(console_authz.VIEWER),
confirm=True,
allocator=_fake_allocator(),
claims_source=_no_claims,
)
self.assertEqual(self._records()[-1]["result"], console_audit.RESULT_DENIED)
def test_assignment_is_audited_and_correlated(self):
with mock.patch.dict(os.environ, {EXEC_FLAG: "1"}):
result = request_service.apply_request(
_request(),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_fake_allocator(),
claims_source=_no_claims,
)
record = self._records()[-1]
self.assertEqual(record["result"], console_audit.RESULT_SUCCEEDED)
self.assertEqual(
record["correlation"]["request_id"], result["correlation_id"]
)
self.assertEqual(
record["metadata"]["assignment_id"],
result["assignment"]["assignment_id"],
)
def test_intent_bearing_a_secret_is_not_persisted_raw(self):
request_service.preview_request(
_request(intent="use token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
)
for record in self._records():
with self.subTest(event=record.get("event_id")):
self.assertFalse(scan_for_secrets(record))
class TestRequestRoutes(unittest.TestCase):
"""The HTTP surface: form page, preview API, apply API."""
def setUp(self):
self.client = TestClient(create_app())
def test_requests_page_renders_form(self):
response = self.client.get("/requests")
self.assertEqual(response.status_code, 200)
body = response.text
self.assertIn("Requests", body)
self.assertIn("desired_role", body)
self.assertIn("intent_summary", body)
def test_requests_page_is_linked_from_nav(self):
from webui.nav import nav_hrefs
self.assertIn("/requests", nav_hrefs())
def test_preview_api_rejects_an_invalid_request(self):
response = self.client.post(
"/api/v1/requests/preview",
json={
"desired_role": "wizard",
"work_kind": "issue",
"work_number": 1,
"intent_summary": "x",
**SCOPE,
},
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.json()["reason_code"], "unknown_role")
def test_preview_api_denies_anonymous(self):
response = self.client.post(
"/api/v1/requests/preview",
json={
"desired_role": "author",
"work_kind": "issue",
"work_number": 643,
"intent_summary": "x",
**SCOPE,
},
)
self.assertEqual(response.status_code, 403)
payload = response.json()
self.assertFalse(payload["authorized"])
self.assertFalse(payload["mutation_performed"])
def test_apply_api_denies_anonymous(self):
response = self.client.post(
"/api/v1/requests/apply",
json={
"desired_role": "author",
"work_kind": "issue",
"work_number": 643,
"intent_summary": "x",
"confirm": True,
**SCOPE,
},
)
self.assertEqual(response.status_code, 403)
payload = response.json()
self.assertFalse(payload["ok"])
self.assertFalse(payload["mutation_performed"])
self.assertIsNone(payload["assignment"])
def test_apply_api_rejects_an_invalid_request(self):
response = self.client.post(
"/api/v1/requests/apply",
json={
"desired_role": "author",
"work_kind": "issue",
"work_number": -1,
"intent_summary": "x",
**SCOPE,
},
)
self.assertEqual(response.status_code, 400)
def test_request_apis_are_post_only(self):
"""GET is not a way in. The app's 405 handler renders a read-only
method against a write route as 404, so that is what is asserted."""
for path in ("/api/v1/requests/preview", "/api/v1/requests/apply"):
with self.subTest(path=path):
self.assertEqual(self.client.get(path).status_code, 404)
def test_form_post_previews_and_never_assigns(self):
response = self.client.post(
"/requests",
data={
"desired_role": "author",
"work_kind": "issue",
"work_number": "643",
"intent_summary": "implement the request surface",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
},
)
self.assertEqual(response.status_code, 200)
self.assertIn("Intent preview", response.text)
class TestRenderingSafety(unittest.TestCase):
"""AC5 — the page escapes hostile input and shows no secret."""
def test_intent_is_escaped(self):
preview = request_service.preview_request(
_request(intent=""),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
html = render_requests_page(preview=preview)
self.assertNotIn("", html)
self.assertIn("<script>", html)
def test_page_renders_a_denial_without_a_preview(self):
_, error = request_service.parse_request(
{
"desired_role": "wizard",
"work_kind": "issue",
"work_number": 1,
"intent_summary": "x",
**SCOPE,
}
)
html = render_requests_page(error=error)
self.assertIn("Request rejected", html)
self.assertIn("unknown_role", html)
def test_page_shows_no_credential_material(self):
preview = request_service.preview_request(
_request(),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
html = render_requests_page(preview=preview)
for needle in ("token=", "Bearer ", "password"):
with self.subTest(needle=needle):
self.assertNotIn(needle, html)
if __name__ == "__main__": # pragma: no cover
unittest.main()