from __future__ import annotations import dataclasses from pathlib import Path import unittest from ai_triage.evidence import ( FailureGate, evaluate_failure_gate, extract_failure_evidence, select_primary_failed_steps, ) from ai_triage.models import FailureCategory FIXTURES = Path(__file__).parent / "fixtures" class FailureGateTests(unittest.TestCase): def test_allows_only_the_exact_scenario_step_for_ai_classification(self) -> None: allowed = evaluate_failure_gate("Run scenario cases") similarly_named = evaluate_failure_gate("Run scenario cases summary") setup = evaluate_failure_gate("Install AWS cli") self.assertEqual(allowed, FailureGate(ai_eligible=True, forced_category=None)) self.assertFalse(similarly_named.ai_eligible) self.assertEqual(similarly_named.forced_category, FailureCategory.INFRASTRUCTURE_ERROR) self.assertFalse(setup.ai_eligible) self.assertEqual(setup.forced_category, FailureCategory.INFRASTRUCTURE_ERROR) def test_custom_allowed_steps_remain_exact_matches(self) -> None: result = evaluate_failure_gate("Execute essential cases", ("Execute essential cases",)) self.assertTrue(result.ai_eligible) self.assertIsNone(result.forced_category) def test_deduplicates_steps_and_suppresses_summary_after_primary_failure(self) -> None: steps = [ {"name": "Run scenario cases", "conclusion": "failure", "number": 6}, {"name": "Run scenario cases", "conclusion": "failure", "number": 6}, {"name": "test-summary", "conclusion": "failure", "number": 7}, {"name": "Set Test Result", "conclusion": "success", "number": 8}, ] result = select_primary_failed_steps(steps) self.assertEqual(result, (steps[0],)) def test_keeps_a_summary_failure_when_it_is_the_only_failure(self) -> None: step = {"name": "test-summary", "conclusion": "failure", "number": 7} self.assertEqual(select_primary_failed_steps([step]), (step,)) def test_keeps_non_success_terminal_step_for_infrastructure_evidence(self) -> None: step = {"name": "Run scenario cases", "conclusion": "timed_out", "number": 7} self.assertEqual(select_primary_failed_steps([step]), (step,)) class ExtractFailureEvidenceTests(unittest.TestCase): def test_extracts_final_retry_and_redacts_every_log_derived_field(self) -> None: raw = (FIXTURES / "scenario_failure.log").read_text(encoding="utf-8") evidence = extract_failure_evidence( raw, job_id=101, job_name="nonprod (sce_cluster_get.yaml)", step_name="Run scenario cases", max_output_chars=800, ) self.assertEqual(evidence.scenario_yaml, "sce_cluster_get.yaml") self.assertEqual(evidence.case_id, "case-cluster-get") self.assertEqual(evidence.operation, "TidbService_GetTidb") self.assertEqual(evidence.order, 2.0) self.assertEqual( evidence.request_url, "https://dedicated.staging.tidbapi.com/v1beta2/tidbs/cluster-1?api_key=%5BREDACTED%5D&page=1", ) self.assertEqual(evidence.request_method, "GET") self.assertIn('"displayName":"fixture"', evidence.request_body) self.assertNotIn("body-secret", evidence.request_body) self.assertIn("status:[200]", evidence.expected) self.assertEqual(evidence.actual_status, 500) self.assertIn("final failure", evidence.response_body) self.assertNotIn("owner@example.com", evidence.response_body) self.assertEqual(evidence.trace_id, "trace-final") self.assertIn("All attempts fail", evidence.error) self.assertNotIn("RATE_LIMITED", evidence.response_body) self.assertEqual(evidence.evidence_gaps, ()) flattened = "\n".join( value for field in dataclasses.fields(evidence) if isinstance((value := getattr(evidence, field.name)), str) ) for secret in ( "super-secret", "qa@example.com", "query-secret", "body-secret", "cookie-secret", "owner@example.com", ): self.assertNotIn(secret, flattened) for field in dataclasses.fields(evidence): value = getattr(evidence, field.name) if isinstance(value, str): self.assertLessEqual(len(value), 800) def test_records_missing_fields_instead_of_inventing_values(self) -> None: raw = "DEBUG: do step, case:case-1, operation:Svc_Op, order:1, request_type:http, saved_placeholder:map[token:secret]" evidence = extract_failure_evidence( raw, job_id=102, job_name="nonprod (sce_missing.yaml)", step_name="Run scenario cases", max_output_chars=200, ) self.assertIsNone(evidence.request_url) self.assertIsNone(evidence.actual_status) self.assertIsNone(evidence.response_body) self.assertIn("request_url", evidence.evidence_gaps) self.assertIn("actual_status", evidence.evidence_gaps) self.assertIn("response_body", evidence.evidence_gaps) self.assertNotIn("secret", evidence.error or "") def test_signal_interrupt_is_not_overwritten_by_later_cleanup_warnings( self, ) -> None: raw = """ DEBUG: do step, case:case-1, operation:ExportService_ListExports, order:2, request_type:http, saved_placeholder:map[] DEBUG: http request, url:https://api.example.invalid/v1beta2/tidbs/1/exports, method:GET, request_body:{}, resp_header:map[], resp_body:{"state":"RUNNING"}, status_code:200 signal: interrupt ##[error]The operation was canceled. Complete job ##[warning]Node.js 20 is deprecated. """ evidence = extract_failure_evidence( raw, job_id=104, job_name="nonprod (ng_cluster_exports_azure.yaml)", step_name="Run scenario cases", max_output_chars=500, ) self.assertIn("signal: interrupt", evidence.error or "") self.assertIn("operation was canceled", (evidence.error or "").lower()) self.assertNotIn("Node.js", evidence.error or "") def test_extracts_dynamic_related_resource_ids_from_failure_context(self) -> None: raw = """ DEBUG: do step, case:case-1, operation:Svc_Op, order:1, request_type:http DEBUG: http request, url:https://api.example.invalid/api/v1beta/projects/project-path/clusters/cluster-path/backups/backup-path?orgId=org-query&tidbIds%5B%5D=tidb-1&tidbIds%5B%5D=tidb-2&traceId=trace-private, method:POST, request_body:{"resourcePoolId":"pool-body","restoreId":"restore-body","importId":"import-body","accessKeyId":"access-secret"}, resp_header:map[], resp_body:{"backupId":"backup-response","changefeedId":"changefeed-response","eventId":"internal-event"}, status_code:500 AZURE_CLIENT_ID:*** OPEN_API_DEDICATED_EXECUTION_ID:execution-internal OPEN_API_DEDICATED_PROJECT_ID:project-path iam_aws_key_id:*** do action failed, scenario:sce_metadata.yaml, case:case-1, order:1 """ evidence = extract_failure_evidence( raw, job_id=105, job_name="nonprod (sce_metadata.yaml)", step_name="Run scenario cases", max_output_chars=500, ) self.assertEqual( { "projectId": ("project-path",), "clusterId": ("cluster-path",), "backupId": ("backup-path", "backup-response"), "orgId": ("org-query",), "tidbIds": ("tidb-1", "tidb-2"), "resourcePoolId": ("pool-body",), "restoreId": ("restore-body",), "importId": ("import-body",), "changefeedId": ("changefeed-response",), }, evidence.related_metadata, ) self.assertNotIn("traceId", evidence.related_metadata) self.assertNotIn("eventId", evidence.related_metadata) self.assertNotIn("accessKeyId", evidence.related_metadata) self.assertNotIn("AZURE_CLIENT_ID", evidence.related_metadata) self.assertNotIn( "OPEN_API_DEDICATED_EXECUTION_ID", evidence.related_metadata, ) self.assertNotIn( "OPEN_API_DEDICATED_PROJECT_ID", evidence.related_metadata, ) self.assertNotIn("iam_aws_key_id", evidence.related_metadata) self.assertNotIn( "***", { item for values in evidence.related_metadata.values() for item in values }, ) def test_extracts_unlisted_id_keys_and_merges_key_spelling_variants(self) -> None: raw = """ DEBUG: do step, case:case-1, operation:Svc_Op, order:1, request_type:http branchId:branch-1 resource_group_id=group-1 clusterID=cluster-1 cluster_id=cluster-2 do action failed, scenario:sce_metadata.yaml, case:case-1, order:1 """ evidence = extract_failure_evidence( raw, job_id=106, job_name="nonprod (sce_metadata.yaml)", step_name="Run scenario cases", max_output_chars=500, ) self.assertEqual(("branch-1",), evidence.related_metadata["branchId"]) self.assertEqual( ("group-1",), evidence.related_metadata["resource_group_id"], ) self.assertEqual( ("cluster-1", "cluster-2"), evidence.related_metadata["clusterID"], ) def test_merges_tenant_id_into_org_id(self) -> None: raw = """ DEBUG: do step, case:case-1, operation:Svc_Op, order:1, request_type:http org_id=org-1 tenant_id=org-1 tenantId=org-1 OPEN_API_DEDICATED_TENANT_ID=org-1 do action failed, scenario:sce_metadata.yaml, case:case-1, order:1 """ evidence = extract_failure_evidence( raw, job_id=107, job_name="nonprod (sce_metadata.yaml)", step_name="Run scenario cases", max_output_chars=500, ) self.assertEqual({"org_id": ("org-1",)}, evidence.related_metadata) def test_non_scenario_log_is_bounded_and_cannot_bypass_infrastructure_gate(self) -> None: raw = (FIXTURES / "infra_failure.log").read_text(encoding="utf-8") gate = evaluate_failure_gate("Install AWS cli") evidence = extract_failure_evidence( raw, job_id=103, job_name="nonprod (sce_infra.yaml)", step_name="Install AWS cli", max_output_chars=120, ) self.assertFalse(gate.ai_eligible) self.assertEqual(gate.forced_category, FailureCategory.INFRASTRUCTURE_ERROR) self.assertNotIn("infra-secret", evidence.error or "") self.assertLessEqual(len(evidence.error or ""), 120) self.assertIsNone(evidence.request_url) if __name__ == "__main__": unittest.main()