tools/code_coverage/test_suite_mapper_test.py

#!/usr/bin/env vpython3
# 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 test_suite_mapper.py."""

import contextlib
import io
import pathlib
import unittest
from unittest.mock import MagicMock, patch

from pyfakefs import fake_filesystem_unittest

import test_suite_mapper

MOCK_ISOLATE_MAP = {
  '//chrome/test:browser_tests': 'browser_tests',
  '//chrome/test:unit_tests': 'unit_tests',
  '//media/midi:midi_unittests': 'midi_unittests',
  '//media:media_unittests': 'media_unittests',
  '//chrome/browser:foo_tests': 'foo_tests',
  '//content/test:bar_tests': 'bar_tests',
  '//foo/bar:baz_tests': 'baz_tests',
  '//base:base_unittests': 'base_unittests',
  '//:blink_wpt_tests': 'blink_wpt_tests',
}


class TestSuiteMapperUnit(fake_filesystem_unittest.TestCase):
  def setUp(self):
    self.setUpPyfakefs()
    self.subprocess_patcher = patch('subprocess.run')
    self.mock_run = self.subprocess_patcher.start()
    self.addCleanup(self.subprocess_patcher.stop)

  def test_common_dir_prefix_len(self):
    self.assertEqual(
      test_suite_mapper.common_dir_prefix_len('a/b/c', 'a/b/c'), 3
    )
    self.assertEqual(
      test_suite_mapper.common_dir_prefix_len('a/b/c', 'a/b/d'), 2
    )
    self.assertEqual(
      test_suite_mapper.common_dir_prefix_len('a/b/c', 'x/y/z'), 0
    )
    self.assertEqual(test_suite_mapper.common_dir_prefix_len('a/b/c', ''), 0)
    self.assertEqual(test_suite_mapper.common_dir_prefix_len('a/b', 'a/b/c'), 2)

  def test_parse_isolate_map_valid(self):
    content = (
      "{'browser_tests': {'label': '//chrome/test:browser_tests'},"
      " 'unit_tests': {'label': '//chrome/test:unit_tests'}}"
    )
    expected = {
      '//chrome/test:browser_tests': 'browser_tests',
      '//chrome/test:unit_tests': 'unit_tests',
    }
    self.assertEqual(
      test_suite_mapper.parse_isolate_map(content, 'test'), expected
    )

  def test_parse_isolate_map_missing_label(self):
    content = "{'suite1': {'type': 'additional_compile_target'}}"
    self.assertEqual(test_suite_mapper.parse_isolate_map(content, 'test'), {})

  def test_parse_isolate_map_invalid(self):
    content = 'invalid dict'
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      self.assertIsNone(test_suite_mapper.parse_isolate_map(content, 'test'))
    self.assertTrue(len(f.getvalue()) > 0)

  def test_parse_isolate_map_not_dict(self):
    content = '[1, 2, 3]'
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      self.assertIsNone(test_suite_mapper.parse_isolate_map(content, 'test'))
    self.assertIn('Expected a dictionary', f.getvalue())

  def test_load_isolate_map_success(self):
    self.fs.create_dir('/src')
    pyl_path = pathlib.Path('/src') / test_suite_mapper.ISOLATE_MAP_PATH
    self.fs.create_file(
      pyl_path,
      contents="{'browser_tests': {'label': '//chrome/test:browser_tests'}}",
    )
    isolate_map = test_suite_mapper.load_isolate_map('/src')
    self.assertEqual(
      isolate_map, {'//chrome/test:browser_tests': 'browser_tests'}
    )

  def test_load_isolate_map_missing_file(self):
    self.fs.create_dir('/src')
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      self.assertIsNone(test_suite_mapper.load_isolate_map('/src'))
    self.assertIn('Isolate map not found', f.getvalue())

  def test_load_isolate_map_read_error(self):
    self.fs.create_dir('/src')
    pyl_path = pathlib.Path('/src') / test_suite_mapper.ISOLATE_MAP_PATH
    self.fs.create_dir(pyl_path)
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      self.assertIsNone(test_suite_mapper.load_isolate_map('/src'))
    self.assertIn('Error reading', f.getvalue())

  def test_extract_test_suites_basic(self):
    refs = [
      '//chrome/browser:foo_tests',
      '//content/test:bar_tests',
      '//foo/bar:baz_tests',
    ]
    suites = test_suite_mapper.extract_test_suites(
      refs, pathlib.Path('chrome/browser/foo.cc'), MOCK_ISOLATE_MAP
    )
    self.assertEqual(suites, ['foo_tests'])

  def test_extract_test_suites_proximity(self):
    refs = [
      '//chrome/test:browser_tests',
      '//media/midi:midi_unittests',
    ]
    suites = test_suite_mapper.extract_test_suites(
      refs, pathlib.Path('media/midi/midi_manager.cc'), MOCK_ISOLATE_MAP
    )
    self.assertEqual(suites, ['midi_unittests'])

  def test_extract_test_suites_filter_non_isolate(self):
    refs = [
      '//media/base:unit_tests',
      '//media:media_unittests',
    ]
    suites = test_suite_mapper.extract_test_suites(
      refs,
      pathlib.Path('media/base/audio_bus_unittest.cc'),
      MOCK_ISOLATE_MAP,
    )
    self.assertEqual(suites, ['media_unittests'])

  def test_extract_test_suites_empty_target(self):
    refs = ['//chrome/browser:foo_tests', '', '  ']
    suites = test_suite_mapper.extract_test_suites(
      refs, pathlib.Path('chrome/browser/foo.cc'), MOCK_ISOLATE_MAP
    )
    self.assertEqual(suites, ['foo_tests'])

  def test_extract_test_suites_no_match(self):
    refs = ['//unrelated:target']
    suites = test_suite_mapper.extract_test_suites(
      refs, pathlib.Path('chrome/browser/foo.cc'), MOCK_ISOLATE_MAP
    )
    self.assertEqual(suites, [])

  def test_extract_test_suites_all_zero_score(self):
    refs = ['//media/midi:midi_unittests', '//foo/bar:baz_tests']
    suites = test_suite_mapper.extract_test_suites(
      refs, pathlib.Path('chrome/browser/foo.cc'), MOCK_ISOLATE_MAP
    )
    self.assertEqual(sorted(suites), ['baz_tests', 'midi_unittests'])

  def test_verify_build_dir_valid(self):
    self.fs.create_file(
      '/src/out/Default/args.gn',
      contents=(
        '# This is a comment\ntarget_os = "linux"\n\nuse_goma = true\n'
      ),
    )
    success, err = test_suite_mapper.verify_build_dir('/src', 'out/Default')
    self.assertTrue(success)
    self.assertIsNone(err)

  def test_verify_build_dir_missing_file(self):
    self.fs.create_dir('/src')
    success, err = test_suite_mapper.verify_build_dir('/src', 'out/Default')
    self.assertFalse(success)
    self.assertIn('args.gn not found', err)

  def test_verify_build_dir_invalid_os(self):
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='target_os = "android"\n'
    )
    success, err = test_suite_mapper.verify_build_dir('/src', 'out/Default')
    self.assertFalse(success)
    self.assertIn('Unsupported target_os: android', err)

  @patch('sys.platform', 'linux')
  def test_verify_build_dir_no_target_os_on_linux(self):
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='is_debug = true\n'
    )
    success, err = test_suite_mapper.verify_build_dir('/src', 'out/Default')
    self.assertTrue(success)
    self.assertIsNone(err)

  @patch('sys.platform', 'darwin')
  def test_verify_build_dir_no_target_os_on_mac(self):
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='is_debug = true\n'
    )
    success, err = test_suite_mapper.verify_build_dir('/src', 'out/Default')
    self.assertFalse(success)
    self.assertIn('Host platform darwin is not Linux', err)

  def test_run_gn_refs_calls_gn(self):
    self.mock_run.return_value = MagicMock(
      returncode=0, stdout='//foo:foo_unittests\n', stderr=''
    )
    suites, err = test_suite_mapper.run_gn_refs(
      '/src', 'out/Default', pathlib.Path('foo/bar.cc')
    )
    self.assertEqual(suites, ['//foo:foo_unittests'])
    self.assertIsNone(err)
    self.mock_run.assert_called_once_with(
      [
        'gn',
        'refs',
        'out/Default',
        '--all',
        '//foo/bar.cc',
      ],
      capture_output=True,
      text=True,
      cwd='/src',
      check=False,
      encoding='utf-8',
    )

  def test_run_gn_refs_failure(self):
    self.mock_run.return_value = MagicMock(
      returncode=1, stdout='', stderr='GN error message\n'
    )
    suites, err = test_suite_mapper.run_gn_refs(
      '/src', 'out/Default', pathlib.Path('foo/bar.cc')
    )
    self.assertIsNone(suites)
    self.assertEqual(err, 'GN error message')

  def test_run_gn_refs_exception(self):
    self.mock_run.side_effect = OSError('Command not found')
    suites, err = test_suite_mapper.run_gn_refs(
      '/src', 'out/Default', pathlib.Path('foo/bar.cc')
    )
    self.assertIsNone(suites)
    self.assertEqual(err, 'Command not found')

  def test_get_build_gn_paths(self):
    refs = [
      '//chrome/browser:foo_tests',
      '//content/test:bar_tests',
      '//:blink_wpt_tests',
      'invalid_target',
    ]
    paths = test_suite_mapper.get_build_gn_paths(refs)
    self.assertEqual(
      paths,
      ['BUILD.gn', 'chrome/browser/BUILD.gn', 'content/test/BUILD.gn'],
    )

  def test_check_graph_stability_stable(self):
    self.mock_run.return_value = MagicMock(returncode=0, stdout='')
    refs = ['//chrome/browser:foo_tests']
    stable, changed = test_suite_mapper.check_graph_stability(
      '/src', 'abc123f', 'chrome/browser/foo.cc', refs
    )
    self.assertTrue(stable)
    self.assertEqual(changed, [])
    self.mock_run.assert_called_once_with(
      [
        'git',
        'diff',
        '--name-only',
        'abc123f..HEAD',
        '--',
        'chrome/browser/foo.cc',
        'chrome/browser/BUILD.gn',
      ],
      capture_output=True,
      text=True,
      cwd='/src',
      check=False,
      encoding='utf-8',
    )

  def test_check_graph_stability_unstable(self):
    self.mock_run.return_value = MagicMock(
      returncode=0,
      stdout='chrome/browser/BUILD.gn\n',
    )
    refs = ['//chrome/browser:foo_tests']
    stable, changed = test_suite_mapper.check_graph_stability(
      '/src', 'abc123f', 'chrome/browser/foo.cc', refs
    )
    self.assertFalse(stable)
    self.assertEqual(changed, ['chrome/browser/BUILD.gn'])

  def test_check_graph_stability_git_fail(self):
    self.mock_run.return_value = MagicMock(
      returncode=1, stderr='Git diff error'
    )
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      stable, changed = test_suite_mapper.check_graph_stability(
        '/src', 'abc123f', 'chrome/browser/foo.cc', ['//chrome:foo']
      )
    self.assertFalse(stable)
    self.assertEqual(changed, [])
    self.assertIn('Stability check failed', f.getvalue())

  def test_get_ignored_paths(self):
    self.mock_run.return_value = MagicMock(
      returncode=0,
      stdout=('!! out/\n!! build/config/gclient_args.gni\n?? untracked_file\n'),
    )
    ignored = test_suite_mapper.get_ignored_paths('/src')
    self.assertEqual(ignored, ['out', 'build/config/gclient_args.gni'])

  def test_find_nested_repos_and_symlinks(self):
    self.fs.create_dir('/src/out')
    self.fs.create_dir('/src/.git')
    self.fs.create_dir('/src/third_party/foo')
    self.fs.create_file('/src/third_party/foo/.git')
    self.fs.create_dir('/src/third_party/bar/.git')
    self.fs.create_dir('/src/third_party/baz')

    # Deep nested repo (depth > 5)
    self.fs.create_file('/src/a/b/c/d/e/f/.git')

    # Symlinks
    self.fs.create_file('/src/third_party/baz/real_file')
    self.fs.create_symlink(
      '/src/third_party/link_to_file', '/src/third_party/baz/real_file'
    )
    self.fs.create_symlink(
      '/src/third_party/link_to_dir', '/src/third_party/baz'
    )
    self.fs.create_symlink(
      '/src/out/bad_link', '/src/third_party/baz/real_file'
    )

    # Symlink inside a nested repo should be skipped
    self.fs.create_symlink(
      '/src/third_party/foo/nested_link', '/src/third_party/baz/real_file'
    )

    targets = test_suite_mapper.find_nested_repos_and_symlinks('/src')
    expected = [
      'third_party/bar',
      'third_party/foo',
      'third_party/link_to_dir',
      'third_party/link_to_file',
    ]
    self.assertEqual(sorted(targets), expected)

  def test_stub_missing_target(self):
    wt_path = pathlib.Path('/wt')
    self.fs.create_dir('/wt')
    target_ref = '//third_party/foo:bar'
    success = test_suite_mapper._stub_missing_target(wt_path, target_ref)
    self.assertTrue(success)

    build_gn_path = wt_path / 'third_party' / 'foo' / 'BUILD.gn'
    self.assertTrue(build_gn_path.exists())
    content = build_gn_path.read_text(encoding='utf-8')
    self.assertIn('group("bar") {}', content)

  def test_stub_missing_target_failure(self):
    wt_path = pathlib.Path('/wt')
    self.fs.create_file('/wt/third_party')
    target_ref = '//third_party/foo:bar'
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      success = test_suite_mapper._stub_missing_target(wt_path, target_ref)
    self.assertFalse(success)
    self.assertTrue(len(f.getvalue()) > 0)

  def test_stub_missing_target_invalid_ref(self):
    wt_path = pathlib.Path('/wt')
    self.assertFalse(test_suite_mapper._stub_missing_target(wt_path, 'foo:bar'))
    self.assertFalse(
      test_suite_mapper._stub_missing_target(wt_path, '//foo/bar')
    )

  def test_spoof_rust_version_success(self):
    wt_path = pathlib.Path('/wt')
    self.fs.create_file(
      '/wt/tools/rust/update_rust.py',
      contents=(
        "RUST_REVISION = 'aaaa05163abcbd08948b3efab796c543ba1ea687'\n"
        "RUST_SUB_REVISION = 2\n"
      ),
    )
    src_dir = pathlib.Path('/src_rust')
    self.fs.create_file(
      '/src_rust/VERSION',
      contents=(
        'rustc 1.96.0 4c4205163abcbd08948b3efab796c543ba1ea687 '
        '(4c4205163abcbd08948b3efab796c543ba1ea687-5-'
        'llvmorg-23-init-10931-g20b6ec66 chromium)\n'
      ),
    )
    dest_dir = pathlib.Path('/dest_rust')
    self.fs.create_dir('/dest_rust')

    success = test_suite_mapper._spoof_rust_version(wt_path, src_dir, dest_dir)
    self.assertTrue(success)

    dest_version = dest_dir / 'VERSION'
    self.assertTrue(dest_version.exists())
    content = dest_version.read_text(encoding='utf-8')
    self.assertEqual(
      content,
      'rustc 1.96.0 aaaa05163abcbd08948b3efab796c543ba1ea687 '
      '(aaaa05163abcbd08948b3efab796c543ba1ea687-2-'
      'llvmorg-23-init-10931-g20b6ec66 chromium)\n',
    )

  def test_spoof_rust_version_missing_update_rust(self):
    wt_path = pathlib.Path('/wt')
    self.fs.create_dir('/wt')
    self.assertFalse(
      test_suite_mapper._spoof_rust_version(
        wt_path, pathlib.Path('/src'), pathlib.Path('/dest')
      )
    )

  def test_spoof_rust_version_invalid_update_rust(self):
    wt_path = pathlib.Path('/wt')
    self.fs.create_file(
      '/wt/tools/rust/update_rust.py', contents='invalid content\n'
    )
    self.assertFalse(
      test_suite_mapper._spoof_rust_version(
        wt_path, pathlib.Path('/src'), pathlib.Path('/dest')
      )
    )

  def test_spoof_rust_version_missing_src_version(self):
    wt_path = pathlib.Path('/wt')
    self.fs.create_file(
      '/wt/tools/rust/update_rust.py',
      contents="RUST_REVISION = 'aaaa'\nRUST_SUB_REVISION = 1\n",
    )
    self.assertFalse(
      test_suite_mapper._spoof_rust_version(
        wt_path, pathlib.Path('/src'), pathlib.Path('/dest')
      )
    )

  def test_spoof_rust_version_invalid_src_version_format(self):
    wt_path = pathlib.Path('/wt')
    self.fs.create_file(
      '/wt/tools/rust/update_rust.py',
      contents="RUST_REVISION = 'aaaa'\nRUST_SUB_REVISION = 1\n",
    )
    src_dir = pathlib.Path('/src')
    self.fs.create_file(src_dir / 'VERSION', contents='invalid format\n')
    self.assertFalse(
      test_suite_mapper._spoof_rust_version(
        wt_path, src_dir, pathlib.Path('/dest')
      )
    )

  @patch('test_suite_mapper.get_ignored_paths')
  @patch('test_suite_mapper.find_nested_repos_and_symlinks')
  def test_setup_worktree_symlinks(self, mock_find, mock_get_ignored):
    mock_get_ignored.return_value = ['third_party/llvm-build', 'out']
    mock_find.return_value = ['third_party/foo', 'third_party/link_to_file']

    self.fs.create_dir('/src')
    self.fs.create_dir('/src/third_party/llvm-build')
    self.fs.create_dir('/src/third_party/foo')
    self.fs.create_file('/src/third_party/link_to_file')
    self.fs.create_dir('/src/out/Default')
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='target_os = "linux"\n'
    )
    self.fs.create_file('/src/out/Default/other_file', contents='ignored\n')

    self.fs.create_dir('/wt')

    test_suite_mapper._setup_worktree_symlinks(
      pathlib.Path('/src'), pathlib.Path('/wt'), 'out/Default'
    )

    self.assertTrue(pathlib.Path('/wt/third_party/llvm-build').is_symlink())
    self.assertTrue(pathlib.Path('/wt/third_party/foo').is_symlink())
    self.assertTrue(pathlib.Path('/wt/third_party/link_to_file').is_symlink())
    self.assertFalse(pathlib.Path('/wt/out').is_symlink())
    self.assertFalse(pathlib.Path('/wt/out').exists())

  @patch('test_suite_mapper.get_ignored_paths')
  @patch('test_suite_mapper.find_nested_repos_and_symlinks')
  @patch('test_suite_mapper._spoof_rust_version')
  def test_setup_worktree_symlinks_rust_toolchain(
    self, mock_spoof, mock_find, mock_get_ignored
  ):
    mock_get_ignored.return_value = [
      test_suite_mapper.RUST_TOOLCHAIN_PATH.as_posix()
    ]
    mock_find.return_value = []
    mock_spoof.return_value = True

    self.fs.create_dir('/src')
    rust_dir = pathlib.Path('/src') / test_suite_mapper.RUST_TOOLCHAIN_PATH
    self.fs.create_file(rust_dir / 'bin/rustc')
    self.fs.create_file(rust_dir / 'VERSION')

    self.fs.create_dir('/wt')
    self.fs.create_file(
      pathlib.Path('/wt') / test_suite_mapper.UPDATE_RUST_PATH
    )

    test_suite_mapper._setup_worktree_symlinks(
      pathlib.Path('/src'), pathlib.Path('/wt'), 'out/Default'
    )

    link_rust_dir = pathlib.Path('/wt') / test_suite_mapper.RUST_TOOLCHAIN_PATH
    self.assertTrue((link_rust_dir / 'bin').is_symlink())
    self.assertTrue((link_rust_dir / 'bin/rustc').exists())
    self.assertFalse((link_rust_dir / 'VERSION').exists())
    mock_spoof.assert_called_once()

  def test_run_gn_gen_with_stubbing_success_immediate(self):
    self.mock_run.return_value = MagicMock(returncode=0)
    success = test_suite_mapper._run_gn_gen_with_stubbing(
      pathlib.Path('/wt'), 'out/Default'
    )
    self.assertTrue(success)
    self.mock_run.assert_called_once_with(
      ['gn', 'gen', 'out/Default'],
      capture_output=True,
      text=True,
      cwd=pathlib.Path('/wt'),
      check=False,
      encoding='utf-8',
    )

  def test_run_gn_gen_with_stubbing_retry_success(self):
    self.mock_run.side_effect = [
      MagicMock(
        returncode=1,
        stdout=(
          'ERROR Unresolved dependencies.\n  //foo:bar needs'
          ' //third_party/baz:qux\n'
        ),
        stderr='',
      ),
      MagicMock(returncode=0),
    ]
    self.fs.create_dir('/wt')
    success = test_suite_mapper._run_gn_gen_with_stubbing(
      pathlib.Path('/wt'), 'out/Default'
    )
    self.assertTrue(success)
    self.assertEqual(self.mock_run.call_count, 2)

    stub_path = pathlib.Path('/wt/third_party/baz/BUILD.gn')
    self.assertTrue(stub_path.exists())
    self.assertIn('group("qux") {}', stub_path.read_text(encoding='utf-8'))

  def test_run_gn_gen_with_stubbing_failure_loop(self):
    self.mock_run.return_value = MagicMock(
      returncode=1,
      stdout=(
        'ERROR Unresolved dependencies.\n  //foo:bar needs'
        ' //third_party/baz:qux\n'
      ),
      stderr='',
    )
    self.fs.create_dir('/wt')
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      success = test_suite_mapper._run_gn_gen_with_stubbing(
        pathlib.Path('/wt'), 'out/Default'
      )
    self.assertFalse(success)
    self.assertIn('already stubbed', f.getvalue())

  def test_run_heavyweight_resolution_success(self):
    isolate_map_content = (
      "{'browser_tests': {'label': '//chrome/test:browser_tests'}}"
    )

    def mock_run_side_effect(cmd, *args, **kwargs):
      if 'worktree' in cmd and 'add' in cmd:
        wt_path = cmd[5]
        self.fs.create_file(
          pathlib.Path(wt_path) / test_suite_mapper.ISOLATE_MAP_PATH,
          contents=isolate_map_content,
        )
        return MagicMock(returncode=0)
      elif 'gen' in cmd:
        return MagicMock(returncode=0)
      elif 'refs' in cmd:
        return MagicMock(
          returncode=0,
          stdout='//chrome/test:browser_tests\n',
          stderr='',
        )
      return MagicMock(returncode=0)

    self.mock_run.side_effect = mock_run_side_effect

    self.fs.create_dir('/src')
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='target_os = "linux"\n'
    )

    with (
      patch('test_suite_mapper.get_ignored_paths', return_value=[]),
      patch(
        'test_suite_mapper.find_nested_repos_and_symlinks', return_value=[]
      ),
    ):
      suites = test_suite_mapper.run_heavyweight_resolution(
        pathlib.Path('/src'),
        'abc123f',
        'out/Default',
        pathlib.Path('chrome/browser/foo.cc'),
      )

    self.assertEqual(suites, ['browser_tests'])

    # Verify worktree add and remove were called with the same path
    add_path = None
    remove_path = None
    for call in self.mock_run.call_args_list:
      cmd = call[0][0]
      if 'worktree' in cmd and 'add' in cmd:
        add_path = cmd[5]
      if 'worktree' in cmd and 'remove' in cmd:
        remove_path = cmd[4]

    self.assertIsNotNone(add_path)
    self.assertIsNotNone(remove_path)
    self.assertEqual(add_path, remove_path)

  def test_run_heavyweight_resolution_worktree_fail(self):
    self.mock_run.return_value = MagicMock(returncode=1, stderr='Git error')

    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      suites = test_suite_mapper.run_heavyweight_resolution(
        pathlib.Path('/src'),
        'abc123f',
        'out/Default',
        pathlib.Path('chrome/browser/foo.cc'),
      )
    self.assertIsNone(suites)
    self.assertIn('Error creating git worktree', f.getvalue())

  def test_run_heavyweight_resolution_gn_gen_fail(self):
    self.mock_run.side_effect = [
      MagicMock(returncode=0),  # worktree add
      MagicMock(returncode=1, stdout='GN gen error', stderr=''),  # gn gen fail
      MagicMock(returncode=0),  # worktree remove
    ]
    self.fs.create_dir('/src')
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='target_os = "linux"\n'
    )
    f = io.StringIO()
    with (
      patch('test_suite_mapper.get_ignored_paths', return_value=[]),
      patch(
        'test_suite_mapper.find_nested_repos_and_symlinks', return_value=[]
      ),
      contextlib.redirect_stderr(f),
    ):
      suites = test_suite_mapper.run_heavyweight_resolution(
        pathlib.Path('/src'),
        'abc123f',
        'out/Default',
        pathlib.Path('chrome/browser/foo.cc'),
      )
    self.assertIsNone(suites)

  def test_run_heavyweight_resolution_gn_refs_fail(self):
    self.mock_run.side_effect = [
      MagicMock(returncode=0),  # worktree add
      MagicMock(returncode=0),  # gn gen
      MagicMock(
        returncode=1, stdout='', stderr='GN refs error'
      ),  # gn refs fail
      MagicMock(returncode=0),  # worktree remove
    ]
    self.fs.create_dir('/src')
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='target_os = "linux"\n'
    )
    f = io.StringIO()
    with (
      patch('test_suite_mapper.get_ignored_paths', return_value=[]),
      patch(
        'test_suite_mapper.find_nested_repos_and_symlinks', return_value=[]
      ),
      contextlib.redirect_stderr(f),
    ):
      suites = test_suite_mapper.run_heavyweight_resolution(
        pathlib.Path('/src'),
        'abc123f',
        'out/Default',
        pathlib.Path('chrome/browser/foo.cc'),
      )
    self.assertIsNone(suites)

  def test_run_heavyweight_resolution_missing_isolate_map(self):
    self.mock_run.side_effect = [
      MagicMock(returncode=0),  # worktree add
      MagicMock(returncode=0),  # gn gen
      MagicMock(returncode=0, stdout='//foo:bar\n', stderr=''),  # gn refs
      MagicMock(returncode=0),  # worktree remove
    ]
    self.fs.create_dir('/src')
    self.fs.create_file(
      '/src/out/Default/args.gn', contents='target_os = "linux"\n'
    )
    f = io.StringIO()
    with (
      patch('test_suite_mapper.get_ignored_paths', return_value=[]),
      patch(
        'test_suite_mapper.find_nested_repos_and_symlinks', return_value=[]
      ),
      contextlib.redirect_stderr(f),
    ):
      suites = test_suite_mapper.run_heavyweight_resolution(
        pathlib.Path('/src'),
        'abc123f',
        'out/Default',
        pathlib.Path('chrome/browser/foo.cc'),
      )
    self.assertIsNone(suites)
    self.assertIn('Isolate map not found in worktree', f.getvalue())

  def test_load_isolate_map_for_args_with_file(self):
    content = "{'suite1': {'label': '//foo:bar'}}"
    self.fs.create_file('/src/map.pyl', contents=content)
    res = test_suite_mapper._load_isolate_map_for_args(
      '/src', '/src/map.pyl', None, False
    )
    self.assertEqual(res, {'//foo:bar': 'suite1'})

    with self.assertRaises(SystemExit):
      test_suite_mapper._load_isolate_map_for_args(
        '/src', '/src/missing.pyl', None, False
      )

  def test_load_isolate_map_for_args_revision(self):
    with self.assertRaises(SystemExit):
      test_suite_mapper._load_isolate_map_for_args(
        '/src', None, 'abc123f', False
      )

    res = test_suite_mapper._load_isolate_map_for_args(
      '/src', None, 'abc123f', True
    )
    self.assertIsNone(res)

  def test_load_isolate_map_for_args_default(self):
    with patch(
      'test_suite_mapper.load_isolate_map', return_value={'a': 'b'}
    ) as mock_load:
      res = test_suite_mapper._load_isolate_map_for_args(
        '/src', None, None, False
      )
      self.assertEqual(res, {'a': 'b'})
      mock_load.assert_called_once_with('/src')

  def test_check_and_warn_stability_stable(self):
    with patch(
      'test_suite_mapper.check_graph_stability', return_value=(True, [])
    ):
      stable = test_suite_mapper._check_and_warn_stability(
        '/src', 'abc123f', pathlib.Path('foo.cc'), ['//foo']
      )
      self.assertTrue(stable)

  def test_check_and_warn_stability_unstable(self):
    with patch(
      'test_suite_mapper.check_graph_stability',
      return_value=(False, ['BUILD.gn']),
    ):
      f = io.StringIO()
      with contextlib.redirect_stderr(f):
        stable = test_suite_mapper._check_and_warn_stability(
          '/src', 'abc123f', pathlib.Path('foo.cc'), ['//foo']
        )
      self.assertFalse(stable)
      self.assertIn(
        'Warning: The GN graph or the file may have changed', f.getvalue()
      )
      self.assertIn('BUILD.gn', f.getvalue())

    with patch(
      'test_suite_mapper.check_graph_stability',
      return_value=(False, ['f1', 'f2', 'f3', 'f4', 'f5', 'f6']),
    ):
      f = io.StringIO()
      with contextlib.redirect_stderr(f):
        stable = test_suite_mapper._check_and_warn_stability(
          '/src', 'abc123f', pathlib.Path('foo.cc'), ['//foo']
        )
      self.assertFalse(stable)
      self.assertIn('f1, f2, f3, f4, f5 and 1 more', f.getvalue())

  def test_should_run_heavyweight(self):
    self.assertTrue(
      test_suite_mapper._should_run_heavyweight(
        '/src', 'abc123f', pathlib.Path('foo.cc'), ['//foo'], True, True, []
      )
    )

    self.assertFalse(
      test_suite_mapper._should_run_heavyweight(
        '/src', 'abc123f', pathlib.Path('foo.cc'), ['//foo'], False, True, []
      )
    )

    with patch(
      'test_suite_mapper.load_isolate_map', return_value={'//foo': 'suite'}
    ):
      self.assertFalse(
        test_suite_mapper._should_run_heavyweight(
          '/src',
          'abc123f',
          pathlib.Path('foo.cc'),
          ['//foo'],
          False,
          False,
          ['suite'],
        )
      )

    with patch(
      'test_suite_mapper.load_isolate_map', return_value={'//foo': 'suite2'}
    ):
      f = io.StringIO()
      with contextlib.redirect_stderr(f):
        res = test_suite_mapper._should_run_heavyweight(
          '/src',
          'abc123f',
          pathlib.Path('foo.cc'),
          ['//foo'],
          False,
          False,
          ['suite1'],
        )
      self.assertTrue(res)
      self.assertIn('Difference detected', f.getvalue())

  @patch('test_suite_mapper.run_heavyweight_resolution')
  def test_execute_heavyweight_resolution_success(self, mock_resolve):
    mock_resolve.return_value = ['suite1']
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      res = test_suite_mapper._execute_heavyweight_resolution(
        '/src', 'abc123f', 'out/Default', pathlib.Path('foo.cc'), ['suite1']
      )
    self.assertEqual(res, ['suite1'])

    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      res = test_suite_mapper._execute_heavyweight_resolution(
        '/src', 'abc123f', 'out/Default', pathlib.Path('foo.cc'), ['suite2']
      )
    self.assertEqual(res, ['suite1'])
    self.assertIn('Results corrected', f.getvalue())

  @patch('test_suite_mapper.run_heavyweight_resolution')
  def test_execute_heavyweight_resolution_fail(self, mock_resolve):
    mock_resolve.return_value = None
    f = io.StringIO()
    with contextlib.redirect_stderr(f):
      res = test_suite_mapper._execute_heavyweight_resolution(
        '/src', 'abc123f', 'out/Default', pathlib.Path('foo.cc'), ['suite1']
      )
    self.assertEqual(res, ['suite1'])
    self.assertIn('Falling back to lightweight results', f.getvalue())

  def test_execute_heavyweight_resolution_no_rev(self):
    with self.assertRaises(SystemExit):
      test_suite_mapper._execute_heavyweight_resolution(
        '/src', None, 'out/Default', pathlib.Path('foo.cc'), []
      )


if __name__ == '__main__':
  unittest.main()