"""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 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", } 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.""" def _drifting(*, request, apply, expected_candidate_set_fingerprint=None): return { "outcome": ( allocator_service.OUTCOME_ASSIGNED if apply else allocator_service.OUTCOME_PREVIEW ), "selected": _selection(number=999 if apply else 643), "assignment": {"assignment_id": "asn-wrong"} if apply else None, "candidate_set_fingerprint": "fp-test", } result = request_service.apply_request( _request(number=643), principal=_principal(console_authz.OPERATOR), confirm=True, allocator=_drifting, claims_source=_no_claims, ) self.assertFalse(result["ok"]) self.assertIsNone(result["assignment"]) self.assertFalse(result["mutation_performed"]) 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 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()