| #!/usr/bin/env python3 |
| # Copyright 2026 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| """Unit tests for inspect_coverage_build.py.""" |
| |
| # pylint: disable=line-too-long |
| |
| import contextlib |
| import io |
| import os |
| from pathlib import Path |
| import tempfile |
| import unittest |
| from unittest import mock |
| import sys |
| |
| import inspect_coverage_build |
| |
| |
| class InspectCoverageBuildTest(unittest.TestCase): |
| """Tests build ID parsing, status verification, config audit, and pipeline check.""" |
| |
| def setUp(self) -> None: |
| """Sets up test fixtures before each test method.""" |
| inspect_coverage_build._ALL_JSON_GZ_CACHE.clear() |
| self.addCleanup(inspect_coverage_build._ALL_JSON_GZ_CACHE.clear) |
| self.run_patcher = mock.patch('subprocess.run') |
| self.mock_run = self.run_patcher.start() |
| self.addCleanup(self.run_patcher.stop) |
| |
| def test_parse_build_id_numeric_id(self) -> None: |
| """Verifies numeric build ID strings are returned as-is.""" |
| self.assertEqual( |
| inspect_coverage_build.parse_build_id('8676997864980431489'), |
| '8676997864980431489', |
| ) |
| |
| def test_parse_build_id_ci_url(self) -> None: |
| """Verifies numeric ID extraction from ci.chromium.org build URLs.""" |
| url = 'https://ci.chromium.org/ui/p/chromium/builders/try/linux-rel/2773124/infra' |
| self.assertEqual( |
| inspect_coverage_build.parse_build_id(url), |
| 'chromium/try/linux-rel/2773124', |
| ) |
| |
| def test_parse_build_id_invalid_raises(self) -> None: |
| """Verifies malformed strings raise ValueError instead of passing through.""" |
| with self.assertRaises(ValueError): |
| inspect_coverage_build.parse_build_id('not-a-build-id-or-url') |
| |
| def test_check_build_status_terminal(self) -> None: |
| """Verifies terminal SUCCESS status contains terminal flag.""" |
| build_data = {'id': '12345', 'status': 'SUCCESS'} |
| result = inspect_coverage_build.check_build_status(build_data) |
| self.assertTrue(result['terminal']) |
| self.assertEqual(result['status'], 'SUCCESS') |
| |
| def test_check_build_status_running(self) -> None: |
| """Verifies non-terminal STARTED status returns terminal=False.""" |
| build_data = {'id': '12345', 'status': 'STARTED'} |
| result = inspect_coverage_build.check_build_status(build_data) |
| self.assertFalse(result['terminal']) |
| self.assertEqual(result['status'], 'STARTED') |
| |
| def test_verify_configuration_valid(self) -> None: |
| """Verifies configuration audit reports valid when gclient flags are set.""" |
| build_data = { |
| 'input': {'properties': {'checkout_clang_coverage_tools': True}}, |
| 'steps': [{'name': 'lookup GN args', 'status': 'SUCCESS'}], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'cpp') |
| self.assertTrue(result['configuration_valid']) |
| self.assertEqual(result['gclient_config_mismatches'], []) |
| |
| def test_verify_configuration_missing_clang_cov(self) -> None: |
| """Verifies config audit flags missing checkout_clang_coverage_tools property.""" |
| build_data = { |
| 'input': {'properties': {}}, |
| 'steps': [{'name': 'lookup GN args', 'status': 'FAILURE'}], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'cpp') |
| self.assertFalse(result['configuration_valid']) |
| self.assertIn( |
| 'Missing checkout_clang_coverage_tools in gclient config.', |
| result['gclient_config_mismatches'], |
| ) |
| |
| def test_verify_configuration_all_languages(self) -> None: |
| """Verifies all supported languages can be checked against config map.""" |
| build_data = { |
| 'input': {'properties': {}}, |
| 'steps': [{'name': 'lookup GN args', 'status': 'SUCCESS'}], |
| } |
| for lang in ['cpp', 'objc', 'rust', 'java', 'js', 'ts']: |
| res = inspect_coverage_build.verify_configuration(build_data, lang) |
| self.assertIn('configuration_valid', res) |
| |
| def test_verify_configuration_unsupported_language_raises(self) -> None: |
| """Verifies unmapped language strings raise ValueError.""" |
| build_data = {'input': {'properties': {}}, 'steps': []} |
| with self.assertRaises(ValueError): |
| inspect_coverage_build.verify_configuration( |
| build_data, 'unsupported-lang' |
| ) |
| |
| def test_verify_coverage_pipeline_success(self) -> None: |
| """Verifies Phase 2 pipeline audit reports success when HTML upload succeeds.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'process clang code coverage data for unit test coverage', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for unit test coverage|generate html report in 42 tests', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for unit test coverage|gsutil Upload coverage artifacts', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| 'output': { |
| 'properties': { |
| 'gsutil_urls': { |
| 'process clang code coverage data for unit test coverage|gsutil Upload coverage artifacts': 'gs://bucket/report' |
| } |
| } |
| }, |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| unit_res = result['pipelines_checked']['cpp_unit'] |
| self.assertTrue(unit_res['pipeline_success']) |
| self.assertEqual(unit_res['tests_processed_count'], 42) |
| self.assertFalse(unit_res['skipped_no_data']) |
| |
| def test_verify_coverage_pipeline_skipped_no_data(self) -> None: |
| """Verifies pipeline audit flags skipped_no_data and records substeps correctly.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'process clang code coverage data for unit test coverage', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for unit test coverage|merge all profile files into a single .profdata', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for unit test coverage|skip processing because no data is found', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| 'output': {'properties': {}}, |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| unit_res = result['pipelines_checked']['cpp_unit'] |
| self.assertFalse(unit_res['pipeline_success']) |
| self.assertTrue(unit_res['skipped_no_data']) |
| self.assertEqual(len(unit_res['child_steps']), 2) |
| |
| def test_run_bb_get_success(self) -> None: |
| """Verifies run_bb_get returns parsed JSON dictionary when bb succeeds.""" |
| self.mock_run.return_value = mock.MagicMock( |
| returncode=0, stdout='{"id": "123", "status": "SUCCESS"}' |
| ) |
| data = inspect_coverage_build.run_bb_get('123') |
| self.assertEqual(data['id'], '123') |
| self.assertEqual(data['status'], 'SUCCESS') |
| |
| def test_run_bb_get_failure_raises(self) -> None: |
| """Verifies run_bb_get raises RuntimeError when bb returns non-zero code.""" |
| self.mock_run.return_value = mock.MagicMock( |
| returncode=1, stderr='not found' |
| ) |
| with self.assertRaises(RuntimeError): |
| inspect_coverage_build.run_bb_get('999') |
| |
| def test_verify_configuration_java(self) -> None: |
| """Verifies configuration audit for Java language context.""" |
| build_data = { |
| 'input': {'properties': {}}, |
| 'steps': [{'name': 'lookup GN args', 'status': 'SUCCESS'}], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'java') |
| self.assertTrue(result['configuration_valid']) |
| |
| def test_verify_configuration_gn_arg_mismatches(self) -> None: |
| """Verifies gn_arg_mismatches are detected when input gn_args conflict with required_gn_args.""" |
| build_data = { |
| 'input': { |
| 'properties': { |
| 'gn_args': { |
| 'use_clang_coverage': False, |
| 'is_clang': True, |
| 'is_debug': False, |
| } |
| } |
| }, |
| 'steps': [], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'cpp') |
| self.assertFalse(result['configuration_valid']) |
| self.assertEqual(len(result['gn_arg_mismatches']), 1) |
| self.assertIn( |
| 'use_clang_coverage = False (expected True)', |
| result['gn_arg_mismatches'][0], |
| ) |
| |
| def test_verify_configuration_in_props_gn_arg_mismatch(self) -> None: |
| """Verifies boolean input properties are checked for required GN args when gn_args dict is omitted.""" |
| build_data = { |
| 'input': { |
| 'properties': { |
| 'use_clang_coverage': False, |
| } |
| }, |
| 'steps': [], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'cpp') |
| self.assertFalse(result['configuration_valid']) |
| self.assertIn( |
| 'use_clang_coverage = False (expected True)', |
| result['gn_arg_mismatches'][0], |
| ) |
| |
| def test_verify_configuration_orchestrator_compilator_lookup_gn_args( |
| self, |
| ) -> None: |
| """Verifies orchestrator builds extract lookup GN args status from compilator child steps.""" |
| build_data = { |
| 'input': { |
| 'properties': { |
| 'recipe': 'chromium_orchestrator', |
| } |
| }, |
| 'steps': [ |
| { |
| 'name': 'compilator steps (with patch)|lookup GN args', |
| 'status': 'SUCCESS', |
| } |
| ], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'cpp') |
| self.assertEqual(result['lookup_gn_args_step_status'], 'SUCCESS') |
| |
| def test_verify_configuration_orchestrator_nested_substeps_lookup_gn_args( |
| self, |
| ) -> None: |
| """Verifies lookup GN args status is found when nested inside substeps/children arrays.""" |
| build_data = { |
| 'input': { |
| 'properties': { |
| 'recipe': 'chromium_orchestrator', |
| } |
| }, |
| 'steps': [ |
| { |
| 'name': 'compilator steps (with patch)', |
| 'status': 'SUCCESS', |
| 'substeps': [ |
| { |
| 'name': 'lookup GN args', |
| 'status': 'SUCCESS', |
| } |
| ], |
| } |
| ], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'cpp') |
| self.assertEqual(result['lookup_gn_args_step_status'], 'SUCCESS') |
| |
| def test_verify_coverage_pipeline_url_conversions(self) -> None: |
| """Verifies URL helper conversion of gs:// paths to https:// and pantheon links.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'process clang code coverage data for overall test coverage', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for overall test coverage|generate html report in 10 tests', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for overall test coverage|gsutil upload html report', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| 'output': { |
| 'properties': { |
| 'gsutil_urls': { |
| 'process clang code coverage data for overall test coverage': 'gs://bucket/meta', |
| 'process clang code coverage data for overall test coverage|gsutil upload html report': 'gs://bucket/html', |
| 'process clang code coverage data for overall test coverage|gsutil upload artifact to GS': 'gs://bucket/profdata', |
| } |
| } |
| }, |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| overall = result['pipelines_checked']['cpp_overall'] |
| |
| self.assertEqual(overall['merged_profdata_gcs_url'], 'gs://bucket/profdata') |
| |
| def test_verify_coverage_pipeline_detects_recipe_messages(self) -> None: |
| """Verifies official recipe skip/error messages and triage actions are detected.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'skip processing clang coverage data because no profile data collected', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for overall test coverage|skip processing because no data is found', |
| 'status': 'SUCCESS', |
| }, |
| ] |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| overall = result['pipelines_checked']['cpp_overall'] |
| messages = [ |
| m['detected_condition'] for m in overall['recipe_error_messages_detected'] |
| ] |
| self.assertIn( |
| 'skip processing clang coverage data because no profile data collected', |
| messages, |
| ) |
| self.assertIn('skip processing because no data is found', messages) |
| |
| def test_format_line_ranges(self) -> None: |
| """Verifies format_line_ranges compresses contiguous line numbers correctly.""" |
| self.assertEqual( |
| inspect_coverage_build.format_line_ranges([1, 2, 3, 5, 8, 9]), |
| '1-3, 5, 8-9', |
| ) |
| self.assertEqual(inspect_coverage_build.format_line_ranges([]), 'None') |
| |
| def test_extract_coverage_artifacts_for_files(self) -> None: |
| """Verifies target file coverage metrics are extracted across pipelines.""" |
| pipeline_report = { |
| 'pipelines_checked': { |
| 'cpp_unit': { |
| 'pipeline_success': True, |
| 'html_report_gcs_url': 'gs://bucket/report', |
| } |
| } |
| } |
| file_res = inspect_coverage_build.extract_coverage_artifacts_for_files( |
| pipeline_report, target_files=['chrome/browser/ui/foo.cc'] |
| ) |
| self.assertIn('chrome/browser/ui/foo.cc', file_res) |
| unit_info = file_res['chrome/browser/ui/foo.cc']['cpp_unit'] |
| self.assertFalse(unit_info['available']) |
| |
| def test_format_file_coverage_report(self) -> None: |
| """Verifies readable text formatting of target file coverage breakdown.""" |
| file_data = { |
| 'foo.cc': { |
| 'cpp_unit': { |
| 'available': True, |
| 'line_coverage': '80.0% (80/100)', |
| 'function_coverage': '100.0% (5/5)', |
| 'region_coverage': '75.0%', |
| 'branch_coverage': '70.0%', |
| 'uncovered_lines': [10, 11, 12, 25], |
| } |
| } |
| } |
| text = inspect_coverage_build.format_file_coverage_report(file_data) |
| self.assertIn('[Target File Coverage Extraction]', text) |
| self.assertIn('* File: foo.cc', text) |
| self.assertIn('Line Coverage: 80.0% (80/100)', text) |
| self.assertIn('Uncovered Lines (4): 10-12, 25', text) |
| |
| def test_verify_configuration_orchestrator_delegated(self) -> None: |
| """Verifies orchestrator builds delegate GN args lookup to compilator.""" |
| build_data = { |
| 'input': {'properties': {'recipe': 'chromium_orchestrator'}}, |
| 'steps': [], |
| } |
| result = inspect_coverage_build.verify_configuration(build_data, 'cpp') |
| self.assertEqual( |
| result['lookup_gn_args_step_status'], |
| inspect_coverage_build.StepStatus.DELEGATED_TO_COMPILATOR, |
| ) |
| |
| def test_verify_coverage_pipeline_all_recipe_error_flags(self) -> None: |
| """Verifies detection of invalid profraw, failure flags, and skip profdata.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'process clang code coverage data for overall test coverage|foo', |
| 'status': 'FAILURE', |
| 'output': {'properties': {'process_coverage_data_failure': True}}, |
| }, |
| { |
| 'name': 'process clang code coverage data for overall test coverage|Found invalid profraw files', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for overall test coverage|skip processing because no profdata was generated', |
| 'status': 'SUCCESS', |
| }, |
| ] |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| overall = result['pipelines_checked']['cpp_overall'] |
| messages = [ |
| m['detected_condition'] for m in overall['recipe_error_messages_detected'] |
| ] |
| self.assertIn('process_coverage_data_failure = True', messages) |
| self.assertIn('Found invalid profraw files', messages) |
| self.assertIn('skip processing because no profdata was generated', messages) |
| |
| def test_extract_coverage_artifacts_for_files_unmocked(self) -> None: |
| """Verifies real file report extraction runs fetch_and_parse or not available.""" |
| pipeline_report = { |
| 'pipelines_checked': { |
| 'cpp_unit': { |
| 'pipeline_success': False, |
| 'html_report_gcs_url': 'gs://bucket/report', |
| } |
| } |
| } |
| file_res = inspect_coverage_build.extract_coverage_artifacts_for_files( |
| pipeline_report, target_files=['foo.cc'] |
| ) |
| self.assertFalse(file_res['foo.cc']['cpp_unit']['available']) |
| |
| def test_format_file_coverage_report_not_available(self) -> None: |
| """Verifies format_file_coverage_report handles unavailable pipelines.""" |
| file_data = {'foo.cc': {'cpp_unit': {'available': False}}} |
| text = inspect_coverage_build.format_file_coverage_report(file_data) |
| self.assertIn("Pipeline 'cpp_unit': Not Available", text) |
| |
| def test_format_inspection_report_comprehensive(self) -> None: |
| """Verifies format_inspection_report outputs all optional sections and mismatches.""" |
| combined_report = { |
| 'status_check': { |
| 'build_id': '12345', |
| 'status': 'STARTED', |
| 'terminal': False, |
| }, |
| 'config_verification': { |
| 'configuration_valid': False, |
| 'lookup_gn_args_step_status': 'FAILURE', |
| 'gn_arg_mismatches': ['use_clang_coverage missing'], |
| 'gclient_config_mismatches': ['checkout_clang_coverage_tools missing'], |
| }, |
| 'coverage_pipeline_verification': { |
| 'all_pipelines_successful': False, |
| 'pipelines_checked': { |
| 'cpp_overall': { |
| 'parent_step_status': 'FAILURE', |
| 'html_report_generation_status': 'FAILURE', |
| 'artifacts_upload_status': 'NOT_FOUND', |
| 'tests_processed_count': 10, |
| 'skipped_no_data': True, |
| 'recipe_error_messages_detected': [ |
| {'detected_condition': 'foo', 'step_name': 'bar'} |
| ], |
| 'child_steps': [{'name': 'substep', 'status': 'FAILURE'}], |
| 'html_report_url': 'https://storage.cloud.google.com/foo/index.html', |
| 'merged_profdata_gcs_url': 'gs://foo/merged.profdata', |
| 'pipeline_success': False, |
| } |
| }, |
| }, |
| 'target_file_coverage': { |
| 'foo.cc': { |
| 'cpp_overall': { |
| 'available': True, |
| 'line_coverage': '50%', |
| 'function_coverage': '50%', |
| 'region_coverage': '50%', |
| 'branch_coverage': '50%', |
| 'uncovered_lines': [1], |
| } |
| } |
| }, |
| } |
| text = inspect_coverage_build.format_inspection_report( |
| combined_report, 'cpp' |
| ) |
| self.assertIn('Build status is non-terminal (STARTED)', text) |
| self.assertIn('use_clang_coverage missing', text) |
| self.assertIn('checkout_clang_coverage_tools missing', text) |
| self.assertIn("DETECTED RECIPE CONDITION: 'foo'", text) |
| self.assertIn("SKIPPED: 'skip processing because no data is found'", text) |
| self.assertIn('Substep breakdown:', text) |
| self.assertIn('Merged Profdata GCS: gs://foo/merged.profdata', text) |
| |
| def test_verify_coverage_pipeline_java_and_js(self) -> None: |
| """Verifies Java and JS coverage pipelines and metadata generation steps.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'process java coverage (overall)', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process java coverage (overall)|Generate Java coverage metadata', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process java coverage (overall)|gsutil Upload coverage artifacts (2)', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process javascript coverage (overall)', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process javascript coverage (overall)|Generate JavaScript coverage metadata', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process javascript coverage (overall)|gsutil Upload coverage artifacts', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| 'output': { |
| 'properties': { |
| 'gsutil_urls': { |
| 'process java coverage (overall)|gsutil Upload coverage artifacts (2)': 'gs://bucket/java_meta', |
| 'process javascript coverage (overall)|gsutil Upload coverage artifacts': 'gs://bucket/js_meta', |
| } |
| } |
| }, |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| java_res = result['pipelines_checked']['java_overall'] |
| self.assertTrue(java_res['pipeline_success']) |
| self.assertEqual( |
| java_res['all_json_gz_url'], 'gs://bucket/java_meta/all.json.gz' |
| ) |
| |
| def test_verify_coverage_pipeline_upload_step_fallback(self) -> None: |
| """Verifies fallback between upload step (2) and without (2) when checking status and GCS URL.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'process clang code coverage data for overall test coverage', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'process clang code coverage data for overall test coverage|generate html report in 5 tests', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'gsutil Upload coverage artifacts', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| 'output': { |
| 'properties': { |
| 'gsutil_urls': { |
| 'gsutil Upload coverage artifacts': 'gs://bucket/fallback', |
| } |
| } |
| }, |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| overall = result['pipelines_checked']['cpp_overall'] |
| self.assertTrue(overall['pipeline_success']) |
| self.assertEqual( |
| overall['all_json_gz_url'], 'gs://bucket/fallback/all.json.gz' |
| ) |
| |
| @mock.patch('inspect_coverage_build.fetch_and_parse_json_file_coverage') |
| def test_extract_coverage_artifacts_for_files_json_gz( |
| self, mock_parse_json: mock.MagicMock |
| ) -> None: |
| """Verifies extract_coverage_artifacts_for_files uses all_json_gz_url when present.""" |
| mock_parse_json.return_value = { |
| 'available': True, |
| 'line_coverage': '90.0% (90/100)', |
| } |
| pipeline_report = { |
| 'pipelines_checked': { |
| 'java_overall': { |
| 'pipeline_success': True, |
| 'all_json_gz_url': 'gs://bucket/all.json.gz', |
| 'html_report_gcs_url': 'gs://bucket/html_report/index.html', |
| } |
| } |
| } |
| file_res = inspect_coverage_build.extract_coverage_artifacts_for_files( |
| pipeline_report, target_files=['foo/bar.java'] |
| ) |
| self.assertTrue(file_res['foo/bar.java']['java_overall']['available']) |
| self.assertEqual( |
| file_res['foo/bar.java']['java_overall']['line_coverage'], |
| '90.0% (90/100)', |
| ) |
| mock_parse_json.assert_called_once_with( |
| 'gs://bucket/all.json.gz', 'foo/bar.java' |
| ) |
| |
| @mock.patch('subprocess.run') |
| def test_fetch_and_parse_json_file_coverage_dict_summaries( |
| self, mock_run: mock.MagicMock |
| ) -> None: |
| """Verifies fetch_and_parse_json_file_coverage parses dict summary format and uncovered lines.""" |
| import gzip |
| import json |
| |
| data = { |
| 'files': [ |
| { |
| 'path': 'foo/bar.java', |
| 'lines': [ |
| {'first': 10, 'last': 12, 'count': 0}, |
| {'first': 13, 'last': 13, 'count': 5}, |
| ], |
| 'summaries': { |
| 'line': {'covered': 8, 'count': 10}, |
| 'method': {'covered': 2, 'total': 2}, |
| }, |
| } |
| ] |
| } |
| with tempfile.NamedTemporaryFile(suffix='.json.gz', delete=False) as tf: |
| tf.write(gzip.compress(json.dumps(data).encode('utf-8'))) |
| src_path = tf.name |
| try: |
| |
| def fake_cp(*args, **kwargs): |
| dest = args[0][4] |
| with open(src_path, 'rb') as f_in, open(dest, 'wb') as f_out: |
| f_out.write(f_in.read()) |
| return mock.MagicMock(returncode=0) |
| |
| mock_run.side_effect = fake_cp |
| res = inspect_coverage_build.fetch_and_parse_json_file_coverage( |
| 'gs://bucket/all.json.gz', 'bar.java' |
| ) |
| self.assertTrue(res['available']) |
| self.assertEqual(res['line_coverage'], '8/10 (80.00%)') |
| self.assertEqual(res['function_coverage'], '2/2 (100.00%)') |
| self.assertEqual(res['uncovered_lines'], [10, 11, 12]) |
| finally: |
| if os.path.exists(src_path): |
| os.remove(src_path) |
| |
| @mock.patch('subprocess.run') |
| def test_fetch_and_parse_json_file_coverage_list_summaries_or_absent( |
| self, mock_run: mock.MagicMock |
| ) -> None: |
| """Verifies list-based summaries and missing target handling in fetch_and_parse_json_file_coverage.""" |
| import gzip |
| import json |
| |
| data = { |
| 'files': [ |
| { |
| 'filename': 'foo/other.js', |
| 'summaries': [{'name': 'line', 'covered': 15, 'count': 20}], |
| } |
| ] |
| } |
| with tempfile.NamedTemporaryFile(suffix='.json.gz', delete=False) as tf: |
| tf.write(gzip.compress(json.dumps(data).encode('utf-8'))) |
| src_path = tf.name |
| try: |
| |
| def fake_cp(*args, **kwargs): |
| dest = args[0][4] |
| with open(src_path, 'rb') as f_in, open(dest, 'wb') as f_out: |
| f_out.write(f_in.read()) |
| return mock.MagicMock(returncode=0) |
| |
| mock_run.side_effect = fake_cp |
| res = inspect_coverage_build.fetch_and_parse_json_file_coverage( |
| 'gs://bucket/all.json.gz', 'other.js' |
| ) |
| self.assertTrue(res['available']) |
| self.assertEqual(res['line_coverage'], '15/20 (75.00%)') |
| |
| res_absent = inspect_coverage_build.fetch_and_parse_json_file_coverage( |
| 'gs://bucket/all.json.gz', 'missing.js' |
| ) |
| self.assertFalse(res_absent['available']) |
| |
| self.assertFalse( |
| inspect_coverage_build.fetch_and_parse_json_file_coverage( |
| '', 'other.js' |
| )['available'] |
| ) |
| finally: |
| if os.path.exists(src_path): |
| os.remove(src_path) |
| |
| def test_verify_coverage_pipeline_upload_step_primary_match(self) -> None: |
| """Verifies primary upload_step_name match when nested upload status is NOT_FOUND.""" |
| build_data = { |
| 'steps': [ |
| { |
| 'name': 'process clang code coverage data for overall test coverage', |
| 'status': 'SUCCESS', |
| }, |
| { |
| 'name': 'gsutil Upload coverage artifacts (2)', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| 'output': {'properties': {'gsutil_urls': {}}}, |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| overall = result['pipelines_checked']['cpp_overall'] |
| |
| @mock.patch.dict(os.environ, {'INSPECT_COVERAGE_NO_NETWORK': ''}) |
| @mock.patch('subprocess.run') |
| def test_verify_coverage_pipeline_gcs_fallback_when_url_missing( |
| self, mock_run: mock.MagicMock |
| ) -> None: |
| """Verifies gsutil ls fallback execution when all_json_gz_url is missing.""" |
| mock_run.return_value = mock.MagicMock( |
| returncode=0, stdout='gs://bucket/fallback/all.json.gz\n' |
| ) |
| build_data = { |
| 'id': '123456789', |
| 'steps': [ |
| { |
| 'name': 'process clang code coverage data for overall test coverage', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| 'output': {'properties': {'gsutil_urls': {}}}, |
| } |
| result = inspect_coverage_build.verify_coverage_pipeline(build_data) |
| overall = result['pipelines_checked']['cpp_overall'] |
| self.assertEqual( |
| overall['all_json_gz_url'], 'gs://bucket/fallback/all.json.gz' |
| ) |
| |
| @mock.patch('subprocess.run') |
| def test_fetch_and_parse_json_file_coverage_uncompressed_fallback( |
| self, mock_run: mock.MagicMock |
| ) -> None: |
| """Verifies fallback to uncompressed json.loads when zlib and gzip decompression both fail.""" |
| import json |
| |
| data = { |
| 'files': [ |
| { |
| 'path': 'foo/raw.js', |
| 'summaries': [{'name': 'line', 'covered': 5, 'count': 5}], |
| } |
| ] |
| } |
| with tempfile.NamedTemporaryFile(suffix='.json.gz', delete=False) as tf: |
| tf.write(json.dumps(data).encode('utf-8')) |
| src_path = tf.name |
| try: |
| |
| def fake_cp(*args, **kwargs): |
| dest = args[0][4] |
| with open(src_path, 'rb') as f_in, open(dest, 'wb') as f_out: |
| f_out.write(f_in.read()) |
| return mock.MagicMock(returncode=0) |
| |
| mock_run.side_effect = fake_cp |
| res = inspect_coverage_build.fetch_and_parse_json_file_coverage( |
| 'gs://bucket/all.json.gz', 'raw.js' |
| ) |
| self.assertTrue(res['available']) |
| self.assertEqual(res['line_coverage'], '5/5 (100.00%)') |
| finally: |
| if os.path.exists(src_path): |
| os.remove(src_path) |
| |
| @mock.patch('subprocess.run') |
| def test_fetch_and_parse_json_file_coverage_exceptions( |
| self, mock_run: mock.MagicMock |
| ) -> None: |
| """Verifies exception handling on corrupt JSON and os.remove failure inside fetch_and_parse.""" |
| with tempfile.NamedTemporaryFile(suffix='.json.gz', delete=False) as tf: |
| tf.write(b'corrupt non-json bytes') |
| src_path = tf.name |
| try: |
| |
| def fake_cp(*args, **kwargs): |
| dest = args[0][4] |
| with open(src_path, 'rb') as f_in, open(dest, 'wb') as f_out: |
| f_out.write(f_in.read()) |
| return mock.MagicMock(returncode=0) |
| |
| mock_run.side_effect = fake_cp |
| with mock.patch('os.remove', side_effect=OSError('Permission denied')): |
| res = inspect_coverage_build.fetch_and_parse_json_file_coverage( |
| 'gs://bucket/all.json.gz', 'any.js' |
| ) |
| self.assertFalse(res['available']) |
| finally: |
| if os.path.exists(src_path): |
| os.remove(src_path) |
| |
| |
| class InspectCoverageBuildMainTest(unittest.TestCase): |
| """Tests CLI main() execution modes and error reporting.""" |
| |
| def setUp(self) -> None: |
| """Initializes stdout, stderr, exit, and run_bb_get patchers for main().""" |
| self.stdout_buffer = io.StringIO() |
| self.stderr_buffer = io.StringIO() |
| self.stdout_context = contextlib.redirect_stdout(self.stdout_buffer) |
| self.stderr_context = contextlib.redirect_stderr(self.stderr_buffer) |
| self.stdout_context.__enter__() |
| self.stderr_context.__enter__() |
| self.addCleanup(self.stdout_context.__exit__, None, None, None) |
| self.addCleanup(self.stderr_context.__exit__, None, None, None) |
| |
| self.exit_patcher = mock.patch('sys.exit', side_effect=SystemExit) |
| self.bb_patcher = mock.patch('inspect_coverage_build.run_bb_get') |
| self.run_patcher = mock.patch('subprocess.run') |
| |
| self.mock_exit = self.exit_patcher.start() |
| self.mock_bb = self.bb_patcher.start() |
| self.mock_run = self.run_patcher.start() |
| |
| self.addCleanup(self.exit_patcher.stop) |
| self.addCleanup(self.bb_patcher.stop) |
| self.addCleanup(self.run_patcher.stop) |
| |
| def test_main_json_output(self) -> None: |
| """Verifies main() formats and outputs JSON when --json flag is passed.""" |
| self.mock_bb.return_value = { |
| 'id': '8676997864980431489', |
| 'status': 'SUCCESS', |
| 'steps': [], |
| } |
| with mock.patch( |
| 'sys.argv', |
| [ |
| 'inspect_coverage_build.py', |
| '--build', |
| '8676997864980431489', |
| '--language', |
| 'cpp', |
| '--json', |
| ], |
| ): |
| inspect_coverage_build.main() |
| self.mock_bb.assert_called_once_with('8676997864980431489') |
| self.mock_exit.assert_not_called() |
| |
| def test_main_text_output(self) -> None: |
| """Verifies main() prints human-readable inspection summary by default.""" |
| self.mock_bb.return_value = { |
| 'id': '8676997864980431489', |
| 'status': 'STARTED', |
| 'steps': [ |
| {'name': 'lookup GN args', 'status': 'SUCCESS'}, |
| { |
| 'name': 'process clang code coverage data for unit test coverage', |
| 'status': 'SUCCESS', |
| }, |
| ], |
| } |
| with mock.patch( |
| 'sys.argv', |
| [ |
| 'inspect_coverage_build.py', |
| '--build', |
| '8676997864980431489', |
| '--language', |
| 'cpp', |
| ], |
| ): |
| inspect_coverage_build.main() |
| self.mock_bb.assert_called_once_with('8676997864980431489') |
| self.mock_exit.assert_not_called() |
| |
| def test_main_with_files_argument(self) -> None: |
| """Verifies main() extracts per-file coverage reports when --files is passed.""" |
| self.mock_bb.return_value = { |
| 'id': '8676997864980431489', |
| 'status': 'SUCCESS', |
| 'steps': [], |
| } |
| with mock.patch( |
| 'sys.argv', |
| [ |
| 'inspect_coverage_build.py', |
| '--build', |
| '8676997864980431489', |
| '--language', |
| 'cpp', |
| '--files', |
| 'foo.cc', |
| ], |
| ): |
| inspect_coverage_build.main() |
| self.mock_bb.assert_called_once_with('8676997864980431489') |
| self.mock_exit.assert_not_called() |
| |
| def test_main_exception_exits(self) -> None: |
| """Verifies main() logs error and exits with status 1 on exceptions.""" |
| self.mock_bb.side_effect = RuntimeError('API Error') |
| with mock.patch( |
| 'sys.argv', |
| [ |
| 'inspect_coverage_build.py', |
| '--build', |
| '8676997864980431489', |
| '--language', |
| 'cpp', |
| ], |
| ): |
| with self.assertRaises(SystemExit): |
| inspect_coverage_build.main() |
| self.mock_exit.assert_called_once_with(1) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
