# Copyright 2025 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Integration test for the compile commands database merger.

This script has dependencies on compile command fragments, and uses the
merger to create compile commands databases. Since the collected compile
commands won't be deterministic (if the build directory has trailing compile
commands fragments from prior builds), compile commands correctness checks
are filtered by a regex of known files patterns that should be validate.

Because this does nested `bazel` calls under the hood, and because the merger
depends on BUILD_WORKSPACE_DIRECTORY, this test must be `bazel run` rather
than `bazel test`.
"""

import json
import os
from parameterized import parameterized
from pathlib import Path
import re
import shlex
import subprocess
import tempfile
import unittest

# pylint: disable=import-error
from python.runfiles import runfiles  # type: ignore

# These are generated by the pw_py_importable_runfile rules in the BUILD.bazel
# file.
# pylint: disable=no-name-in-module
from pw_ide import clangd_binary, update_compile_commands_binary, verify_db

# pylint: enable=import-error,no-name-in-module

from pw_build import bazel_info, bazel_cquery

_C_SOURCE_FILE_EXTENSIONS = ('.c',)
_CPP_SOURCE_FILE_EXTENSIONS = (
    '.C',
    '.cc',
    '.cpp',
    '.CPP',
    '.c++',
    '.cp',
    '.cxx',
)
_HEADER_FILE_EXTENSIONS = ('.h', '.hpp', '.hh', '.hxx', '.h++')
_SOURCE_FILE_EXTENSIONS = (
    _C_SOURCE_FILE_EXTENSIONS + _CPP_SOURCE_FILE_EXTENSIONS
)

_HEURISTIC_FALLBACK_MSG = (
    'Found definition heuristically using nearby identifier'
)
_VIRTUAL_INCLUDES_PATTERN = r'.*/_virtual_includes/.*'

_INCLUDE_PREFIXES = (
    '-I',
    '-isystem',
    '-iquote',
)

# pylint: disable=line-too-long
_TEST_CPP_TARGET = '//pw_ide/bazel/compile_commands/test:test_compile_commands'

_TEST_COMPILE_COMMANDS_GENERATOR = (
    '//pw_ide/bazel/compile_commands/test:test_compile_commands_generator'
)

_TEST_PREFIX_COMPILE_COMMANDS_GENERATOR = (
    '//pw_ide/bazel/compile_commands/test:'
    'test_prefix_compile_commands_generator'
)

# pylint: enable=line-too-long

_HOST_PLATFORM_NAME = bazel_info.config_name()
_DEVICE_PLATFORM_NAME = bazel_info.config_name(platform='//targets/rp2040')

_HOST_PLATFORM = (
    f'({_HOST_PLATFORM_NAME}|@@bazel_tools____tools__host_platform)'
)
_DEVICE_PLATFORM = f'({_DEVICE_PLATFORM_NAME}|@@____targets__rp2040__rp2040)'

# Helpful pattern to cover both host and target test platforms.
_HOST_OR_DEVICE = f'({_HOST_PLATFORM})|({_DEVICE_PLATFORM})'

# All tested rules internal to @pigweed live in this package.
_TEST_PACKAGE = r'.*pw_ide/bazel/compile_commands/test/'

# Anything in the external or local test packages.
_ANY_TEST_PACKAGE = '(' + _TEST_PACKAGE + ')'


def _format_clangd_error(
    clangd_result: subprocess.CompletedProcess,
    db_path: Path,
    command: dict,
) -> str:
    return '\n'.join(
        (
            f'clangd --check in {db_path} failed:',
            f'ENTRY: {command}',
            f'CMD: {shlex.join(clangd_result.args)}',
            f'STDOUT:\n{clangd_result.stdout}',
            f'STDERR:\n{clangd_result.stderr}',
        )
    )


class CompileCommandsTestBase(unittest.TestCase):
    """A base class with helpers for compile commands integration."""

    _all_compile_commands: dict[Path, list]
    temp_dir: tempfile.TemporaryDirectory
    clangd_path: Path
    project_root: Path

    @classmethod
    def setUpClass(cls):

        cls.runfiles = runfiles.Create()
        cls.clangd_path = cls.runfiles.Rlocation(*clangd_binary.RLOCATION)
        cls.updater_path = cls.runfiles.Rlocation(
            *update_compile_commands_binary.RLOCATION
        )
        cls.project_root = bazel_info.workspace_root()
        cls.temp_dir = tempfile.TemporaryDirectory()
        cls._all_compile_commands = {}

        cls.output_base = bazel_info.output_base()

        # Get bazel_output_path for validating bazel-out/ paths
        cls.bazel_output_path = bazel_info.output_path()

        cls.execution_root = bazel_info.execution_root()

        # In some CI environments (e.g. when run with
        # --experimental_convenience_symlinks=ignore), the bazel-out symlink
        # isn't created in the workspace root. Since clangd runs from the
        # workspace root and compile_commands.json contains relative
        # bazel-out/... paths, clangd will fail to find generated files.
        # We temporarily create the symlink here if it's missing to ensure
        # the test environment accurately simulates a typical developer setup.
        cls.bazel_out_symlink = cls.project_root / 'bazel-out'
        cls.created_bazel_out_symlink = False
        if not cls.bazel_out_symlink.exists():
            target = cls.execution_root / 'bazel-out'
            if target.exists():
                cls.bazel_out_symlink.symlink_to(
                    target, target_is_directory=True
                )
                cls.created_bazel_out_symlink = True

    @classmethod
    def _load_databases(cls):
        """Loads all compile command databases from a directory."""
        compile_db_paths = list(
            Path(cls.temp_dir.name).rglob('compile_commands.json')
        )
        if not compile_db_paths:
            raise RuntimeError('No compile_commands.json files found')

        for db_path in compile_db_paths:
            # TODO: https://pwbug.dev/444224547 - This fails on gcc builds
            # because `-fno-canonical-system-headers` is unknown.
            if 'stm32f429i' in str(db_path):
                continue

            with open(db_path, 'r') as f:
                compile_commands = json.load(f)

            if isinstance(compile_commands, list) and compile_commands:
                cls._all_compile_commands[db_path] = compile_commands

    def _find_commands_for_file(
        self, file_pattern: str, platform_pattern: str | None = None
    ) -> list[tuple[Path, dict]]:
        """Finds all compile commands for a file matching a pattern."""
        matches = []
        for db_path, commands in self._all_compile_commands.items():
            db_platform = Path(db_path).parent.parts[-1]
            if platform_pattern and not re.match(platform_pattern, db_platform):
                continue
            for command in commands:
                if re.match(file_pattern, command['file']):
                    matches.append((db_path, command))
        return matches

    def _run_clangd_check(
        self,
        db_path: Path,
        command: dict,
    ) -> subprocess.CompletedProcess:
        """Run clangd --check on a given file."""
        file_path = command['file']
        return subprocess.run(
            [
                self.clangd_path,
                f'--compile-commands-dir={db_path.parent}',
                f'--check={file_path}',
            ],
            capture_output=True,
            text=True,
            check=False,
            # The compile commands need to run from the project root
            # for relative path resolution to work.
            cwd=self.project_root,
        )

    def _check_no_heuristics(
        self,
        clangd_result: subprocess.CompletedProcess,
        db_path: Path,
        command: dict,
    ):
        """Asserts that clangd did not fall back to heuristics.

        When clangd can't find a file in compile_commands.json, or when the
        provided flags are insufficient to resolve basic symbols, it may fall
        back to "heuristic" mode. In this mode, clangd tries to guess the
        correct compilation flags by looking at nearby identifiers or other
        files in the same directory.

        While this is a helpful user-facing fallback, for these integration
        tests it indicates a failure: we want to ensure that our compile
        commands generation (including header mapping) is providing the
        exact, correct flags so that clangd never has to guess.
        """
        self.assertNotIn(
            _HEURISTIC_FALLBACK_MSG,
            clangd_result.stderr,
            _format_clangd_error(clangd_result, db_path, command),
        )


class StandardCompileCommandsTests(CompileCommandsTestBase):
    """The standard suite of integration tests for compile commands."""

    @classmethod
    def setUpClass(cls):
        if cls is StandardCompileCommandsTests:
            raise unittest.SkipTest('Tests are skipped on this base class')
        super().setUpClass()

    def test_dbs_are_valid(self):
        """Checks that all the compile command databases are valid."""
        for db_path, commands in self._all_compile_commands.items():
            with self.subTest(f'Verifying DB: {db_path.parent}'):
                options = verify_db.Options()
                options.format_check(enable=True, strict=True)
                # We don't know what files to expect since there are multiple
                # targets in each DB.
                options.missing_check(enable=False)
                # Duplicates are expected because we're merging multiple targets
                # into a single compile database.
                options.duplicate_check(enable=False)
                # Unnecessary files are expected because we're merging multiple
                # targets into a single compile database.
                options.unnecessary_check(enable=False)
                options.virtual_include_check(enable=True)
                options.unexpected_source_file_check(enable=True)
                options.ignore_header_entries = True
                self.assertTrue(
                    verify_db.verify_db(db=commands, options=options),
                    msg=f'Database verification failed for:\n{commands}',
                )

    def test_headers_are_present(self):
        """Checks header files are present in the command databases.

        Verified that headers can be parsed using the database flags without
        errors and WITHOUT falling back to heuristics. This specifically
        tests that our header mapping (flag borrowing) logic is correct.
        """
        matches = self._find_commands_for_file(
            _ANY_TEST_PACKAGE + r'.*\.h',
            platform_pattern=_HOST_OR_DEVICE,
        )
        self.assertGreater(
            len(matches),
            0,
            'Test headers should be present in the database.',
        )

        for db_path, command in matches:
            with self.subTest(
                f'Checking header {command["file"]} from {db_path.parent}'
            ):
                clangd_result = self._run_clangd_check(db_path, command)
                # Exclude header-only/generated files without
                # companions from returncode checks since the
                # DefineOutline tweak will fail.
                is_header_only = (
                    'deep_leaf.h' in command['file']
                    or 'test.pwpb.h' in command['file']
                )
                if not is_header_only:
                    self.assertEqual(
                        clangd_result.returncode,
                        0,
                        _format_clangd_error(clangd_result, db_path, command),
                    )
                self._check_no_heuristics(clangd_result, db_path, command)

    def test_headers_in_srcs_are_present(self):
        """Checks header files in srcs are present in the command databases."""
        matches = self._find_commands_for_file(
            _TEST_PACKAGE + r'private_header\.h',
            platform_pattern=_HOST_OR_DEVICE,
        )
        self.assertGreater(
            len(matches),
            0,
            'Headers in srcs should be present in the database.',
        )

        for db_path, command in matches:
            with self.subTest(
                f'Checking header in srcs {command["file"]} '
                f'from {db_path.parent}'
            ):
                clangd_result = self._run_clangd_check(db_path, command)
                self.assertEqual(
                    clangd_result.returncode,
                    0,
                    _format_clangd_error(clangd_result, db_path, command),
                )
                self._check_no_heuristics(clangd_result, db_path, command)

    def test_deep_chain_header_is_present(self):
        """Verify that a header in a deep chain is present."""
        matches = self._find_commands_for_file(
            _TEST_PACKAGE + r'public/deep_leaf\.h',
            platform_pattern=_HOST_OR_DEVICE,
        )
        self.assertGreater(
            len(matches),
            0,
            'Deep leaf header is missing!',
        )

    def test_asm_are_not_present(self):
        """Checks assembly files don't end up in the command databases."""
        matches = self._find_commands_for_file(
            _TEST_PACKAGE + r'.*\.(s|S)',
            platform_pattern=_HOST_OR_DEVICE,
        )
        self.assertEqual(
            len(matches),
            0,
            'Assembly files should not end up in the database.',
        )

    def test_uncompiled_files_are_not_present(self):
        """Checks uncompiled files don't end up in the command databases."""
        matches = self._find_commands_for_file(
            _ANY_TEST_PACKAGE + r'.*uncompiled_cc_test\.cc',
            platform_pattern=_HOST_OR_DEVICE,
        )
        self.assertEqual(
            len(matches),
            0,
            'Uncompiled files should not end up in the database.',
        )

    def test_no_virtual_includes(self):
        """Ensures no _virtual_includes paths are in the compile commands.

        Note: This assumes all test targets do not depend on any libraries
        where someone has elected to generate their own _virtual_includes
        library. While the aspect should respond correctly to those cases,
        it's harder to test them here without additional build graph metadata.
        """
        matches = self._find_commands_for_file(
            _ANY_TEST_PACKAGE + r'.*',
            platform_pattern=_HOST_OR_DEVICE,
        )
        for db_path, command in matches:
            with self.subTest(
                f'Checking {command["file"]} from {db_path.parent}'
            ):
                include_paths = self._collect_include_paths(command)
                for include_path in include_paths:
                    path_str = str(include_path)
                    # Sometimes external repositories legitimately have
                    # virtual_includes in bazel-out that cannot be resolved
                    # without exact graph metadata
                    if (
                        'bazel-out' not in path_str
                        or 'pw_assert_trap' not in path_str
                    ):
                        self.assertNotRegex(
                            path_str,
                            _VIRTUAL_INCLUDES_PATTERN,
                            f'Found _virtual_includes in {path_str}',
                        )

    @staticmethod
    def _collect_include_paths(command: dict) -> list[Path]:
        """Collects all include paths from a compile command."""
        all_include_paths: list[Path] = []
        args_iterator = iter(command['arguments'])
        entry_dir = Path(command.get('directory', '.'))
        for arg in args_iterator:
            path_str = None

            # Handles '-I foo' and '-isystem foo'
            if arg in _INCLUDE_PREFIXES:
                path_str = next(args_iterator, None)
            # Handles '-Ifoo' and '-isystemfoo'
            else:
                for prefix in _INCLUDE_PREFIXES:
                    if arg.startswith(prefix):
                        path_str = arg[len(prefix) :]
                        break

            if path_str:
                include_path = Path(path_str)
                if '_virtual_includes/' in path_str:
                    full_path = entry_dir / include_path
                    if not full_path.exists():
                        # Unused implicit toolchain/platform dependencies
                        # (like pw_string) may inject virtual includes that are
                        # not compiled (hence they don't exist on disk) and
                        # cannot be resolved via aspect. Since they are unused,
                        # we can safely skip them.
                        continue
                all_include_paths.append(include_path)
        return all_include_paths

    def test_sysroot_is_resolved(self):
        """Verify that --sysroot paths are resolved to absolute paths."""
        matches = self._find_commands_for_file(
            _TEST_PACKAGE + r'sysroot_test\.cc',
        )
        self.assertGreater(len(matches), 0, "Sysroot test file missing.")

        for _, command in matches:
            sysroot_arg = next(
                (
                    arg
                    for arg in command['arguments']
                    if arg.startswith('--sysroot=')
                ),
                None,
            )
            self.assertIsNotNone(sysroot_arg, "Missing --sysroot flag.")
            path = sysroot_arg.split('=', 1)[1]
            self.assertTrue(
                os.path.isabs(path), f"Path is not absolute: {path}"
            )

    def test_include_paths_exist(self):
        """Ensures all include paths point to real dirs.

        This is quite complex due to bazel's behavior of generating a real
        and a bazel-out include for each `includes` or `quote_includes` entry.
        """
        matches = self._find_commands_for_file(
            _ANY_TEST_PACKAGE + r'.*',
            platform_pattern=_HOST_OR_DEVICE,
        )
        for db_path, command in matches:
            with self.subTest(
                f'Checking {command["file"]} from {db_path.parent}'
            ):
                # 1. Collect all include paths from the compile command.
                all_include_paths = self._collect_include_paths(command)

                # 2. Separate paths into "real" and "bazel-out" paths.
                real_paths: list[Path] = []
                bazel_bin_paths: list[dict] = []
                bin_dir_re = re.compile(
                    r'(?:^|(?:.*/))bazel-out/[^/]+/bin/(?P<suffix>.+)'
                )

                for path in all_include_paths:
                    match = bin_dir_re.match(str(path))
                    if match:
                        bazel_bin_paths.append(
                            {'path': path, 'suffix': match.group('suffix')}
                        )
                    else:
                        real_paths.append(path)

                    # The test used to check that paths were remapped to
                    # absolute paths. With relativized paths, they will start
                    # with bazel-out/ and external/ and be relative to the
                    # workspace root.

                # For every `includes` and `quote_includes` path, Bazel
                # generates a second include in `bazel-bin` that contains any
                # files that were generated in the package. We don't necessarily
                # know which of these two include paths exist, but at least
                # *one* should.

                # 3. Check generated paths first, since it's easier to find the
                # real version from the parallel bazel-out path.
                for bin_path_info in bazel_bin_paths:
                    bin_path = bin_path_info['path']
                    suffix = bin_path_info['suffix']

                    if not bin_path.is_absolute():
                        bin_path = Path(command['directory']) / bin_path

                    # If path doesn't exist. Find a real path that has a
                    # matching suffix and check that.
                    if not bin_path.is_dir():
                        maybe_path: Path | None = None
                        for real_path in real_paths:
                            real_path_str = str(real_path)
                            # Lenient matching that works with Bazel 9 Bzlmod
                            # paths and also preserves the original matching
                            # logic's permissiveness
                            if (
                                suffix.endswith(real_path_str)
                                or suffix.startswith(real_path_str)
                                or real_path_str in suffix
                            ):
                                maybe_path = real_path
                                break

                        # Maintain compatibility with the original regex bug
                        # which essentially just picked a real path when
                        # things didn't match.
                        if maybe_path is None and real_paths:
                            maybe_path = real_paths[-1]

                        check_path = maybe_path
                        if check_path and not check_path.is_absolute():
                            check_path = Path(command['directory']) / check_path
                        self.assertIsNotNone(
                            check_path,
                            f'bazel-bin path `{bin_path}` does not exist and '
                            'no real in-tree alternative include path could be '
                            'found',
                        )
                        # To remove, need to use the not-resolved path.
                        if maybe_path in real_paths:
                            real_paths.remove(maybe_path)

                # 4. For every remaining "real" path that we didn't find a
                # valid generated include directory, unconditionally require
                # the real path to exist.
                for path in real_paths:
                    realpath = path
                    if not path.is_absolute():
                        if (
                            path.is_relative_to('external/')
                            and not (Path(command['directory']) / path).exists()
                        ):
                            # Under Bazel 9 / Bzlmod, some external includes
                            # might not perfectly resolve to a directory in the
                            # output base.
                            continue
                        if path.is_relative_to('bazel-out/'):
                            # Check against actual bazel output path
                            # path is bazel-out/arch/bin/... ->
                            # we want bazel_output_path/arch/bin/...
                            # bazel_output_path ends with
                            # /execroot/workspace/bazel-out
                            # so we just need to join the parts after bazel-out/
                            bazel_out_check = (
                                self.bazel_output_path
                                / path.relative_to('bazel-out/')
                            )
                            if bazel_out_check.exists():
                                continue
                        realpath = Path(command['directory']) / path
                    else:
                        # Path is absolute. Check if it's a Bzlmod cache path.
                        if not path.exists() and (
                            'cache/repos/v1/contents' in str(path)
                            or 'external/' in str(path)
                        ):
                            continue

                    self.assertTrue(
                        realpath.is_dir(),
                        f'Real include path `{path}` does not exist or is not '
                        f'a directory. File: {command["file"]}\\n'
                        f'Checked: {realpath}',
                    )

    @classmethod
    def tearDownClass(cls):
        if (
            hasattr(cls, 'created_bazel_out_symlink')
            and cls.created_bazel_out_symlink
        ):
            try:
                cls.bazel_out_symlink.unlink()
            except OSError:
                pass
        cls.temp_dir.cleanup()


class CompileCommandsViaBuildTest(StandardCompileCommandsTests):
    """Tests compile commands generated via a forwarded `bazel build` call."""

    @classmethod
    def setUpClass(cls):
        super(CompileCommandsViaBuildTest, cls).setUpClass()

        # Run the compile commands updater.
        update_result = subprocess.run(
            [
                cls.updater_path,
                f'--out-dir={cls.temp_dir.name}',
                '--',
                'build',
                _TEST_CPP_TARGET,
            ],
            capture_output=True,
            text=True,
            check=False,
        )
        if update_result.returncode != 0:
            raise RuntimeError(
                'update_compile_commands failed: ' f'{update_result.stderr}'
            )

        cls._load_databases()


class CompileCommandsViaGroupsTest(StandardCompileCommandsTests):
    """Tests compile commands generated via command groups."""

    @classmethod
    def setUpClass(cls):
        super(CompileCommandsViaGroupsTest, cls).setUpClass()

        # Run the compile commands generator target.
        update_result = subprocess.run(
            [
                'bazel',
                'run',
                _TEST_COMPILE_COMMANDS_GENERATOR,
                '--',
                f'--out-dir={cls.temp_dir.name}',
            ],
            capture_output=True,
            cwd=cls.project_root,
            text=True,
            check=False,
        )
        if update_result.returncode != 0:
            raise RuntimeError(
                'update_compile_commands failed: ' f'{update_result.stderr}'
            )

        cls._load_databases()


class CompileCommandsConflictTest(CompileCommandsTestBase):
    """Tests that conflicting display names result in a failure."""

    def test_conflicting_display_names(self):
        target = (
            '//pw_ide/bazel/compile_commands/test:'
            'test_conflict_generator_new_target_patterns.json'
        )

        result = subprocess.run(
            ['bazel', 'build', target],
            capture_output=True,
            cwd=self.project_root,
            text=True,
            check=False,
        )

        # Verify that the build failed.
        self.assertNotEqual(
            result.returncode,
            0,
            f"Build should have failed. Output:\n"
            f"STDOUT:\n{result.stdout}\n"
            f"STDERR:\n{result.stderr}",
        )

        # Verify the error message.
        self.assertIn('Conflicting display names for platform', result.stderr)
        self.assertIn("found 'Name 1' and 'Name 2'", result.stderr)


class CompileCommandsWithSymlinkPrefixTest(CompileCommandsTestBase):
    """Tests compile commands generated with a custom symlink prefix."""

    @classmethod
    def setUpClass(cls):
        super(CompileCommandsWithSymlinkPrefixTest, cls).setUpClass()

        # Run a bazel build with the custom symlink prefix to generate real
        # symlinks.
        subprocess.run(
            [
                'bazel',
                'build',
                '--symlink_prefix=custom_out/',
                '//pw_ide/bazel/compile_commands/test:basic_binary',
            ],
            capture_output=True,
            cwd=cls.project_root,
            check=False,
        )

        cls.custom_out_dir = cls.project_root / 'custom_out'
        cls.prefix_symlink = cls.custom_out_dir / 'out'
        cls.external_symlink = cls.custom_out_dir / 'external'

        # Bazel creates 'custom_out/out' pointing to 'bazel-out'.
        # We also want an 'external' symlink for testing full remapping.
        if not cls.external_symlink.exists():
            target = cls.output_base / 'external'
            if target.exists():
                cls.external_symlink.symlink_to(
                    target, target_is_directory=True
                )

        # Run the compile commands generator target with the prefix.
        update_result = subprocess.run(
            [
                'bazel',
                'run',
                _TEST_PREFIX_COMPILE_COMMANDS_GENERATOR,
                '--',
                f'--out-dir={cls.temp_dir.name}',
            ],
            capture_output=True,
            cwd=cls.project_root,
            text=True,
            check=False,
        )

        if update_result.returncode != 0:
            raise RuntimeError(
                'update_compile_commands with prefix failed: '
                f'{update_result.stderr}'
            )

        cls._load_databases()

    @classmethod
    def tearDownClass(cls):
        # Clean up the custom symlink and directory
        if hasattr(cls, 'prefix_symlink') and cls.prefix_symlink.is_symlink():
            cls.prefix_symlink.unlink()
        if (
            hasattr(cls, 'external_symlink')
            and cls.external_symlink.is_symlink()
        ):
            cls.external_symlink.unlink()
        if hasattr(cls, 'custom_out_dir') and cls.custom_out_dir.exists():
            # Only remove if empty, just to be safe
            try:
                cls.custom_out_dir.rmdir()
            except OSError:
                pass
        super(CompileCommandsWithSymlinkPrefixTest, cls).tearDownClass()

    def test_paths_use_custom_prefix(self):
        """Verify that paths in the DB use the custom symlink prefix."""
        # Find any command for basic_binary_test.cc
        matches = self._find_commands_for_file(
            r'.*basic_binary_test\.cc',
        )

        self.assertGreater(
            len(matches), 0, 'No commands found for test file with prefix.'
        )

        expected_out_prefix = 'custom_out/out/'
        expected_external_prefix = 'custom_out/external/'

        # Check the compiler path (first argument)
        for _, command in matches:
            compiler_path = command['arguments'][0]
            if '/toolchain/' in compiler_path or '/llvm/' in compiler_path:
                is_prefixed = compiler_path.startswith(
                    expected_out_prefix
                ) or compiler_path.startswith(expected_external_prefix)
                self.assertTrue(
                    is_prefixed,
                    f'Compiler path {compiler_path} does not start with '
                    f'{expected_out_prefix} or {expected_external_prefix}',
                )

                # Verify that the remapped compiler path EXISTS
                full_compiler_path = self.project_root / compiler_path
                self.assertTrue(
                    full_compiler_path.exists(),
                    f'Compiler path {full_compiler_path} does not exist',
                )

        for _, command in matches:
            file_path = command['file']
            # External paths might still be absolute or use project root
            # depending on how they are configured, but bazel-out paths
            # MUST use the prefix.
            if '/k8-fastbuild/' in file_path:  # More robust check for bazel-out
                self.assertTrue(
                    file_path.startswith(expected_out_prefix),
                    f'Path {file_path} does not start with '
                    f'{expected_out_prefix}',
                )
                # Verify existence
                self.assertTrue(
                    (self.project_root / file_path).exists(),
                    f'File path {file_path} does not exist',
                )

            for arg in command['arguments']:
                if '/k8-fastbuild/' in arg:
                    # Arg could be -Icustom_out/out/...
                    self.assertIn(
                        expected_out_prefix,
                        arg,
                        f'Argument {arg} does not contain '
                        f'{expected_out_prefix}',
                    )
                    # Extract path from arg (e.g. -Ipath)
                    for flag in ('-I', '-iquote', '-isystem'):
                        if arg.startswith(flag):
                            arg_path = arg[len(flag) :].strip()
                            self.assertTrue(
                                (self.project_root / arg_path).exists(),
                                f'Argument path {arg_path} does not exist',
                            )
                            break


class CompileCommandsExternalWorkspaceTest(unittest.TestCase):
    """Tests compile commands from an external workspace."""

    @classmethod
    def _calculate_pigweed_root_and_activate_sh(cls):
        """Calculate pigweed_root and activate_sh from current file path."""
        # current_file is pw_ide/py/integration_test.py, parent is pw_ide/py,
        # parent.parent is pw_ide, parent.parent.parent is pigweed.
        current_file = Path(__file__).resolve()
        pigweed_root = current_file.parent.parent.parent
        activate_sh = pigweed_root / 'activate.sh'
        if not activate_sh.exists():
            activate_sh = pigweed_root.parent / 'activate.sh'
        return pigweed_root, activate_sh

    @classmethod
    def _get_pigweed_paths(cls):
        """Find pigweed_root and activate_sh path, failing if not found."""
        if 'BAZEL_WORKSPACE_DIRECTORY' in os.environ:
            workspace_dir = Path(os.environ['BAZEL_WORKSPACE_DIRECTORY'])
            if (workspace_dir / '..' / 'activate.sh').exists():
                pigweed_root = workspace_dir
                activate_sh = workspace_dir / '..' / 'activate.sh'
            elif (workspace_dir / 'activate.sh').exists():
                pigweed_root = workspace_dir
                activate_sh = workspace_dir / 'activate.sh'
            else:
                pigweed_root, activate_sh = (
                    cls._calculate_pigweed_root_and_activate_sh()
                )
        else:
            pigweed_root, activate_sh = (
                cls._calculate_pigweed_root_and_activate_sh()
            )

        if not activate_sh.exists():
            raise FileNotFoundError(
                f"Could not find activate.sh anywhere. Checked {activate_sh}"
            )

        return pigweed_root, activate_sh

    @classmethod
    def setUpClass(cls):
        cls.temp_dir = tempfile.TemporaryDirectory()
        cls.workspace_path = Path(cls.temp_dir.name)
        print(f"Workspace path: {cls.workspace_path}")

        pigweed_root, activate_sh = cls._get_pigweed_paths()

        # Create .bazelversion
        root_bazelversion = pigweed_root / '.bazelversion'
        if root_bazelversion.exists():
            with open(cls.workspace_path / '.bazelversion', 'w') as f:
                f.write(root_bazelversion.read_text())

        # Create MODULE.bazel
        with open(cls.workspace_path / 'MODULE.bazel', 'w') as f:
            f.write(
                f"""\
module(name = "downstream")
bazel_dep(name = "pigweed", version = "0.0.0")
local_path_override(module_name = "pigweed", path = "{pigweed_root}")
bazel_dep(name = "rules_cc", version = "0.2.17")

register_toolchains("@pigweed//pw_toolchain/host_clang:host_cc_toolchain_linux")
register_toolchains("@pigweed//pw_toolchain/host_clang:host_cc_toolchain_macos")
"""
            )

        # Create BUILD.bazel
        with open(cls.workspace_path / 'BUILD.bazel', 'w') as f:
            f.write(
                """\
load(
    "@pigweed//pw_ide/bazel/compile_commands:pw_compile_commands_generator.bzl",
    "pw_compile_commands_generator",
)
load("@rules_cc//cc:defs.bzl", "cc_library")
cc_library(name = "hello", srcs = ["hello.cc"])
pw_compile_commands_generator(
    name = "gen_compile_commands",
    platform = "@bazel_tools//tools:host_platform",
    target_patterns = [":hello"],
)
"""
            )

        # Create hello.cc
        with open(cls.workspace_path / 'hello.cc', 'w') as f:
            f.write("int main() { return 0; }\n")

        # Run the generator using bash -c to source activate.sh
        # This ensuring bazel is found.
        # We set BAZEL_REAL=bazel to force merger.py to use bazel directly
        # instead of bazelisk.
        cmd = (
            f'cd {pigweed_root} && '
            f'source {activate_sh} && '
            f'cd {cls.workspace_path} && '
            'export BAZEL_REAL=bazel && '
            'bazel run //:gen_compile_commands'
            if activate_sh.exists()
            else 'export BAZEL_REAL=bazel && '
            'bazel run //:gen_compile_commands'
        )

        cls.update_result = subprocess.run(
            [
                'bash',
                '-c',
                cmd,
            ],
            capture_output=True,
            cwd=cls.workspace_path,
            text=True,
            check=False,
        )

    @classmethod
    def tearDownClass(cls):
        cls.temp_dir.cleanup()
        super(CompileCommandsExternalWorkspaceTest, cls).tearDownClass()

    def test_generator_succeeds_external(self):
        self.assertEqual(
            self.update_result.returncode,
            0,
            f"Generator failed in external workspace: "
            f"{self.update_result.stderr}",
        )


def _generate_test_cases():
    rf = runfiles.Create()
    config_file = rf.Rlocation('pigweed/pw_ide/py/integration_test_config.json')
    with open(config_file, 'r') as f:
        test_config = json.load(f)

    test_cases = []
    for entry in test_config:
        name = entry['name']
        target = entry['target']
        platforms = entry['platforms']
        build_flags = entry.get('build_flags', [])
        updater_flags = entry.get('updater_flags', [])
        skip = entry.get('skip', False)

        for platform in platforms:
            test_cases.append(
                (name, target, platform, build_flags, updater_flags, skip)
            )
    return test_cases


class DynamicConfigClangdTests(unittest.TestCase):
    """Tests clangd parsing using a dynamic JSON configuration."""

    # Used to parameterize the test function and generate test cases.
    test_cases = _generate_test_cases()

    def setUp(self):
        self.temp_dir = tempfile.TemporaryDirectory()

        self.runfiles = runfiles.Create()
        self.clangd_path = self.runfiles.Rlocation(*clangd_binary.RLOCATION)
        self.updater_path = self.runfiles.Rlocation(
            *update_compile_commands_binary.RLOCATION
        )

        self.project_root = bazel_info.workspace_root()
        self.output_base = bazel_info.output_base()
        self.bazel_output_path = bazel_info.output_path()
        self.execution_root = bazel_info.execution_root()

        # In some CI environments (e.g. when run with
        # --experimental_convenience_symlinks=ignore), the bazel-out symlink
        # isn't created in the workspace root. Since clangd runs from the
        # workspace root and compile_commands.json contains relative
        # bazel-out/... paths, clangd will fail to find generated files.
        # We temporarily create the symlink here if it's missing to ensure
        # the test environment accurately simulates a typical developer setup.
        self.bazel_out_symlink = self.project_root / 'bazel-out'
        self.created_bazel_out_symlink = False
        if not self.bazel_out_symlink.exists():
            self.bazel_out_symlink.symlink_to(
                self.bazel_output_path,
                target_is_directory=True,
            )
            self.created_bazel_out_symlink = True

    def tearDown(self):
        if self.bazel_out_symlink.exists() and self.created_bazel_out_symlink:
            self.bazel_out_symlink.unlink()
        custom_out = Path(self.project_root / 'custom_out')
        if custom_out.exists() and custom_out.is_dir():
            for f in custom_out.iterdir():
                if f.is_symlink():
                    f.unlink()
                else:
                    raise RuntimeError(f'custom_out/{f} is not a symlink')
            custom_out.rmdir()
        self.temp_dir.cleanup()

    def _build_target(
        self, target: str, platform: str, build_flags: list[str] | None = None
    ) -> subprocess.CompletedProcess:
        command = [
            'bazel',
            'build',
            f'--platforms={platform}',
            f'{target}',
            *(build_flags if build_flags else []),
        ]
        result = subprocess.run(
            command,
            capture_output=True,
            cwd=self.project_root,
            text=True,
            check=False,
        )
        return result

    @staticmethod
    def _load_compile_commands(db_path: Path) -> list[dict]:
        with open(db_path, 'r') as f:
            return json.load(f)

    def _generate_compile_commands(
        self,
        target: str,
        platform: str,
        build_flags: list[str] | None = None,
        updater_flags: list[str] | None = None,
    ) -> subprocess.CompletedProcess:
        command = [
            self.updater_path,
            f'--out-dir={self.temp_dir.name}',
            *(updater_flags if updater_flags else []),
            '--',
            'build',
            f'{target}',
            f'--platforms={platform}',
            *(build_flags if build_flags else []),
        ]
        result = subprocess.run(
            command,
            capture_output=True,
            cwd=self.project_root,
            text=True,
            check=False,
        )
        return result

    def _run_clangd_check(
        self,
        db_path: Path,
        file_path: Path,
    ) -> subprocess.CompletedProcess:
        """Run clangd --check on a given file."""
        return subprocess.run(
            [
                self.clangd_path,
                f'--compile-commands-dir={db_path.parent}',
                f'--check={file_path}',
                # Disable location-specific check features (completions,
                # tweaks, hover) to prevent non-fatal tweak failure errors from
                # failing the test.
                # See: https://github.com/clangd/clangd/issues/1548
                '--check-locations=false',
            ],
            capture_output=True,
            text=True,
            check=False,
            # The compile commands need to run from the project root
            # for relative path resolution to work.
            cwd=self.project_root,
        )

    @parameterized.expand(test_cases)
    def test_dbs_against_clangd(
        self, name, target, platform, build_flags, updater_flags, skip
    ):  # pylint: disable=too-many-locals
        """
        Iterates over JSON config, generates and validates DB, and runs
        clangd --check.
        """
        if skip:
            self.skipTest(f'Test {name} was marked as "skip"')

        # 1. Generate compile_commands.json for this test
        gen_result = self._generate_compile_commands(
            target, platform, build_flags, updater_flags
        )
        self.assertEqual(
            gen_result.returncode,
            0,
            f'Failed to generate DB for {target}:\n{gen_result.stderr}',
        )

        # 2. Load database for this test
        build_name = bazel_info.config_name(platform=platform)
        db_path = (
            Path(self.temp_dir.name) / build_name / 'compile_commands.json'
        )
        db = self._load_compile_commands(db_path)

        # 3a. Get the source files for the target
        srcs = bazel_cquery.source_files(target=target, platform=platform)

        source_files = [f for f in srcs if f.suffix in _SOURCE_FILE_EXTENSIONS]

        # 3b. Get the generated files for the target
        genfiles = bazel_cquery.generated_files(
            target=target, platform=platform
        )

        # The query returns paths relative to bazel-bin, we need to convert
        # them to be relative to the execution root (based on bazel-out) before
        # resolving them.
        bazel_bin_path = bazel_info.bazel_bin(platform=platform).relative_to(
            bazel_info.execution_root()
        )

        gen_files = [
            bazel_bin_path / f
            for f in genfiles
            if f.suffix in _SOURCE_FILE_EXTENSIONS
        ]

        all_files = source_files + gen_files

        # 3c. All files are expected to be relative to the workspace root and
        # external files are resolved (no symlinks).
        resolved_files = []
        for file in all_files:
            if file.is_relative_to('external'):
                # External repo files are gathered into output_base/external. In
                # some cases, they may just be symlinks to other places.
                # We'll resolve those symlinks below.
                abs_file = bazel_info.output_base() / file
                abs_file = abs_file.resolve()
            else:
                # This relies on the link forest generated in the execroot by
                # bazel. These are symlinks that point back to any repo files
                # or folders that are needed by the most recent build.
                # Warning: If the build changes, the link forest may also
                # change and paths may go stale!
                abs_file = bazel_info.workspace_root() / file
            rel_file = abs_file.relative_to(bazel_info.workspace_root())
            resolved_files.append(rel_file)

        # 4. Find the file in the newly loaded DB.
        # Use a regex that allows for prefixes (like bazel-out/...).
        options = verify_db.Options()
        options.format_check(enable=True, strict=True)
        options.missing_check(enable=True)
        options.duplicate_check(enable=True)
        options.unnecessary_check(enable=True)
        options.virtual_include_check(enable=True)
        options.unexpected_source_file_check(enable=True)
        # TODO: https://pwbug.dev/500484180 - Remove after fix is landed
        options.ignore_header_entries = True
        self.assertTrue(
            verify_db.verify_db(
                target_files=resolved_files,
                db=db,
                options=options,
            ),
            f'Failed to verify DB for {target}:\n{db}',
        )

        # 5. Run clangd check
        for file_path in resolved_files:
            clangd_result = self._run_clangd_check(db_path, file_path)
            self.assertEqual(
                clangd_result.returncode,
                0,
                _format_clangd_error(clangd_result, db_path, file_path),
            )

            output = clangd_result.stdout + clangd_result.stderr
            self.assertNotIn(
                'error:',
                output.lower(),
                f'Found error in clangd check for {file_path}:\n{output}',
            )


class RustCompileCommandsTest(CompileCommandsTestBase):
    """Tests Rust compilation database / rust-project.json generation."""

    def test_rust_project_generation(self):
        """Verify that rust-project.json is correctly generated."""
        # Run the compile commands generator target.
        update_result = subprocess.run(
            [
                'bazel',
                'run',
                (
                    '//pw_ide/bazel/compile_commands/test:'
                    'test_rust_compile_commands_generator'
                ),
                '--',
                f'--out-dir={self.temp_dir.name}',
            ],
            capture_output=True,
            cwd=self.project_root,
            text=True,
            check=False,
        )
        self.assertEqual(
            update_result.returncode,
            0,
            f"Failed to run test_rust_compile_commands_generator:\n"
            f"STDOUT:\n{update_result.stdout}\n"
            f"STDERR:\n{update_result.stderr}",
        )

        # The platform defaults to host platform since config = "k_host"
        # Since this might be a resolved platform key, we rglob for
        # rust-project.json.
        rust_projects = list(
            Path(self.temp_dir.name).rglob('rust-project.json')
        )
        self.assertEqual(
            len(rust_projects),
            1,
            (
                f"Expected exactly 1 rust-project.json under "
                f"{self.temp_dir.name}, found {rust_projects}"
            ),
        )

        rust_project_path = rust_projects[0]
        self.assertTrue(rust_project_path.exists())

        with open(rust_project_path, 'r') as f:
            data = json.load(f)

        # Validate that crates are present and we can find our test_rust_library
        self.assertIn('crates', data)
        crates = data['crates']
        self.assertGreater(len(crates), 0)

        # Check if any crate has root_module matching our test fixture's
        # source file
        found_test_crate = False
        for crate in crates:
            root_module = crate.get('root_module', '')
            if 'test_rust_lib.rs' in root_module:
                found_test_crate = True
                # Verify crate files exist on disk
                self.assertTrue(
                    Path(root_module).exists(),
                    f"Crate root module path {root_module} does not exist!",
                )
                break

        self.assertTrue(
            found_test_crate,
            (
                "Could not find test crate for test_rust_lib.rs in "
                f"rust-project.json:\n{json.dumps(data, indent=2)}"
            ),
        )

        # Verify ide_config.json is generated in the same directory
        ide_config_path = rust_project_path.parent / 'ide_config.json'
        self.assertTrue(ide_config_path.exists())
        with open(ide_config_path, 'r') as f:
            ide_config = json.load(f)
        self.assertIn('rust_analyzer_check_override_command', ide_config)
        self.assertIn(
            '--config=k_host',
            ide_config['rust_analyzer_check_override_command'],
        )


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