diff --git a/components/bt/common/ble_log/extension/log_compression/CMakeLists.txt b/components/bt/common/ble_log/extension/log_compression/CMakeLists.txt index fba9d33cfea..cc5962dbcce 100644 --- a/components/bt/common/ble_log/extension/log_compression/CMakeLists.txt +++ b/components/bt/common/ble_log/extension/log_compression/CMakeLists.txt @@ -185,18 +185,6 @@ if(LOG_COMPRESSED_MODULE) endforeach() string(REPLACE "|" ";" LOG_COMPRESSED_MODULE_CODE_PATH "${MODULE_CODE_PATH}") - # Some header files of NIMBLE are not added to include_dirs, - # but rely on relative path searches. This will cause the header - # files to be found due to the change in the source code location - # after using the log compression scheme. - # Therefore, these paths are added to include_dirs here to avoid - # unfinished compilation errors. - if(CONFIG_BT_NIMBLE_ENABLED) - list(APPEND include_dirs - "host/nimble/nimble/nimble/host/src" - "host/nimble/nimble/nimble/host/store/config/src") - endif() - if(BLE_COMPRESSED_LIB_LOG_BUILD) execute_process(COMMAND ${BLE_PYTHON_EXECUTABLE} ${PYTHON_SCRIPT} compress diff --git a/components/bt/common/ble_log/extension/log_compression/scripts/ble_log_compress.py b/components/bt/common/ble_log/extension/log_compression/scripts/ble_log_compress.py index fc49799812e..f7fdeb44e58 100644 --- a/components/bt/common/ble_log/extension/log_compression/scripts/ble_log_compress.py +++ b/components/bt/common/ble_log/extension/log_compression/scripts/ble_log_compress.py @@ -20,6 +20,7 @@ This script processes Bluetooth source files to compress logging statements. import argparse import enum import importlib.util +import json import logging import os import re @@ -130,6 +131,10 @@ LINE_MACROS = { '__LINE__', } +MIRRORED_INCLUDE_SUFFIXES = frozenset({'.h', '.inc'}) +MIRRORED_INCLUDE_MANIFEST = '.mirrored_includes.json' +MIRRORED_INCLUDE_MANIFEST_VERSION = 1 + class ARG_SIZE_TYPE(enum.IntEnum): U32 = 0 @@ -687,6 +692,151 @@ class LogCompressor: return generated_macros + @staticmethod + def _require_path_within(path: Path, root: Path, label: str) -> None: + try: + path.relative_to(root) + except ValueError as error: + raise ValueError(f'{label} escapes allowed root {root}: {path}') from error + + def _validated_mirror_destination(self, relative: Path) -> Path: + if relative.is_absolute() or '..' in relative.parts or relative.suffix not in MIRRORED_INCLUDE_SUFFIXES: + raise ValueError(f'Invalid mirrored include path: {relative}') + + mirror_root = self.bt_compressed_srcs_path.resolve(strict=True) + destination = self.bt_compressed_srcs_path / relative + resolved_destination = destination.resolve(strict=False) + self._require_path_within(resolved_destination, mirror_root, 'Mirrored include destination') + return destination + + def _discover_local_includes(self) -> dict[Path, Path]: + try: + code_base = self.code_base_path.resolve(strict=True) + except (FileNotFoundError, NotADirectoryError) as error: + raise ValueError(f'Invalid CODE_BASE_PATH: {self.code_base_path}') from error + + if not code_base.is_dir(): + raise ValueError(f'Invalid CODE_BASE_PATH: {code_base}') + + includes: dict[Path, Path] = {} + for info in self.module_info.values(): + for configured_path in info['code_path']: + candidate = Path(configured_path) + if not candidate.is_absolute(): + candidate = code_base / candidate + try: + module_root = candidate.resolve(strict=True) + except (FileNotFoundError, NotADirectoryError) as error: + raise ValueError(f'Invalid module code path: {configured_path}') from error + if not module_root.is_dir(): + raise ValueError(f'Invalid module code path: {configured_path}') + try: + module_root.relative_to(code_base) + except ValueError as error: + raise ValueError(f'Module code path escapes CODE_BASE_PATH: {module_root}') from error + + for source in module_root.rglob('*'): + if source.suffix not in MIRRORED_INCLUDE_SUFFIXES or not source.is_file(): + continue + resolved_source = source.resolve(strict=True) + self._require_path_within(resolved_source, module_root, 'Local include source') + relative = source.relative_to(code_base) + includes[relative] = source + return includes + + def _load_mirrored_include_manifest(self) -> set[Path]: + manifest_path = self.bt_compressed_srcs_path / MIRRORED_INCLUDE_MANIFEST + if manifest_path.is_symlink(): + raise ValueError(f'Invalid mirrored include manifest: {manifest_path}') + if not manifest_path.exists(): + return set() + if not manifest_path.is_file(): + raise ValueError(f'Invalid mirrored include manifest: {manifest_path}') + + try: + payload = json.loads(manifest_path.read_text(encoding='utf-8')) + except (OSError, json.JSONDecodeError) as error: + raise ValueError(f'Invalid mirrored include manifest: {manifest_path}') from error + if ( + not isinstance(payload, dict) + or payload.get('version') != MIRRORED_INCLUDE_MANIFEST_VERSION + or not isinstance(payload.get('files'), list) + ): + raise ValueError(f'Invalid mirrored include manifest: {manifest_path}') + + relative_paths: set[Path] = set() + for value in payload['files']: + if not isinstance(value, str): + raise ValueError(f'Invalid mirrored include manifest entry: {value!r}') + relative = Path(value) + self._validated_mirror_destination(relative) + relative_paths.add(relative) + return relative_paths + + def _remove_empty_mirror_parents(self, destination: Path) -> None: + mirror_root = self.bt_compressed_srcs_path.resolve(strict=True) + parent = destination.parent + while parent != mirror_root: + try: + parent.rmdir() + except OSError: + break + parent = parent.parent + + def _remove_stale_mirrored_includes(self, stale_paths: set[Path]) -> None: + for relative in sorted(stale_paths, key=lambda path: len(path.parts), reverse=True): + destination = self._validated_mirror_destination(relative) + if destination.is_dir(): + raise ValueError(f'Managed mirrored include is a directory: {destination}') + if destination.exists() or destination.is_symlink(): + destination.unlink() + LOGGER.info(f'Removed stale mirrored include: {destination}') + self._remove_empty_mirror_parents(destination) + + def _write_mirrored_include_manifest(self, relative_paths: set[Path]) -> None: + manifest_path = self.bt_compressed_srcs_path / MIRRORED_INCLUDE_MANIFEST + temporary_path = manifest_path.with_name(f'{manifest_path.name}.tmp') + if manifest_path.is_symlink() or (manifest_path.exists() and not manifest_path.is_file()): + raise ValueError(f'Invalid mirrored include manifest path: {manifest_path}') + if temporary_path.is_symlink() or temporary_path.exists(): + if temporary_path.is_dir(): + raise ValueError(f'Invalid manifest temporary path: {temporary_path}') + temporary_path.unlink() + + payload = { + 'version': MIRRORED_INCLUDE_MANIFEST_VERSION, + 'files': sorted(path.as_posix() for path in relative_paths), + } + try: + temporary_path.write_text(f'{json.dumps(payload, indent=2)}\n', encoding='utf-8') + os.replace(temporary_path, manifest_path) + finally: + if temporary_path.exists() or temporary_path.is_symlink(): + temporary_path.unlink() + + def mirror_local_includes(self) -> None: + """Mirror unchanged local headers for generated C source include lookup.""" + self.bt_compressed_srcs_path.mkdir(parents=True, exist_ok=True) + previous_paths = self._load_mirrored_include_manifest() + includes = self._discover_local_includes() + current_paths = set(includes) + newly_created_paths: set[Path] = set() + try: + for relative, source in sorted(includes.items(), key=lambda item: item[0].as_posix()): + destination = self._validated_mirror_destination(relative) + destination.parent.mkdir(parents=True, exist_ok=True) + if not destination.exists() and not destination.is_symlink(): + newly_created_paths.add(relative) + shutil.copy2(source, destination) + LOGGER.info(f'Mirrored local include: {source} -> {destination}') + self._remove_stale_mirrored_includes(previous_paths - current_paths) + self._write_mirrored_include_manifest(current_paths) + except Exception: + # Do not leave newly introduced headers unowned by the preserved + # manifest when copying, cleanup, or manifest replacement fails. + self._remove_stale_mirrored_includes(newly_created_paths) + raise + def prepare_source_files(self, srcs: list[str]) -> None: """ Prepare source files for processing. @@ -884,6 +1034,9 @@ class LogCompressor: config_path = self.build_dir / 'ble_log/module_info.yml' self.load_config(str(config_path), modules) + # Preserve local quoted-include semantics for generated C sources. + self.mirror_local_includes() + # Initialize database db_path = self.build_dir / self.config.get('db_path', 'log_db') db_manager = LogDBManager( diff --git a/components/bt/common/ble_log/extension/log_compression/tests/README.md b/components/bt/common/ble_log/extension/log_compression/tests/README.md index 0350b801e3d..84f27b3f372 100644 --- a/components/bt/common/ble_log/extension/log_compression/tests/README.md +++ b/components/bt/common/ble_log/extension/log_compression/tests/README.md @@ -19,6 +19,7 @@ tests/ ├── test_macro_generation.py # Bluedroid and Mesh module macro generators ├── test_pipeline_e2e.py # End-to-end .c -> .h with golden file comparison ├── test_incremental.py # Incremental compression scenarios +├── test_header_mirroring.py # Local .h/.inc mirror, manifest safety, pipeline isolation ├── update_golden.py # Script to regenerate golden expected files └── fixtures/ ├── c_sources/ # Test input C files @@ -145,6 +146,14 @@ Tests caching, re-runs, and config change behavior. | `TestAddNewFile` | New file IDs continue from `max_id + 1` | | `TestConfigChange` | Config change triggers `SOURCE_LOG_UPDATE_FULL` | +### test_header_mirroring.py — Local Header Mirroring + +Validates that enabled compression-module `.h` and `.inc` files are copied +unchanged into `.compressed_srcs` with paths relative to `CODE_BASE_PATH`. +Coverage includes overlapping module roots, modified and deleted headers, +manifest validation, path-escape rejection, production pipeline wiring, and +removal of the NimBLE global include-directory workaround. + ## Golden File Workflow The `fixtures/expected/` directory contains known-good header outputs. End-to-end tests compare generated output against these files (with copyright year normalized). diff --git a/components/bt/common/ble_log/extension/log_compression/tests/test_header_mirroring.py b/components/bt/common/ble_log/extension/log_compression/tests/test_header_mirroring.py new file mode 100644 index 00000000000..2435133cfbf --- /dev/null +++ b/components/bt/common/ble_log/extension/log_compression/tests/test_header_mirroring.py @@ -0,0 +1,252 @@ +# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD +# SPDX-License-Identifier: Apache-2.0 +"""Tests for unchanged local-header mirroring into .compressed_srcs.""" + +import json +import os +import shutil +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import test_utils # noqa: F401 — must be first to set up sys.path +from ble_log_compress import LogCompressor +from test_utils import BLUEDROID_SCRIPT +from test_utils import BLUEDROID_TAGS +from test_utils import write_yaml_config + + +class _HeaderMirrorTestCase(unittest.TestCase): + def setUp(self) -> None: + self.tmp = Path(tempfile.mkdtemp()) + self.code_base = self.tmp / 'code_base' + self.mirror = self.tmp / 'build' / 'ble_log' / '.compressed_srcs' + self.code_base.mkdir(parents=True) + self.mirror.mkdir(parents=True) + + self.compressor = LogCompressor() + self.compressor.code_base_path = self.code_base + self.compressor.bt_compressed_srcs_path = self.mirror + self.compressor.module_info = {'BLE_HOST': {'code_path': ['module']}} + + def tearDown(self) -> None: + shutil.rmtree(self.tmp, ignore_errors=True) + + def write_source(self, relative: str, content: str) -> Path: + path = self.code_base / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content) + return path + + +class TestHeaderDiscoveryAndCopy(_HeaderMirrorTestCase): + def test_recursively_mirrors_h_and_inc_only(self) -> None: + private_h = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n') + config_inc = self.write_source('module/shared/config.inc', '#define CONFIG_VALUE 9\n') + self.write_source('module/src/ignored.c', 'int ignored;\n') + self.write_source('module/CMakeLists.txt', 'set(ignored yes)\n') + + self.compressor.mirror_local_includes() + + mirrored_h = self.mirror / private_h.relative_to(self.code_base) + mirrored_inc = self.mirror / config_inc.relative_to(self.code_base) + self.assertEqual(mirrored_h.read_bytes(), private_h.read_bytes()) + self.assertEqual(mirrored_inc.read_bytes(), config_inc.read_bytes()) + self.assertEqual(mirrored_h.stat().st_mtime_ns, private_h.stat().st_mtime_ns) + self.assertFalse((self.mirror / 'module/src/ignored.c').exists()) + self.assertFalse((self.mirror / 'module/CMakeLists.txt').exists()) + + manifest = json.loads((self.mirror / '.mirrored_includes.json').read_text()) + self.assertEqual( + manifest, + { + 'version': 1, + 'files': ['module/shared/config.inc', 'module/src/private.h'], + }, + ) + + def test_overlapping_module_roots_copy_each_header_once(self) -> None: + self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n') + self.compressor.module_info = { + 'BLE_HOST': {'code_path': ['module']}, + 'BLE_ISO': {'code_path': ['module/src']}, + } + + with mock.patch('ble_log_compress.shutil.copy2', wraps=shutil.copy2) as copy2: + self.compressor.mirror_local_includes() + + self.assertEqual(copy2.call_count, 1) + + def test_rejects_missing_and_non_directory_module_roots(self) -> None: + cases = ('missing', 'not_a_directory') + self.write_source('not_a_directory', 'not a directory\n') + + for code_path in cases: + with self.subTest(code_path=code_path): + self.compressor.module_info = {'BLE_HOST': {'code_path': [code_path]}} + with self.assertRaisesRegex(ValueError, 'Invalid module code path'): + self.compressor.mirror_local_includes() + + def test_rejects_module_root_outside_code_base(self) -> None: + outside = self.tmp / 'outside' + outside.mkdir() + link = self.code_base / 'module' + try: + link.symlink_to(outside, target_is_directory=True) + except OSError as error: + self.skipTest(f'Directory symlinks unavailable: {error}') + + with self.assertRaisesRegex(ValueError, 'escapes CODE_BASE_PATH'): + self.compressor.mirror_local_includes() + + +class TestHeaderMirrorSynchronization(_HeaderMirrorTestCase): + def test_modified_header_updates_content_and_timestamp(self) -> None: + source = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n') + self.compressor.mirror_local_includes() + destination = self.mirror / 'module/src/private.h' + + updated_mtime = source.stat().st_mtime_ns + 2_000_000_000 + source.write_text('#define PRIVATE_VALUE 11\n') + os.utime(source, ns=(updated_mtime, updated_mtime)) + self.compressor.mirror_local_includes() + + self.assertEqual(destination.read_bytes(), source.read_bytes()) + self.assertEqual(destination.stat().st_mtime_ns, source.stat().st_mtime_ns) + + def test_deleted_header_is_removed_without_touching_unmanaged_outputs(self) -> None: + source = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n') + self.compressor.mirror_local_includes() + generated_c = self.mirror / 'module/src/generated.c' + unmanaged_h = self.mirror / 'module/src/unmanaged.h' + generated_c.write_text('int generated;\n') + unmanaged_h.write_text('#define UNMANAGED 1\n') + + source.unlink() + self.compressor.mirror_local_includes() + + self.assertFalse((self.mirror / 'module/src/private.h').exists()) + self.assertEqual(generated_c.read_text(), 'int generated;\n') + self.assertEqual(unmanaged_h.read_text(), '#define UNMANAGED 1\n') + manifest = json.loads((self.mirror / '.mirrored_includes.json').read_text()) + self.assertEqual(manifest, {'version': 1, 'files': []}) + + def test_rejects_malformed_manifest_before_copying(self) -> None: + source = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n') + (self.mirror / '.mirrored_includes.json').write_text('{not-json') + + with self.assertRaisesRegex(ValueError, 'Invalid mirrored include manifest'): + self.compressor.mirror_local_includes() + + self.assertFalse((self.mirror / source.relative_to(self.code_base)).exists()) + + def test_rejects_unsafe_manifest_path_without_deleting_outside_file(self) -> None: + outside = self.mirror.parent / 'outside.h' + outside.write_text('#define OUTSIDE 1\n') + manifest = { + 'version': 1, + 'files': ['../outside.h'], + } + (self.mirror / '.mirrored_includes.json').write_text(json.dumps(manifest)) + + with self.assertRaisesRegex(ValueError, 'Invalid mirrored include path'): + self.compressor.mirror_local_includes() + + self.assertEqual(outside.read_text(), '#define OUTSIDE 1\n') + + def test_manifest_replace_failure_preserves_previous_manifest(self) -> None: + self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n') + self.compressor.mirror_local_includes() + manifest_path = self.mirror / '.mirrored_includes.json' + previous_manifest = manifest_path.read_bytes() + second = self.write_source('module/src/second.h', '#define SECOND_VALUE 2\n') + + with mock.patch('ble_log_compress.os.replace', side_effect=OSError('replace failed')): + with self.assertRaisesRegex(OSError, 'replace failed'): + self.compressor.mirror_local_includes() + + self.assertEqual(manifest_path.read_bytes(), previous_manifest) + second.unlink() + self.compressor.mirror_local_includes() + self.assertFalse((self.mirror / second.relative_to(self.code_base)).exists()) + + def test_rejects_header_symlink_leaving_module_root(self) -> None: + outside = self.write_source('outside.h', '#define OUTSIDE 1\n') + link = self.code_base / 'module/src/external.h' + link.parent.mkdir(parents=True, exist_ok=True) + try: + link.symlink_to(outside) + except OSError as error: + self.skipTest(f'File symlinks unavailable: {error}') + + with self.assertRaisesRegex(ValueError, 'Local include source escapes allowed root'): + self.compressor.mirror_local_includes() + + +class TestHeaderMirrorPipeline(unittest.TestCase): + def setUp(self) -> None: + self.tmp = Path(tempfile.mkdtemp()) + + def tearDown(self) -> None: + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_main_mirrors_same_directory_and_parent_relative_includes(self) -> None: + code_base = self.tmp / 'code_base' + build_dir = self.tmp / 'build' + compressed_srcs = build_dir / 'ble_log' / '.compressed_srcs' + source_dir = code_base / 'test_src' / 'src' + shared_dir = code_base / 'test_src' / 'shared' + source_dir.mkdir(parents=True) + shared_dir.mkdir(parents=True) + source = source_dir / 'local_include.c' + private_h = source_dir / 'private.h' + config_inc = shared_dir / 'config.inc' + source.write_text( + '#include "private.h"\n' + '#include "../shared/config.inc"\n' + 'void local_include(void) { APPL_TRACE_DEBUG("value %d", PRIVATE_VALUE); }\n' + ) + private_h.write_text('#define PRIVATE_VALUE 7\n') + config_inc.write_text('#define CONFIG_VALUE 9\n') + write_yaml_config( + str(self.tmp), + tags=BLUEDROID_TAGS, + script_path=BLUEDROID_SCRIPT, + ) + + argv = [ + 'ble_log_compress.py', + 'compress', + '--srcs', + 'test_src/src/local_include.c', + '--code_base_path', + str(code_base), + '--module', + 'BLE_HOST', + '--build_path', + str(build_dir), + '--compressed_srcs_path', + str(compressed_srcs), + ] + with mock.patch.object(sys, 'argv', argv): + result = LogCompressor().main() + + self.assertEqual(result, 0) + generated_c = compressed_srcs / 'test_src/src/local_include.c' + mirrored_h = generated_c.parent / 'private.h' + mirrored_inc = generated_c.parent / '../shared/config.inc' + self.assertTrue(generated_c.exists()) + self.assertEqual(mirrored_h.read_bytes(), private_h.read_bytes()) + self.assertEqual(mirrored_inc.resolve().read_bytes(), config_inc.read_bytes()) + + def test_cmake_does_not_add_nimble_local_header_directories(self) -> None: + cmake_path = Path(__file__).resolve().parent.parent / 'CMakeLists.txt' + cmake = cmake_path.read_text() + self.assertNotIn('"host/nimble/nimble/nimble/host/src"', cmake) + self.assertNotIn('"host/nimble/nimble/nimble/host/store/config/src"', cmake) + + +if __name__ == '__main__': + unittest.main() diff --git a/components/bt/common/ble_log/extension/log_compression/tests/test_utils.py b/components/bt/common/ble_log/extension/log_compression/tests/test_utils.py index 6ef002afe2d..a576872fe61 100644 --- a/components/bt/common/ble_log/extension/log_compression/tests/test_utils.py +++ b/components/bt/common/ble_log/extension/log_compression/tests/test_utils.py @@ -141,7 +141,8 @@ class PipelineContext: return str(rel_path) def run_compression(self, src_list: list[str]) -> dict[str, list[tuple[int, str]]]: - """Run prepare + compress + header generation. Returns generated macros.""" + """Run mirror + prepare + compress + header generation. Returns generated macros.""" + self.compressor.mirror_local_includes() self.compressor.prepare_source_files(src_list) files_to_process = []