diff --git a/scripts/diagnose_cached_stage_failure.py b/scripts/diagnose_cached_stage_failure.py index bce66f8..fd3d9fa 100644 --- a/scripts/diagnose_cached_stage_failure.py +++ b/scripts/diagnose_cached_stage_failure.py @@ -310,6 +310,17 @@ def _audit_message_terms(texts: list[str], *, available: bool) -> dict[str, bool return {term: bool(_AUDIT_MESSAGE_TERM_PATTERNS[term].search(joined)) for term in _AUDIT_MESSAGE_TERMS} +def _tag_length_details(texts: list[str]) -> dict[str, Any]: + messages = [text for text in texts if _REASON_PATTERNS["traffic_tag_length"].search(text)] + text = "\n".join(messages) + limits = sorted({int(value) for value in re.findall(r"\b([0-9]{1,2})\s+characters?\b", text, re.I) if 0 < int(value) <= 63}) + return { + "character_limits": limits, + "minimum_requirement": bool(re.search(r"at least|minimum|too short|shorter", text, re.I)), + "maximum_requirement": bool(re.search(r"at most|maximum|too long|longer|exceed", text, re.I)), + } + + def summarize( *, service: Any, @@ -406,6 +417,7 @@ def summarize( audit_texts, available=audit_status == "ok" and audit_entries_count is not None, ), + "traffic_tag_length_details": _tag_length_details(audit_texts) if audit_status == "ok" else None, **_control_summary(policy, jobs), "audit_query_status": audit_status, "audit_entry_count": audit_entries_count if audit_status == "ok" else None, diff --git a/tests/test_cached_stage_failure_diagnostic.py b/tests/test_cached_stage_failure_diagnostic.py index 3d7dcde..428475c 100644 --- a/tests/test_cached_stage_failure_diagnostic.py +++ b/tests/test_cached_stage_failure_diagnostic.py @@ -101,6 +101,7 @@ def test_summary_exposes_only_closed_statuses_counts_and_categories(): "latest_created_revision_ready": False, "latest_revision_error_categories": ["image"], "latest_revision_error_status_codes": [], + "traffic_tag_length_details": {"character_limits": [], "minimum_requirement": False, "maximum_requirement": False}, "audit_error_message_terms": { "container": False, "name": False, @@ -172,6 +173,17 @@ def test_traffic_tag_rejections_remain_closed(message, reason): assert "PRIVATE_TAG" not in json.dumps(summary) +def test_tag_limit_details_only_expose_small_explicit_character_limits(): + summary = diagnostic._tag_length_details([ + "Traffic tag [PRIVATE_TAG] must be at least 3 characters long", + "Traffic tag [PRIVATE_TAG] must be at most 63 characters long", + "Traffic tag [PRIVATE_TAG] has 12345678 characters", + "Other error requires 8 characters", + ]) + assert summary == {"character_limits": [3, 63], "minimum_requirement": True, "maximum_requirement": True} + assert "PRIVATE_TAG" not in json.dumps(summary) + + def test_missing_logging_permission_is_reported_without_guessing_failure_category(): service, revisions, policy, jobs, audit = _evidence("permission denied") summary = diagnostic.summarize(