Mirrors/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-09-22 13:01:16 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'fix/ble_log_compression_safety_v5.2' into 'release/v5.2'

Browse Source 
fix(ble_log): fix unaligned access and buffer safety in log compression (5.2)

See merge request espressif/esp-idf!47927
This commit is contained in:
Island
2026-07-23 14:25:24 +08:00
parent 9db78e59b3 984c9940d9
commit 82389b3e96
26 changed files with 2347 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
components/bt/common/ble_log/extension/log_compression/ble_log_compression.c
View File

@@ -16,6 +16,13 @@
#if CONFIG_BLE_COMPRESSED_LOG_ENABLE
#define BLE_CP_TRY_PUSH(expr) do { \
if ((expr) != 0) { \
return -1; \
} \
} while (0)
#define BUF_NAME(name, idx) name##_buffer##idx
#define BUF_MGMT_NAME(name) name##_log_buffer_mgmt
@@ -74,16 +81,16 @@ int ble_compressed_log_cb_get(uint8_t source, ble_cp_log_buffer_mgmt_t **mgmt)
#endif
default:
assert(0 && "Unsupported log source");
break;
return -1;
}
for (int i = 0; i < LOG_CP_MAX_LOG_BUFFER_USED_SIMU; i++) {
if (ble_log_cas_acquire(&(buffer_mgmt[i].busy))) {
*mgmt = &buffer_mgmt[i];
ble_log_cp_push_u8(*mgmt, source);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(*mgmt, source));
if (*last_handle == NULL ||
*last_handle != cur_handle) {
ble_log_cp_push_u8(*mgmt, LOG_HEADER(LOG_TYPE_INFO, LOG_TYPE_INFO_TASK_SWITCH));
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(*mgmt, LOG_HEADER(LOG_TYPE_INFO, LOG_TYPE_INFO_TASK_SWITCH)));
*last_handle = cur_handle;
}
return 0;
@@ -95,7 +102,7 @@ int ble_compressed_log_cb_get(uint8_t source, ble_cp_log_buffer_mgmt_t **mgmt)
static inline int ble_compressed_log_buffer_free(ble_cp_log_buffer_mgmt_t *mgmt)
{
#if BLE_LOG_CP_CONTENT_CHECK_ENBALE
#if BLE_LOG_CP_CONTENT_CHECK_ENABLE
memset(mgmt->buffer, BLE_LOG_CP_CONTENT_CHECK_VAL, mgmt->idx);
#endif
mgmt->idx = 0;
@@ -107,9 +114,14 @@ static inline
int ble_log_compressed_hex_print_internal(ble_cp_log_buffer_mgmt_t *mgmt, uint32_t log_index, size_t args_cnt, va_list args)
{
uint8_t arg_type = 0;
uint16_t header_size = 1 + 2 + (args_cnt + 1) / 2; // header + log_index + size_info
ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_HEX_ARGS, args_cnt));
ble_log_cp_push_u16(mgmt, log_index);
if (ble_log_cp_buffer_safe_check(mgmt, header_size)) {
return -1;
}
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_HEX_ARGS, args_cnt)));
BLE_CP_TRY_PUSH(ble_log_cp_push_u16(mgmt, log_index));
uint8_t size_info_idx = mgmt->idx;
uint8_t *cur = &(mgmt->buffer)[mgmt->idx];
uint8_t size_info = 0;
@@ -117,13 +129,13 @@ int ble_log_compressed_hex_print_internal(ble_cp_log_buffer_mgmt_t *mgmt, uint32
for (size_t i = 0; i < args_cnt; i++) {
if (i % 2) {
arg_type = va_arg(args, size_t);
ble_log_cp_push_u8(mgmt, size_info|arg_type);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, size_info|arg_type));
size_info = 0;
cur++;
} else {
arg_type = va_arg(args, size_t);
if (i == args_cnt - 1) {
ble_log_cp_push_u8(mgmt, arg_type);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, arg_type << 4));
} else {
size_info = arg_type << 4;
}
@@ -148,17 +160,17 @@ int ble_log_compressed_hex_print_internal(ble_cp_log_buffer_mgmt_t *mgmt, uint32
uint32_t u32v = va_arg(args, size_t);
if (likely(u32v)) {
if (u32v <= 0xff) {
ble_log_cp_push_u8(mgmt, 3);
ble_log_cp_push_u8(mgmt, u32v);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, 3));
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, u32v));
ble_log_cp_update_half_byte(mgmt, size_info_idx + i/2, ARG_SIZE_TYPE_LZU32, !(i%2));
break;
} else if (u32v <= 0xffff) {
ble_log_cp_push_u8(mgmt, 2);
ble_log_cp_push_u16(mgmt, u32v);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, 2));
BLE_CP_TRY_PUSH(ble_log_cp_push_u16(mgmt, u32v));
ble_log_cp_update_half_byte(mgmt, size_info_idx + i/2, ARG_SIZE_TYPE_LZU32, !(i%2));
break;
} else {
ble_log_cp_push_u32(mgmt, u32v);
BLE_CP_TRY_PUSH(ble_log_cp_push_u32(mgmt, u32v));
}
} else {
ble_log_cp_update_half_byte(mgmt, size_info_idx + i/2, ARG_SIZE_TYPE_AZU32, !(i%2));
@@ -168,7 +180,7 @@ int ble_log_compressed_hex_print_internal(ble_cp_log_buffer_mgmt_t *mgmt, uint32
uint64_t u64v = va_arg(args, uint64_t);
if (likely(u64v)) {
if (unlikely(u64v >> 48)) {
ble_log_cp_push_u64(mgmt, u64v);
BLE_CP_TRY_PUSH(ble_log_cp_push_u64(mgmt, u64v));
} else {
uint32_t tmpv = 0;
uint8_t lz = 0;
@@ -179,30 +191,33 @@ int ble_log_compressed_hex_print_internal(ble_cp_log_buffer_mgmt_t *mgmt, uint32
tmpv = u64v >> 32;
}
lz += __builtin_clz(tmpv) / 8;
ble_log_cp_push_u8(mgmt, lz);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, lz));
switch (8-lz) {
case 5:
ble_log_cp_push_u32(mgmt, (uint32_t)u64v);
BLE_CP_TRY_PUSH(ble_log_cp_push_u32(mgmt, (uint32_t)u64v));
[[fallthrough]];
case 1:
ble_log_cp_push_u8(mgmt, (uint8_t)tmpv);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, (uint8_t)tmpv));
break;
case 6:
ble_log_cp_push_u32(mgmt, (uint32_t)u64v);
BLE_CP_TRY_PUSH(ble_log_cp_push_u32(mgmt, (uint32_t)u64v));
[[fallthrough]];
case 2:
ble_log_cp_push_u16(mgmt, (uint16_t)tmpv);
BLE_CP_TRY_PUSH(ble_log_cp_push_u16(mgmt, (uint16_t)tmpv));
break;
case 7:
ble_log_cp_push_u32(mgmt, (uint32_t)u64v);
BLE_CP_TRY_PUSH(ble_log_cp_push_u32(mgmt, (uint32_t)u64v));
[[fallthrough]];
case 3:
ble_log_cp_push_u8(mgmt, (uint8_t)tmpv);
ble_log_cp_push_u16(mgmt, (uint16_t)(tmpv >> 8));
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, (uint8_t)tmpv));
BLE_CP_TRY_PUSH(ble_log_cp_push_u16(mgmt, (uint16_t)(tmpv >> 8)));
break;
case 4:
BLE_CP_TRY_PUSH(ble_log_cp_push_u32(mgmt, (uint32_t)u64v));
break;
default:
assert(0);
break;
return -1;
}
ble_log_cp_update_half_byte(mgmt, size_info_idx + i/2, ARG_SIZE_TYPE_LZU64, !(i%2));
}
@@ -212,12 +227,16 @@ int ble_log_compressed_hex_print_internal(ble_cp_log_buffer_mgmt_t *mgmt, uint32
break;
case ARG_SIZE_TYPE_STR:
char *str_p = (char *)va_arg(args, char *);
ble_log_cp_push_buf(mgmt, (const uint8_t *)str_p, strlen(str_p) + 1);
if (str_p) {
BLE_CP_TRY_PUSH(ble_log_cp_push_buf(mgmt, (const uint8_t *)str_p, strlen(str_p) + 1));
} else {
BLE_CP_TRY_PUSH(ble_log_cp_push_buf(mgmt, (const uint8_t *)"(null str)", sizeof("(null str)")));
}
break;
default:
printf("Invalid size %d\n", arg_type);
assert(0);
break;
return -1;
}
}
return 0;
@@ -246,8 +265,8 @@ int ble_log_compressed_hex_print(uint8_t source, uint32_t log_index, size_t args
}
if (args_cnt == 0) {
ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_HEX_ARGS, 0));
ble_log_cp_push_u16(mgmt, log_index);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_HEX_ARGS, 0)));
BLE_CP_TRY_PUSH(ble_log_cp_push_u16(mgmt, log_index));
} else {
va_list args;
va_start(args, args_cnt);
@@ -268,17 +287,17 @@ int ble_log_compressed_hex_print_buf(uint8_t source, uint32_t log_index, uint8_t
return 0;
}
if (buf == NULL && len != 0) {
ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_INFO, LOG_TYPE_INFO_NULL_BUF));
ble_log_cp_push_u16(mgmt, log_index);
if (buf == NULL) {
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_INFO, LOG_TYPE_INFO_NULL_BUF)));
BLE_CP_TRY_PUSH(ble_log_cp_push_u16(mgmt, log_index));
ble_compressed_log_output(source, mgmt->buffer, mgmt->idx);
ble_compressed_log_buffer_free(mgmt);
return 0;
}
ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_HEX_BUF, buf_idx));
ble_log_cp_push_u16(mgmt, log_index);
ble_log_cp_push_buf(mgmt, buf, len);
BLE_CP_TRY_PUSH(ble_log_cp_push_u8(mgmt, LOG_HEADER(LOG_TYPE_HEX_BUF, buf_idx)));
BLE_CP_TRY_PUSH(ble_log_cp_push_u16(mgmt, log_index));
BLE_CP_TRY_PUSH(ble_log_cp_push_buf(mgmt, buf, len));
ble_compressed_log_output(source, mgmt->buffer, mgmt->idx);
ble_compressed_log_buffer_free(mgmt);
return 0;

16
components/bt/common/ble_log/extension/log_compression/include/log_compression/utils.h
View File

@@ -1,5 +1,5 @@
/*
* SPDX-FileCopyrightText: 2025 Espressif Systems (Shanghai) CO LTD
* SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
@@ -8,6 +8,7 @@
#include "ble_log.h"
#include <stdio.h>
#include <string.h>
#define CONCAT(a, b) a##b
#define _CONCAT(a, b) CONCAT(a, b)
@@ -86,7 +87,7 @@ typedef struct {
#define CONTENT_CHECK(idx, buf, except_val, len)
#define LENGTH_CHECK(idx, pbuffer_mgmt) do ( if(unlikely((idx) > (pbuffer_mgmt->len))) assert(0 && "Maximum log buffer length exceeded");) while(0)
#define BLE_LOG_CP_CONTENT_CHECK_ENBALE 0
#define BLE_LOG_CP_CONTENT_CHECK_ENABLE 0
#define BLE_LOG_CP_CONTENT_CHECK_VAL 0x00
static inline int ble_log_cp_buffer_safe_check(ble_cp_log_buffer_mgmt_t *pbuf_mgmt, uint16_t write_len)
@@ -95,7 +96,7 @@ static inline int ble_log_cp_buffer_safe_check(ble_cp_log_buffer_mgmt_t *pbuf_mg
printf("Maximum length of buffer(%p) idx %d write_len %d exceed\n", pbuf_mgmt, pbuf_mgmt->idx, write_len);
return -1;
}
#if BLE_LOG_CP_CONTENT_CHECK_ENBALE
#if BLE_LOG_CP_CONTENT_CHECK_ENABLE
for (int i = pbuf_mgmt->idx; i < pbuf_mgmt->idx + write_len; i++) {
if (pbuf_mgmt->buffer[i] != BLE_LOG_CP_CONTENT_CHECK_VAL) {
printf("The value(%02x) in the buffer does not match the expected(%02x)\n", pbuf_mgmt->buffer[i], BLE_LOG_CP_CONTENT_CHECK_VAL);
@@ -121,8 +122,7 @@ static inline int ble_log_cp_push_u16(ble_cp_log_buffer_mgmt_t *pbuf_mgmt, uint1
if (ble_log_cp_buffer_safe_check(pbuf_mgmt, 2)) {
return -1;
}
uint16_t *p = (uint16_t *)&(pbuf_mgmt->buffer[pbuf_mgmt->idx]);
*p = val;
memcpy(&(pbuf_mgmt->buffer[pbuf_mgmt->idx]), &val, sizeof(val));
pbuf_mgmt->idx+=2;
return 0;
}
@@ -132,8 +132,7 @@ static inline int ble_log_cp_push_u32(ble_cp_log_buffer_mgmt_t *pbuf_mgmt, uint3
if (ble_log_cp_buffer_safe_check(pbuf_mgmt, 4)) {
return -1;
}
uint32_t *p = (uint32_t *)&(pbuf_mgmt->buffer[pbuf_mgmt->idx]);
*p = val;
memcpy(&(pbuf_mgmt->buffer[pbuf_mgmt->idx]), &val, sizeof(val));
pbuf_mgmt->idx+=4;
return 0;
}
@@ -143,8 +142,7 @@ static inline int ble_log_cp_push_u64(ble_cp_log_buffer_mgmt_t *pbuf_mgmt, uint6
if (ble_log_cp_buffer_safe_check(pbuf_mgmt, 8)) {
return -1;
}
uint64_t *p = (uint64_t *)&(pbuf_mgmt->buffer[pbuf_mgmt->idx]);
*p = val;
memcpy(&(pbuf_mgmt->buffer[pbuf_mgmt->idx]), &val, sizeof(val));
pbuf_mgmt->idx+=8;
return 0;
}

47
components/bt/common/ble_log/extension/log_compression/scripts/ble_log_compress.py
View File

@@ -38,6 +38,7 @@ from typing import cast
import tree_sitter_c as tsc
import yaml
from c_format_parse import FormatToken
from c_format_parse import parse_format_string
from inttypes_map import TYPES_MACRO_MAP
from LogDBManager import LogDBManager
@@ -341,14 +342,14 @@ class LogCompressor:
tokens_tuple_map: list[int] = []
need_args = 0
for idx, tk in enumerate(tokens):
if isinstance(tk, tuple):
if isinstance(tk, FormatToken):
tokens_tuple_map.append(idx)
need_args = need_args + 1
if tk[4] == '*': # dynamic width
if tk.width == '*': # dynamic width
need_args = need_args + 1
log_info['hexify'] = False
return log_info
if tk[5] == '*': # dynamic precision
if tk.precision == '*': # dynamic precision
need_args = need_args + 1
log_info['hexify'] = False
return log_info
@@ -359,7 +360,7 @@ class LogCompressor:
raise SyntaxError(f'LogSyntaxError:{node.text.decode("utf-8")}')
# Process each argument
for i, (token, arg_node) in enumerate(zip([t for t in tokens if isinstance(t, tuple)], arguments)):
for i, (token, arg_node) in enumerate(zip([t for t in tokens if isinstance(t, FormatToken)], arguments)):
arg_text = arg_node.text.decode('utf-8')
log_info['arguments'].append((arg_text, arg_node.start_byte, arg_node.end_byte))
@@ -369,13 +370,9 @@ class LogCompressor:
# Handle special identifiers
if arg_text in FUNC_MACROS:
token_list = list(token)
token_list[6] = '@func' # Modify conversion char to special marker
tokens[tokens_tuple_map[i]] = tuple(token_list)
tokens[tokens_tuple_map[i]] = token._replace(conv_char='@func')
elif arg_text in LINE_MACROS:
token_list = list(token)
token_list[6] = '@line'
tokens[tokens_tuple_map[i]] = tuple(token_list)
tokens[tokens_tuple_map[i]] = token._replace(conv_char='@line')
# Handle hex functions
if (
@@ -393,9 +390,7 @@ class LogCompressor:
len_node = abs(hex_func_info[2])
else:
len_node = hex_args.named_children[hex_func_info[2]].text.decode('utf-8')
token_list = list(token)
token_list[6] = f'@hex_func@{buf_node}@{len_node}'
tokens[tokens_tuple_map[i]] = tuple(token_list)
tokens[tokens_tuple_map[i]] = token._replace(conv_char=f'@hex_func@{buf_node}@{len_node}')
log_info['argu_tokens'] = tokens
@@ -423,9 +418,9 @@ class LogCompressor:
raise ValueError(f'Unsupported node in concatenated string: {child.type}')
return ''.join(parts)
def _can_be_hexified(self, token: tuple[int, int, str, str, str, str, str], node: Node) -> bool:
def _can_be_hexified(self, token: FormatToken, node: Node) -> bool:
"""Determine if a node can be represented in hex format."""
if token[-1] != 's':
if token.conv_char != 's':
return True
if node.type == 'identifier' and node.text.decode('utf-8') in FUNC_MACROS:
@@ -485,7 +480,7 @@ class LogCompressor:
if log_info['hexify']:
# Count of arguments that are not special (__func__, __LINE__, etc.)
arg_tokens = [t for t in log_info['argu_tokens'] if isinstance(t, tuple)]
arg_tokens = [t for t in log_info['argu_tokens'] if isinstance(t, FormatToken)]
arg_count = len(arg_tokens)
arguments = []
sizes = []
@@ -498,23 +493,23 @@ class LogCompressor:
[a[0] for a in log_info['arguments'][1:]],
):
# Skip special tokens
if token[6] in ('@func', '@line'):
if token.conv_char in ('@func', '@line'):
arg_count -= 1
continue
# Handle hex function
if token[6].startswith('@hex_func'):
if token.conv_char.startswith('@hex_func'):
if not hex_func:
hex_func = []
hex_func.append(token[6])
hex_func.append(token.conv_char)
arg_count -= 1
continue
arguments.append(argument)
if token[6] == 'f' or token[5] == 'll': # float or long long
if token.conv_char == 'f' or token.length == 'll': # float or long long
sizes.append(f'{int(ARG_SIZE_TYPE.U64)}')
elif token[6] == 's':
elif token.conv_char == 's':
sizes.append(f'{int(ARG_SIZE_TYPE.STR)}')
else:
sizes.append(f'{int(ARG_SIZE_TYPE.U32)}')
@@ -585,16 +580,16 @@ class LogCompressor:
simple_fmt_list: list[str] = []
hex_buffer_cnt = 0
for token in log['argu_tokens']:
if isinstance(token, tuple):
if '@func' in token[6] or '@line' in token[6]:
if isinstance(token, FormatToken):
if '@func' in token.conv_char or '@line' in token.conv_char:
continue
if '@hex_func' in token[6]:
if '@hex_func' in token.conv_char:
simple_fmt_list.append(f'@hex_buffer{hex_buffer_cnt}')
no_buf_fmt += f'@hex_buffer{hex_buffer_cnt}'
hex_buffer_cnt += 1
continue
simple_fmt_list.append(token[2])
no_buf_fmt += token[2]
simple_fmt_list.append(token.full_spec)
no_buf_fmt += token.full_spec
else:
no_buf_fmt += token
simple_fmt_str = ' '.join(simple_fmt_list) if simple_fmt_list else None

47
components/bt/common/ble_log/extension/log_compression/scripts/c_format_parse.py
View File

@@ -8,10 +8,24 @@ Parses C-style format strings and handles argument formatting for log compressio
"""
import struct
from typing import NamedTuple
from typing import Union
def parse_format_string(format_str: str) -> list[Union[str, tuple[int, int, str, str, str, str, str, str]]]:
class FormatToken(NamedTuple):
"""Parsed C format specifier, e.g. ``%08llx`` -> FormatToken(start, end, '%08llx', '0', '8', '', 'll', 'x')."""
start: int # index of '%' in the source string
end: int # index one-past the conversion char
full_spec: str # the raw specifier text, e.g. '%08llx'
flags: str # '-', '+', ' ', '#', '0', or ''
width: str # e.g. '10', '*', or ''
precision: str # e.g. '.2' content (without '.'), '*', or ''
length: str # 'h', 'hh', 'l', 'll', 'j', 'z', 't', or ''
conv_char: str # 'd', 'i', 'u', 'o', 'x', 'X', 'f', 's', etc.
def parse_format_string(format_str: str) -> list[Union[str, FormatToken]]:
"""
Parse a format string into tokens.
@@ -19,10 +33,9 @@ def parse_format_string(format_str: str) -> list[Union[str, tuple[int, int, str,
format_str: C-style format string
Returns:
List of tokens (strings or format spec tuples)
Tuple format: (start, end, full_spec, flags, width, precision, length, conv_char)
List of tokens (literal strings or FormatToken named-tuples)
"""
tokens: list[Union[str, tuple[int, int, str, str, str, str, str, str]]] = []
tokens: list[Union[str, FormatToken]] = []
i = 0
n = len(format_str)
@@ -85,7 +98,7 @@ def parse_format_string(format_str: str) -> list[Union[str, tuple[int, int, str,
conv_char = format_str[i]
i += 1
full_spec = format_str[start:i]
tokens.append((start, i, full_spec, flags, width, precision, length, conv_char))
tokens.append(FormatToken(start, i, full_spec, flags, width, precision, length, conv_char))
else:
# Invalid format spec, treat as literal text
tokens.append(format_str[start:i])
@@ -216,10 +229,8 @@ def parse_compressed_arguments(byte_sequence: bytes, format_str: str) -> str:
arg_index = 0
for token in tokens:
if isinstance(token, tuple):
start, end, full_spec, flags, width, precision, length_mod, conv_char = token
if conv_char == '%':
if isinstance(token, FormatToken):
if token.conv_char == '%':
output.append('%')
else:
if arg_index >= len(args):
@@ -229,19 +240,19 @@ def parse_compressed_arguments(byte_sequence: bytes, format_str: str) -> str:
arg_index += 1
# Character type
if conv_char == 'c':
if token.conv_char == 'c':
# Pad to 4 bytes for unpacking
padded = arg_bytes.ljust(4, b'\x00')
char_code = struct.unpack('>I', padded)[0]
output.append(chr(char_code))
# Pointer type
elif conv_char == 'p':
elif token.conv_char == 'p':
ptr_value = int.from_bytes(arg_bytes, 'big', signed=False)
output.append(hex(ptr_value))
# Floating point types
elif conv_char in 'fFeEgGaA':
elif token.conv_char in 'fFeEgGaA':
if len(arg_bytes) == 4:
float_value = struct.unpack('>f', arg_bytes)[0]
elif len(arg_bytes) == 8:
@@ -251,13 +262,13 @@ def parse_compressed_arguments(byte_sequence: bytes, format_str: str) -> str:
output.append(str(float_value))
# Integer types
elif conv_char in 'diuoxX':
signed = conv_char in 'di'
elif token.conv_char in 'diuoxX':
signed = token.conv_char in 'di'
# Determine expected size
if length_mod == 'll':
if token.length == 'll':
expected_size = 8
elif length_mod in ('l', 'z', 'j', 't') or conv_char == 'p':
elif token.length in ('l', 'z', 'j', 't') or token.conv_char == 'p':
expected_size = 4
else:
expected_size = len(arg_bytes)
@@ -275,9 +286,9 @@ def parse_compressed_arguments(byte_sequence: bytes, format_str: str) -> str:
# Convert to integer
int_value = int.from_bytes(arg_bytes, 'big', signed=signed)
output.append(format_integer(int_value, conv_char, flags, width, length_mod))
output.append(format_integer(int_value, token.conv_char, token.flags, token.width, token.length))
else:
raise ValueError(f'Unsupported conversion: {conv_char}')
raise ValueError(f'Unsupported conversion: {token.conv_char}')
else:
output.append(token)

183
components/bt/common/ble_log/extension/log_compression/tests/README.md Normal file
View File

@@ -0,0 +1,183 @@
# BLE Log Compression Test Suite
## Overview
This test suite validates the BLE log compression pipeline — from C source input through tree-sitter AST parsing, log ID assignment, compressed macro generation, to final `log_index.h` header output.
All tests use Python standard library `unittest` only. No third-party test frameworks required.
## Directory Structure
```
tests/
├── README.md # This file
├── test_utils.py # Shared utilities: path setup, PipelineContext, helpers
├── test_format_parser.py # FormatToken NamedTuple and parse_format_string
├── test_arg_size_types.py # ARG_SIZE_TYPE determination (U32/U64/STR)
├── test_db_manager.py # LogDBManager state persistence and incremental logic
├── test_log_extraction.py # Tree-sitter AST log extraction
├── 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
├── update_golden.py # Script to regenerate golden expected files
└── fixtures/
├── c_sources/ # Test input C files
│ ├── simple_logs.c # Basic: no-arg, single-arg, multi-arg, string
│ ├── format_specifiers.c # All format -> size type mappings
│ ├── special_tokens.c # __func__, __LINE__, MAC2STR, bt_hex
│ ├── concatenated_strings.c # PRIu64, PRId64 macro concatenation
│ ├── multi_level.c # ERROR/WARNING/API/DEBUG/EVENT/VERBOSE
│ ├── no_logs.c # Valid C with no matching log tags
│ └── mesh_logs.c # BLE Mesh module log patterns
└── expected/ # Golden output headers for comparison
├── simple_logs_index.h
├── format_specifiers_index.h
├── special_tokens_index.h
├── multi_level_index.h
└── multi_file_index.h
```
## Running Tests
From the `tests/` directory:
```bash
# Run all tests
python3 -m unittest discover -v
# Run a single test module
python3 -m unittest test_format_parser -v
# Run a single test class
python3 -m unittest test_arg_size_types.TestU64SizeType -v
# Run a single test method
python3 -m unittest test_db_manager.TestAddLog.test_sequential_ids -v
```
## Test Modules
### test_format_parser.py — Format String Parsing
Validates `FormatToken(NamedTuple)` and `parse_format_string()` in `c_format_parse.py`.
| Test Class | Coverage |
|---|---|
| `TestFormatTokenStructure` | NamedTuple field names, named vs index access, tuple subclass |
| `TestBasicSpecifiers` | 18 conversion characters (`%d`, `%s`, `%f`, `%p`, ...), escaped `%%`, empty string |
| `TestPositionTracking` | `FormatToken.start` / `.end` positions in source string |
| `TestLengthModifiers` | `h`, `hh`, `l`, `ll`, `z`, `j`, `t` |
| `TestWidthAndPrecision` | Fixed width/precision, dynamic `%*d` / `%.*s`, both dynamic |
| `TestFlags` | `-`, `+`, ` `, `#`, `0` |
| `TestMixedContent` | Interleaved text and specifiers, quoted format strings |
### test_arg_size_types.py — Size Type Determination
**Core regression test for the off-by-one bug fix.** Verifies that `FormatToken.conv_char` and `FormatToken.length` (not the old incorrect indices) are used to determine `ARG_SIZE_TYPE`.
| Format | Expected Size Type | Field Checked |
|---|---|---|
| `%d`, `%u`, `%x`, `%X`, `%o` | `U32` (0) | `token.conv_char` |
| `%s` | `STR` (1) | `token.conv_char` |
| `%f` | `U64` (2) | `token.conv_char` |
| `%lld`, `%llx` | `U64` (2) | `token.length` |
| `__func__` / `__LINE__` | skipped | `token.conv_char == '@func'` / `'@line'` |
| `MAC2STR(addr)` | buffer arg | `token.conv_char.startswith('@hex_func')` |
### test_db_manager.py — Database State Management
Tests `LogDBManager` in `LogDBManager.py`.
| Test Class | Coverage |
|---|---|
| `TestAddLog` | SUCCESS / LOG_EXISTS return codes, sequential IDs, unique key fields |
| `TestPersistence` | `save_all()` + reload preserves data |
| `TestFileProcessing` | `is_file_processed`: new file, unchanged, modified |
| `TestConfigUpdate` | `is_config_updated`: fresh DB, same config after reload, changed config |
| `TestSourceUpdateState` | `SOURCE_LOG_UPDATE_FULL` / `PARTIAL` / `NONE` transitions |
### test_log_extraction.py — AST Log Extraction
Tests `LogCompressor.extract_log_calls()` using tree-sitter.
| Test Class | Coverage |
|---|---|
| `TestExtractNoLogs` | No matching tags, no functions |
| `TestExtractBasicLogs` | Single log, no-arg log, multiple tags |
| `TestExtractFormatTokens` | Token population, hexify=True, dynamic width/precision filtered |
| `TestExtractSpecialTokens` | `__func__` -> `@func`, `__LINE__` -> `@line`, `MAC2STR` -> `@hex_func` |
| `TestExtractCallerInfo` | Correct enclosing function identification |
| `TestExtractArgMismatch` | Format/argument count mismatch raises `SyntaxError` |
### test_macro_generation.py — Module-Specific Macro Generators
Tests `make_bluedroid_log_macro.py` and `make_mesh_log_macro.py` directly.
| Test Class | Coverage |
|---|---|
| `TestBluedroidMacroGen` | `gen_header_head`, zero/single/multi-arg macros, level checks (`appl_trace_level >= BT_TRACE_LEVEL_*`), buffer args |
| `TestMeshMacroGen` | Header declarations, `BT_ERR`/`BT_DBG` level checks |
### test_pipeline_e2e.py — End-to-End Pipeline
Full `.c` -> `.h` pipeline tests with golden file comparison.
| Test Class | Coverage |
|---|---|
| `TestSimpleLogsE2E` | Header generation, tag replacement in source, header structure, sorted IDs, preserve tags |
| `TestFormatSpecifiersE2E` | All size types (U32/STR/U64) present in output |
| `TestSpecialTokensE2E` | `ble_log_compressed_hex_print_buf` generated, `__func__` excluded |
| `TestNoLogsE2E` | Zero macros for file without matching tags |
| `TestMultiLevelE2E` | Unique IDs per level, different level check prefixes |
| `TestMultiFileE2E` | Globally unique IDs across multiple files |
| `TestConcatenatedStringsE2E` | `PRIu64` produces U64 size type |
| `TestGoldenFileComparison` | Exact output comparison against `fixtures/expected/*.h` |
### test_incremental.py — Incremental Compression
Tests caching, re-runs, and config change behavior.
| Test Class | Coverage |
|---|---|
| `TestFirstRun` | All files compressed, header generated |
| `TestRerunUnchanged` | File hashes tracked, log IDs stable on re-add |
| `TestRerunModifiedFile` | Modified file reprocessed, existing IDs stable |
| `TestAddNewFile` | New file IDs continue from `max_id + 1` |
| `TestConfigChange` | Config change triggers `SOURCE_LOG_UPDATE_FULL` |
## 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).
When the compression logic changes intentionally:
```bash
# Regenerate all golden files
python3 update_golden.py
# Review the diff
git diff fixtures/expected/
# Commit after verification
```
## Fixture C Source Files
These files are parsed by tree-sitter only — they do not need to compile. Clang warnings about implicit declarations and unknown types are expected and irrelevant.
| File | Purpose |
|---|---|
| `simple_logs.c` | 4 functions: no-arg, single `%d`, triple `%d`, `%s` |
| `format_specifiers.c` | All `ARG_SIZE_TYPE` mappings: `%d/%u/%x/%X/%o` (U32), `%f` (U64), `%lld/%llx` (U64), `%s` (STR), mixed |
| `special_tokens.c` | `__func__`, `__FUNCTION__`, `__LINE__`, `MAC2STR`, mixed special + normal args |
| `concatenated_strings.c` | `PRIu64`, `PRId64`, `PRIx64` macro concatenation |
| `multi_level.c` | ERROR, WARNING, API, DEBUG, EVENT, VERBOSE in one function |
| `no_logs.c` | Only `printf`, no matching log tags |
| `mesh_logs.c` | BLE Mesh tags: `BT_ERR`, `BT_WARN`, `BT_INFO`, `BT_DBG` |
## Dependencies
- Python 3.8+
- `tree-sitter` + `tree-sitter-c` (same versions as in `scripts/requirements.txt`)
- `pyyaml`
- No additional test frameworks required (`unittest` is in the Python standard library)

18
components/bt/common/ble_log/extension/log_compression/tests/fixtures/c_sources/concatenated_strings.c vendored Normal file
View File

@@ -0,0 +1,18 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
/* Test fixture: PRIu64 and other macro-concatenated format strings */
void test_pri_u64(uint64_t val64) {
APPL_TRACE_DEBUG("val %" PRIu64, val64);
}
void test_pri_d64(int64_t sval64) {
APPL_TRACE_DEBUG("signed %" PRId64, sval64);
}
void test_pri_x64(uint64_t hval64) {
APPL_TRACE_DEBUG("hex %" PRIx64, hval64);
}

30
components/bt/common/ble_log/extension/log_compression/tests/fixtures/c_sources/format_specifiers.c vendored Normal file
View File

@@ -0,0 +1,30 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
/* Test fixture: all format specifier -> size type mappings */
void test_u32_types(int a, unsigned int b, int c, int d, int e) {
APPL_TRACE_DEBUG("int:%d unsigned:%u hex:%x HEX:%X octal:%o", a, b, c, d, e);
}
void test_u64_float(double f_val) {
APPL_TRACE_DEBUG("float: %f", f_val);
}
void test_u64_long_long(long long ll_val, long long llx_val) {
APPL_TRACE_DEBUG("long long: %lld hex_ll: %llx", ll_val, llx_val);
}
void test_str_type(const char *str1, const char *str2) {
APPL_TRACE_DEBUG("string: %s, another: %s", str1, str2);
}
void test_mixed_types(int i, const char *s, long long ll, double f, int h) {
APPL_TRACE_DEBUG("mixed: %d %s %lld %f %x", i, s, ll, f, h);
}
void test_width_flags(int val, int num, const char *str) {
APPL_TRACE_DEBUG("padded: %08x, width: %10d, left: %-20s", val, num, str);
}

22
components/bt/common/ble_log/extension/log_compression/tests/fixtures/c_sources/mesh_logs.c vendored Normal file
View File

@@ -0,0 +1,22 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
/* Test fixture: BLE Mesh module log patterns */
void mesh_error_func(int err) {
BT_ERR("Mesh error %d", err);
}
void mesh_warn_func(int code) {
BT_WARN("Mesh warning %d", code);
}
void mesh_info_func(void) {
BT_INFO("Mesh info");
}
void mesh_debug_func(int val) {
BT_DBG("Mesh debug %d", val);
}

15
components/bt/common/ble_log/extension/log_compression/tests/fixtures/c_sources/multi_level.c vendored Normal file
View File

@@ -0,0 +1,15 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
/* Test fixture: multiple log levels in one function */
void multi_level_func(int status, int handle) {
APPL_TRACE_ERROR("critical error %d", status);
APPL_TRACE_WARNING("warning condition %d", status);
APPL_TRACE_API("API call with handle %d", handle);
APPL_TRACE_DEBUG("debug info %d %d", status, handle);
APPL_TRACE_EVENT("event occurred");
APPL_TRACE_VERBOSE("verbose detail %d", handle);
}

13
components/bt/common/ble_log/extension/log_compression/tests/fixtures/c_sources/no_logs.c vendored Normal file
View File

@@ -0,0 +1,13 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
/* Test fixture: valid C file with no matching log tags */
#include <stdio.h>
void regular_function(int a, int b) {
int c = a + b;
printf("Result: %d\n", c);
}

22
components/bt/common/ble_log/extension/log_compression/tests/fixtures/c_sources/simple_logs.c vendored Normal file
View File

@@ -0,0 +1,22 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
/* Test fixture: basic log patterns */
void simple_no_args(void) {
APPL_TRACE_DEBUG("Simple message with no args");
}
void simple_one_int(int val) {
APPL_TRACE_ERROR("Value is %d", val);
}
void simple_multi_int(int a, int b, int c) {
APPL_TRACE_API("Values: %d, %d, %d", a, b, c);
}
void simple_string(const char *name) {
APPL_TRACE_WARNING("Name is %s", name);
}

26
components/bt/common/ble_log/extension/log_compression/tests/fixtures/c_sources/special_tokens.c vendored Normal file
View File

@@ -0,0 +1,26 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
/* Test fixture: __func__, __LINE__, MAC2STR, bt_hex */
void test_func_macro(void) {
APPL_TRACE_DEBUG("%s starting", __func__);
}
void test_function_macro(void) {
APPL_TRACE_DEBUG("%s called", __FUNCTION__);
}
void test_line_macro(int err_code) {
APPL_TRACE_ERROR("%s error at line %d with code %d", __func__, __LINE__, err_code);
}
void test_mac2str(const uint8_t *bd_addr) {
APPL_TRACE_DEBUG("addr=" MACSTR, MAC2STR(bd_addr));
}
void test_mixed_special(int handle, const uint8_t *addr) {
APPL_TRACE_DEBUG("%s handle=0x%x addr=" MACSTR, __func__, handle, MAC2STR(addr));
}

49
components/bt/common/ble_log/extension/log_compression/tests/fixtures/expected/format_specifiers_index.h vendored Normal file
View File

@@ -0,0 +1,49 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef __BLE_HOST_INTERNAL_LOG_INDEX_H
#define __BLE_HOST_INTERNAL_LOG_INDEX_H
#include <stddef.h>
#include <stdlib.h>
// Compression function declarations
extern int ble_log_compressed_hex_print
(uint8_t source, uint32_t log_index, size_t args_size_cnt, ...);
extern int ble_log_compressed_hex_print_buf
(uint8_t source, uint32_t log_index, uint8_t buf_idx, const uint8_t *buf, size_t len);
#define APPL_TRACE_DEBUG_1(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,1, 5, 0, 0, 0, 0, 0, a, b, c, d, e);\
;}\
}
#define APPL_TRACE_DEBUG_2(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,2, 1, 2, f_val);\
;}\
}
#define APPL_TRACE_DEBUG_3(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,3, 2, 2, 2, ll_val, llx_val);\
;}\
}
#define APPL_TRACE_DEBUG_4(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,4, 2, 1, 1, str1, str2);\
;}\
}
#define APPL_TRACE_DEBUG_5(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,5, 5, 0, 1, 2, 2, 0, i, s, ll, f, h);\
;}\
}
#define APPL_TRACE_DEBUG_6(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,6, 3, 0, 0, 1, val, num, str);\
;}\
}
#endif // __BLE_HOST_INTERNAL_LOG_INDEX_H

73
components/bt/common/ble_log/extension/log_compression/tests/fixtures/expected/multi_file_index.h vendored Normal file
View File

@@ -0,0 +1,73 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef __BLE_HOST_INTERNAL_LOG_INDEX_H
#define __BLE_HOST_INTERNAL_LOG_INDEX_H
#include <stddef.h>
#include <stdlib.h>
// Compression function declarations
extern int ble_log_compressed_hex_print
(uint8_t source, uint32_t log_index, size_t args_size_cnt, ...);
extern int ble_log_compressed_hex_print_buf
(uint8_t source, uint32_t log_index, uint8_t buf_idx, const uint8_t *buf, size_t len);
#define APPL_TRACE_ERROR_1(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_ERROR && BT_LOG_LEVEL_CHECK(APPL, ERROR)) ble_log_compressed_hex_print(0,1, 1, 0, status);\
;}\
APPL_TRACE_ERROR(fmt, ##__VA_ARGS__);\
}
#define APPL_TRACE_WARNING_2(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_WARNING && BT_LOG_LEVEL_CHECK(APPL, WARNING)) ble_log_compressed_hex_print(0,2, 1, 0, status);\
;}\
APPL_TRACE_WARNING(fmt, ##__VA_ARGS__);\
}
#define APPL_TRACE_API_3(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_API && BT_LOG_LEVEL_CHECK(APPL, API)) ble_log_compressed_hex_print(0,3, 1, 0, handle);\
;}\
}
#define APPL_TRACE_DEBUG_4(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,4, 2, 0, 0, status, handle);\
;}\
}
#define APPL_TRACE_EVENT_5(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_EVENT && BT_LOG_LEVEL_CHECK(APPL, EVENT)) ble_log_compressed_hex_print(0,5, 0);\
;}\
}
#define APPL_TRACE_VERBOSE_6(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_VERBOSE && BT_LOG_LEVEL_CHECK(APPL, VERBOSE)) ble_log_compressed_hex_print(0,6, 1, 0, handle);\
;}\
}
#define APPL_TRACE_DEBUG_7(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,7, 0);\
;}\
}
#define APPL_TRACE_ERROR_8(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_ERROR && BT_LOG_LEVEL_CHECK(APPL, ERROR)) ble_log_compressed_hex_print(0,8, 1, 0, val);\
;}\
APPL_TRACE_ERROR(fmt, ##__VA_ARGS__);\
}
#define APPL_TRACE_API_9(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_API && BT_LOG_LEVEL_CHECK(APPL, API)) ble_log_compressed_hex_print(0,9, 3, 0, 0, 0, a, b, c);\
;}\
}
#define APPL_TRACE_WARNING_10(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_WARNING && BT_LOG_LEVEL_CHECK(APPL, WARNING)) ble_log_compressed_hex_print(0,10, 1, 1, name);\
;}\
APPL_TRACE_WARNING(fmt, ##__VA_ARGS__);\
}
#endif // __BLE_HOST_INTERNAL_LOG_INDEX_H

51
components/bt/common/ble_log/extension/log_compression/tests/fixtures/expected/multi_level_index.h vendored Normal file
View File

@@ -0,0 +1,51 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef __BLE_HOST_INTERNAL_LOG_INDEX_H
#define __BLE_HOST_INTERNAL_LOG_INDEX_H
#include <stddef.h>
#include <stdlib.h>
// Compression function declarations
extern int ble_log_compressed_hex_print
(uint8_t source, uint32_t log_index, size_t args_size_cnt, ...);
extern int ble_log_compressed_hex_print_buf
(uint8_t source, uint32_t log_index, uint8_t buf_idx, const uint8_t *buf, size_t len);
#define APPL_TRACE_ERROR_1(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_ERROR && BT_LOG_LEVEL_CHECK(APPL, ERROR)) ble_log_compressed_hex_print(0,1, 1, 0, status);\
;}\
APPL_TRACE_ERROR(fmt, ##__VA_ARGS__);\
}
#define APPL_TRACE_WARNING_2(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_WARNING && BT_LOG_LEVEL_CHECK(APPL, WARNING)) ble_log_compressed_hex_print(0,2, 1, 0, status);\
;}\
APPL_TRACE_WARNING(fmt, ##__VA_ARGS__);\
}
#define APPL_TRACE_API_3(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_API && BT_LOG_LEVEL_CHECK(APPL, API)) ble_log_compressed_hex_print(0,3, 1, 0, handle);\
;}\
}
#define APPL_TRACE_DEBUG_4(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,4, 2, 0, 0, status, handle);\
;}\
}
#define APPL_TRACE_EVENT_5(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_EVENT && BT_LOG_LEVEL_CHECK(APPL, EVENT)) ble_log_compressed_hex_print(0,5, 0);\
;}\
}
#define APPL_TRACE_VERBOSE_6(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_VERBOSE && BT_LOG_LEVEL_CHECK(APPL, VERBOSE)) ble_log_compressed_hex_print(0,6, 1, 0, handle);\
;}\
}
#endif // __BLE_HOST_INTERNAL_LOG_INDEX_H

41
components/bt/common/ble_log/extension/log_compression/tests/fixtures/expected/simple_logs_index.h vendored Normal file
View File

@@ -0,0 +1,41 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef __BLE_HOST_INTERNAL_LOG_INDEX_H
#define __BLE_HOST_INTERNAL_LOG_INDEX_H
#include <stddef.h>
#include <stdlib.h>
// Compression function declarations
extern int ble_log_compressed_hex_print
(uint8_t source, uint32_t log_index, size_t args_size_cnt, ...);
extern int ble_log_compressed_hex_print_buf
(uint8_t source, uint32_t log_index, uint8_t buf_idx, const uint8_t *buf, size_t len);
#define APPL_TRACE_DEBUG_1(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,1, 0);\
;}\
}
#define APPL_TRACE_ERROR_2(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_ERROR && BT_LOG_LEVEL_CHECK(APPL, ERROR)) ble_log_compressed_hex_print(0,2, 1, 0, val);\
;}\
APPL_TRACE_ERROR(fmt, ##__VA_ARGS__);\
}
#define APPL_TRACE_API_3(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_API && BT_LOG_LEVEL_CHECK(APPL, API)) ble_log_compressed_hex_print(0,3, 3, 0, 0, 0, a, b, c);\
;}\
}
#define APPL_TRACE_WARNING_4(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_WARNING && BT_LOG_LEVEL_CHECK(APPL, WARNING)) ble_log_compressed_hex_print(0,4, 1, 1, name);\
;}\
APPL_TRACE_WARNING(fmt, ##__VA_ARGS__);\
}
#endif // __BLE_HOST_INTERNAL_LOG_INDEX_H

47
components/bt/common/ble_log/extension/log_compression/tests/fixtures/expected/special_tokens_index.h vendored Normal file
View File

@@ -0,0 +1,47 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef __BLE_HOST_INTERNAL_LOG_INDEX_H
#define __BLE_HOST_INTERNAL_LOG_INDEX_H
#include <stddef.h>
#include <stdlib.h>
// Compression function declarations
extern int ble_log_compressed_hex_print
(uint8_t source, uint32_t log_index, size_t args_size_cnt, ...);
extern int ble_log_compressed_hex_print_buf
(uint8_t source, uint32_t log_index, uint8_t buf_idx, const uint8_t *buf, size_t len);
#define APPL_TRACE_DEBUG_1(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,1, 0);\
;}\
}
#define APPL_TRACE_DEBUG_2(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,2, 0);\
;}\
}
#define APPL_TRACE_ERROR_3(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_ERROR && BT_LOG_LEVEL_CHECK(APPL, ERROR)) ble_log_compressed_hex_print(0,3, 1, 0, err_code);\
;}\
APPL_TRACE_ERROR(fmt, ##__VA_ARGS__);\
}
#define APPL_TRACE_DEBUG_4(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,4, 0);\
ble_log_compressed_hex_print_buf(0, 4, 0, (const uint8_t *)bd_addr, 6);\
;}\
}
#define APPL_TRACE_DEBUG_5(fmt, ...) {\
{if (appl_trace_level >= BT_TRACE_LEVEL_DEBUG && BT_LOG_LEVEL_CHECK(APPL, DEBUG)) ble_log_compressed_hex_print(0,5, 1, 0, handle);\
ble_log_compressed_hex_print_buf(0, 5, 0, (const uint8_t *)addr, 6);\
;}\
}
#endif // __BLE_HOST_INTERNAL_LOG_INDEX_H

138
components/bt/common/ble_log/extension/log_compression/tests/test_arg_size_types.py Normal file
View File

@@ -0,0 +1,138 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Tests for argument size type determination in generate_compressed_macro.
Verifies that FormatToken.conv_char and FormatToken.length are used correctly
to determine ARG_SIZE_TYPE (the fix for the off-by-one bug).
"""
import re
import shutil
import tempfile
import unittest
from typing import List
from typing import Optional
import test_utils # noqa: F401 — must be first to set up sys.path
from ble_log_compress import ARG_SIZE_TYPE
from test_utils import PipelineContext
def _extract_size_types_from_macro(macro_str: str) -> Optional[List[int]]:
"""Extract size type values from a generated macro string."""
m = re.search(r'ble_log_compressed_hex_print\(\d+,\d+, (\d+)(?:, (.+?))?(?:,\s*[a-zA-Z_])', macro_str)
if not m:
m = re.search(r'ble_log_compressed_hex_print\(\d+,\d+, 0\)', macro_str)
if m:
return []
return None
arg_count = int(m.group(1))
if arg_count == 0:
return []
size_str = m.group(2)
return [int(x.strip()) for x in size_str.split(',')[:arg_count]]
def _compress_and_get_macro(ctx: 'PipelineContext', c_code: bytes) -> str:
"""Compress a C code snippet and return the generated macro string."""
src_dir = ctx.code_base / 'test_src'
src_dir.mkdir(parents=True, exist_ok=True)
(src_dir / 'test_size.c').write_bytes(c_code)
tmp_dir = ctx.compressed_srcs / 'test_src'
tmp_dir.mkdir(parents=True, exist_ok=True)
(tmp_dir / 'test_size.c.tmp').write_bytes(c_code)
file_macros = ctx.compressor.compress_file(('BLE_HOST', str(tmp_dir / 'test_size.c.tmp')))
assert len(file_macros) > 0, 'No macros generated'
result: str = file_macros[0][2]
return result
class _PipelineTestCase(unittest.TestCase):
"""Base class that sets up and tears down a full pipeline context."""
def setUp(self) -> None:
self.tmp = tempfile.mkdtemp()
self.ctx = PipelineContext(self.tmp)
def tearDown(self) -> None:
self.ctx.close()
shutil.rmtree(self.tmp, ignore_errors=True)
class TestU32SizeType(_PipelineTestCase):
def test_int_d(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("v %d", a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U32)])
def test_unsigned_u(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("v %u", a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U32)])
def test_hex_x(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("v %x", a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U32)])
class TestU64SizeType(_PipelineTestCase):
def test_float_f(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("v %f", a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U64)])
def test_long_long_d(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("v %lld", a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U64)])
def test_long_long_x(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("v %llx", a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U64)])
class TestSTRSizeType(_PipelineTestCase):
def test_string_s(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("v %s", a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.STR)])
class TestMixedSizeTypes(_PipelineTestCase):
def test_d_s_lld_f(self) -> None:
macro = _compress_and_get_macro(
self.ctx, b'void f(void) { APPL_TRACE_DEBUG("a %d b %s c %lld d %f", i, s, ll, fv); }'
)
self.assertEqual(
_extract_size_types_from_macro(macro),
[
int(ARG_SIZE_TYPE.U32),
int(ARG_SIZE_TYPE.STR),
int(ARG_SIZE_TYPE.U64),
int(ARG_SIZE_TYPE.U64),
],
)
class TestSpecialTokensSkipped(_PipelineTestCase):
def test_func_excluded_from_args(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("%s val %d", __func__, a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U32)])
def test_line_excluded_from_args(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("line %d val %d", __LINE__, a); }')
self.assertEqual(_extract_size_types_from_macro(macro), [int(ARG_SIZE_TYPE.U32)])
def test_zero_args_after_func(self) -> None:
macro = _compress_and_get_macro(self.ctx, b'void f(void) { APPL_TRACE_DEBUG("%s starting", __func__); }')
self.assertEqual(_extract_size_types_from_macro(macro), [])
class TestHexFuncSkipped(_PipelineTestCase):
def test_mac2str_generates_buf_print(self) -> None:
macro = _compress_and_get_macro(
self.ctx, b'void f(void) { APPL_TRACE_DEBUG("addr=" MACSTR, MAC2STR(bd_addr)); }'
)
self.assertIn('ble_log_compressed_hex_print_buf', macro)
self.assertIn('bd_addr', macro)
if __name__ == '__main__':
unittest.main()

330
components/bt/common/ble_log/extension/log_compression/tests/test_db_manager.py Normal file
View File

@@ -0,0 +1,330 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Tests for LogDBManager: state persistence, incremental logic, file tracking."""
import shutil
import tempfile
import unittest
import test_utils # noqa: F401 — must be first to set up sys.path
from LogDBManager import LogDBManager
class TestAddLog(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.mkdtemp()
self.db = LogDBManager(data_dir=f'{self.tmp}/db', sources={'BLE_HOST': 'test_config_v1'})
def tearDown(self) -> None:
self.db.close()
shutil.rmtree(self.tmp, ignore_errors=True)
def test_add_new_returns_success(self) -> None:
result, log_id = self.db.add_log(
source='BLE_HOST',
log_tag='APPL_TRACE_DEBUG',
log_format='"message %d"',
log_line_number=10,
hexify=True,
caller_func='foo',
caller_line=5,
file_name='test.c',
)
self.assertEqual(result, LogDBManager.SUCCESS)
self.assertEqual(log_id, 1)
def test_add_duplicate_returns_exists(self) -> None:
kw = dict(
source='BLE_HOST',
log_tag='APPL_TRACE_DEBUG',
log_format='"message %d"',
log_line_number=10,
hexify=True,
caller_func='foo',
caller_line=5,
file_name='test.c',
)
self.db.add_log(**kw)
result, log_id = self.db.add_log(**kw)
self.assertEqual(result, LogDBManager.LOG_EXISTS)
self.assertEqual(log_id, 1)
def test_sequential_ids(self) -> None:
_, id1 = self.db.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m1"',
log_line_number=10,
hexify=True,
caller_func='foo',
caller_line=5,
file_name='test.c',
)
_, id2 = self.db.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m2"',
log_line_number=20,
hexify=True,
caller_func='bar',
caller_line=15,
file_name='test.c',
)
self.assertEqual(id1, 1)
self.assertEqual(id2, 2)
def test_unique_key_distinguishes_all_fields(self) -> None:
_, id1 = self.db.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"msg %d"',
log_line_number=10,
hexify=True,
caller_func='foo',
caller_line=5,
file_name='a.c',
)
_, id2 = self.db.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"msg %d"',
log_line_number=20,
hexify=True,
caller_func='bar',
caller_line=15,
file_name='b.c',
)
self.assertNotEqual(id1, id2)
class TestPersistence(unittest.TestCase):
def test_save_and_reload(self) -> None:
tmp = tempfile.mkdtemp()
try:
db_path = f'{tmp}/db'
sources = {'BLE_HOST': 'cfg_v1'}
mgr1 = LogDBManager(data_dir=db_path, sources=sources)
mgr1.add_log(
source='BLE_HOST',
log_tag='APPL_TRACE_ERROR',
log_format='"error %d"',
log_line_number=10,
hexify=True,
caller_func='func1',
caller_line=5,
file_name='f.c',
)
mgr1.save_all()
mgr1.close()
mgr2 = LogDBManager(data_dir=db_path, sources=sources)
result, log_id = mgr2.add_log(
source='BLE_HOST',
log_tag='APPL_TRACE_ERROR',
log_format='"error %d"',
log_line_number=10,
hexify=True,
caller_func='func1',
caller_line=5,
file_name='f.c',
)
self.assertEqual(result, LogDBManager.LOG_EXISTS)
self.assertEqual(log_id, 1)
mgr2.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
class TestFileProcessing(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.mkdtemp()
self.db = LogDBManager(data_dir=f'{self.tmp}/db', sources={'BLE_HOST': 'cfg_v1'})
def tearDown(self) -> None:
self.db.close()
shutil.rmtree(self.tmp, ignore_errors=True)
def _write(self, name: str, content: str) -> str:
import os
p = os.path.join(self.tmp, name)
with open(p, 'w') as f:
f.write(content)
return p
def test_new_file_not_processed(self) -> None:
src = self._write('src.c', 'void foo(void) {}')
comp = self._write('src.c.tmp', 'void foo(void) {}')
self.assertFalse(self.db.is_file_processed('BLE_HOST', src, comp))
def test_marked_file_is_processed(self) -> None:
src = self._write('src.c', 'void foo(void) {}')
comp = self._write('src.c.tmp', 'void foo_compressed(void) {}')
self.db.mark_file_processed('BLE_HOST', src, comp)
self.db.save_all()
self.assertTrue(self.db.is_file_processed('BLE_HOST', src, comp))
def test_modified_file_not_processed(self) -> None:
src = self._write('src.c', 'void foo(void) {}')
comp = self._write('src.c.tmp', 'void foo_compressed(void) {}')
self.db.mark_file_processed('BLE_HOST', src, comp)
self.db.save_all()
# Modify source
with open(src, 'w') as f:
f.write('void foo(void) { /* changed */ }')
self.assertFalse(self.db.is_file_processed('BLE_HOST', src, comp))
class TestConfigUpdate(unittest.TestCase):
def test_fresh_db_config_is_updated(self) -> None:
tmp = tempfile.mkdtemp()
try:
db = LogDBManager(data_dir=f'{tmp}/db', sources={'BLE_HOST': 'cfg_v1'})
self.assertTrue(db.is_config_updated('BLE_HOST'))
db.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_same_config_not_updated_after_save_reload(self) -> None:
tmp = tempfile.mkdtemp()
try:
sources = {'BLE_HOST': 'cfg_v1'}
mgr1 = LogDBManager(data_dir=f'{tmp}/db', sources=sources)
mgr1.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m"',
log_line_number=1,
hexify=True,
caller_func='f',
caller_line=1,
file_name='x.c',
)
mgr1.save_all()
mgr1.close()
mgr2 = LogDBManager(data_dir=f'{tmp}/db', sources=sources)
self.assertFalse(mgr2.is_config_updated('BLE_HOST'))
mgr2.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_different_config_is_updated(self) -> None:
tmp = tempfile.mkdtemp()
try:
mgr1 = LogDBManager(data_dir=f'{tmp}/db', sources={'BLE_HOST': 'cfg_v1'})
mgr1.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m"',
log_line_number=1,
hexify=True,
caller_func='f',
caller_line=1,
file_name='x.c',
)
mgr1.save_all()
mgr1.close()
mgr2 = LogDBManager(data_dir=f'{tmp}/db', sources={'BLE_HOST': 'cfg_v2_changed'})
self.assertTrue(mgr2.is_config_updated('BLE_HOST'))
mgr2.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
class TestSourceUpdateState(unittest.TestCase):
def test_fresh_db_returns_full(self) -> None:
tmp = tempfile.mkdtemp()
try:
db = LogDBManager(data_dir=f'{tmp}/db', sources={'BLE_HOST': 'cfg_v1'})
self.assertEqual(db.source_update_state('BLE_HOST'), LogDBManager.SOURCE_LOG_UPDATE_FULL)
db.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_loaded_with_logs_no_new_returns_none(self) -> None:
tmp = tempfile.mkdtemp()
try:
sources = {'BLE_HOST': 'cfg_v1'}
mgr1 = LogDBManager(data_dir=f'{tmp}/db', sources=sources)
mgr1.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m"',
log_line_number=1,
hexify=True,
caller_func='f',
caller_line=1,
file_name='x.c',
)
mgr1.save_all()
mgr1.close()
mgr2 = LogDBManager(data_dir=f'{tmp}/db', sources=sources)
self.assertEqual(mgr2.source_update_state('BLE_HOST'), LogDBManager.SOURCE_LOG_UPDATE_NONE)
mgr2.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_new_log_added_returns_partial(self) -> None:
tmp = tempfile.mkdtemp()
try:
sources = {'BLE_HOST': 'cfg_v1'}
mgr1 = LogDBManager(data_dir=f'{tmp}/db', sources=sources)
mgr1.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m1"',
log_line_number=1,
hexify=True,
caller_func='f',
caller_line=1,
file_name='x.c',
)
mgr1.save_all()
mgr1.close()
mgr2 = LogDBManager(data_dir=f'{tmp}/db', sources=sources)
mgr2.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m2"',
log_line_number=2,
hexify=True,
caller_func='g',
caller_line=2,
file_name='x.c',
)
self.assertEqual(mgr2.source_update_state('BLE_HOST'), LogDBManager.SOURCE_LOG_UPDATE_PARTIAL)
mgr2.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_config_change_returns_full(self) -> None:
tmp = tempfile.mkdtemp()
try:
mgr1 = LogDBManager(data_dir=f'{tmp}/db', sources={'BLE_HOST': 'cfg_v1'})
mgr1.add_log(
source='BLE_HOST',
log_tag='T',
log_format='"m"',
log_line_number=1,
hexify=True,
caller_func='f',
caller_line=1,
file_name='x.c',
)
mgr1.save_all()
mgr1.close()
mgr2 = LogDBManager(data_dir=f'{tmp}/db', sources={'BLE_HOST': 'cfg_v2'})
self.assertEqual(mgr2.source_update_state('BLE_HOST'), LogDBManager.SOURCE_LOG_UPDATE_FULL)
mgr2.close()
finally:
shutil.rmtree(tmp, ignore_errors=True)
if __name__ == '__main__':
unittest.main()

200
components/bt/common/ble_log/extension/log_compression/tests/test_format_parser.py Normal file
View File

@@ -0,0 +1,200 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Tests for c_format_parse.py: FormatToken NamedTuple and parse_format_string."""
import unittest
import test_utils # noqa: F401 — must be first to set up sys.path
from c_format_parse import FormatToken
from c_format_parse import parse_format_string
class TestFormatTokenStructure(unittest.TestCase):
def test_is_tuple_subclass(self) -> None:
tokens = parse_format_string('%d')
self.assertIsInstance(tokens[0], tuple)
self.assertIsInstance(tokens[0], FormatToken)
def test_field_names(self) -> None:
self.assertEqual(
FormatToken._fields,
('start', 'end', 'full_spec', 'flags', 'width', 'precision', 'length', 'conv_char'),
)
def test_named_access_matches_index(self) -> None:
ft = parse_format_string('%08llx')[0]
self.assertEqual(ft.start, ft[0])
self.assertEqual(ft.end, ft[1])
self.assertEqual(ft.full_spec, ft[2])
self.assertEqual(ft.flags, ft[3])
self.assertEqual(ft.width, ft[4])
self.assertEqual(ft.precision, ft[5])
self.assertEqual(ft.length, ft[6])
self.assertEqual(ft.conv_char, ft[7])
def test_complex_token_fields(self) -> None:
ft = parse_format_string('%08llx')[0]
self.assertEqual(ft.flags, '0')
self.assertEqual(ft.width, '8')
self.assertEqual(ft.precision, '')
self.assertEqual(ft.length, 'll')
self.assertEqual(ft.conv_char, 'x')
self.assertEqual(ft.full_spec, '%08llx')
class TestBasicSpecifiers(unittest.TestCase):
def test_single_specifiers(self) -> None:
cases = [
('%d', 'd'),
('%i', 'i'),
('%u', 'u'),
('%x', 'x'),
('%X', 'X'),
('%o', 'o'),
('%c', 'c'),
('%s', 's'),
('%p', 'p'),
('%f', 'f'),
('%F', 'F'),
('%e', 'e'),
('%E', 'E'),
('%g', 'g'),
('%G', 'G'),
('%a', 'a'),
('%A', 'A'),
('%n', 'n'),
]
for fmt, conv in cases:
with self.subTest(fmt=fmt):
tokens = parse_format_string(fmt)
self.assertEqual(len(tokens), 1)
self.assertIsInstance(tokens[0], FormatToken)
self.assertEqual(tokens[0].conv_char, conv)
def test_escaped_percent(self) -> None:
tokens = parse_format_string('100%%')
self.assertTrue(all(isinstance(t, str) for t in tokens))
def test_empty_string(self) -> None:
self.assertEqual(parse_format_string(''), [])
def test_no_specifiers(self) -> None:
tokens = parse_format_string('hello world')
self.assertEqual(len(tokens), 1)
self.assertEqual(tokens[0], 'hello world')
class TestPositionTracking(unittest.TestCase):
def test_start_end_single(self) -> None:
ft = parse_format_string('%d')[0]
self.assertEqual(ft.start, 0)
self.assertEqual(ft.end, 2)
def test_start_end_with_prefix(self) -> None:
tokens = parse_format_string('abc%dxyz')
ft = [t for t in tokens if isinstance(t, FormatToken)][0]
self.assertEqual(ft.start, 3)
self.assertEqual(ft.end, 5)
def test_multiple_positions(self) -> None:
tokens = parse_format_string('%d %s')
fts = [t for t in tokens if isinstance(t, FormatToken)]
self.assertEqual(fts[0].start, 0)
self.assertEqual(fts[0].end, 2)
self.assertEqual(fts[1].start, 3)
self.assertEqual(fts[1].end, 5)
class TestLengthModifiers(unittest.TestCase):
def test_length_modifiers(self) -> None:
cases = [
('%hd', 'h', 'd'),
('%hhd', 'hh', 'd'),
('%ld', 'l', 'd'),
('%lld', 'll', 'd'),
('%zu', 'z', 'u'),
('%jd', 'j', 'd'),
('%td', 't', 'd'),
('%llx', 'll', 'x'),
('%lx', 'l', 'x'),
('%hu', 'h', 'u'),
]
for fmt, length, conv in cases:
with self.subTest(fmt=fmt):
ft = parse_format_string(fmt)[0]
self.assertEqual(ft.length, length)
self.assertEqual(ft.conv_char, conv)
class TestWidthAndPrecision(unittest.TestCase):
def test_fixed_width(self) -> None:
ft = parse_format_string('%10d')[0]
self.assertEqual(ft.width, '10')
self.assertEqual(ft.precision, '')
def test_fixed_precision(self) -> None:
ft = parse_format_string('%.2f')[0]
self.assertEqual(ft.width, '')
self.assertEqual(ft.precision, '2')
def test_width_and_precision(self) -> None:
ft = parse_format_string('%10.2f')[0]
self.assertEqual(ft.width, '10')
self.assertEqual(ft.precision, '2')
def test_dynamic_width(self) -> None:
ft = parse_format_string('%*d')[0]
self.assertEqual(ft.width, '*')
def test_dynamic_precision(self) -> None:
ft = parse_format_string('%.*s')[0]
self.assertEqual(ft.precision, '*')
def test_both_dynamic(self) -> None:
ft = parse_format_string('%*.*f')[0]
self.assertEqual(ft.width, '*')
self.assertEqual(ft.precision, '*')
class TestFlags(unittest.TestCase):
def test_flags(self) -> None:
cases = [
('%-d', '-'),
('%+d', '+'),
('% d', ' '),
('%#x', '#'),
('%0d', '0'),
('%08x', '0'),
]
for fmt, expected_flags in cases:
with self.subTest(fmt=fmt):
ft = parse_format_string(fmt)[0]
self.assertEqual(ft.flags, expected_flags)
class TestMixedContent(unittest.TestCase):
def test_text_and_specifiers(self) -> None:
tokens = parse_format_string('val=%d name=%s')
fts = [t for t in tokens if isinstance(t, FormatToken)]
strs = [t for t in tokens if isinstance(t, str)]
self.assertEqual(len(fts), 2)
self.assertEqual(fts[0].conv_char, 'd')
self.assertEqual(fts[1].conv_char, 's')
self.assertTrue(any('val=' in s for s in strs))
def test_many_specifiers(self) -> None:
tokens = parse_format_string('%d %s %lld %f %x')
fts = [t for t in tokens if isinstance(t, FormatToken)]
self.assertEqual(len(fts), 5)
self.assertEqual([ft.conv_char for ft in fts], ['d', 's', 'd', 'f', 'x'])
self.assertEqual(fts[2].length, 'll')
def test_quoted_format_string(self) -> None:
tokens = parse_format_string('"hello %d world"')
fts = [t for t in tokens if isinstance(t, FormatToken)]
self.assertEqual(len(fts), 1)
self.assertEqual(fts[0].conv_char, 'd')
if __name__ == '__main__':
unittest.main()

162
components/bt/common/ble_log/extension/log_compression/tests/test_incremental.py Normal file
View File

@@ -0,0 +1,162 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Tests for incremental compression: caching, re-runs, config changes."""
import shutil
import tempfile
import unittest
import test_utils # noqa: F401 — must be first to set up sys.path
import yaml
from LogDBManager import LogDBManager
from test_utils import PipelineContext
class _IncrementalTestCase(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.mkdtemp()
self.ctx = PipelineContext(self.tmp)
def tearDown(self) -> None:
self.ctx.close()
shutil.rmtree(self.tmp, ignore_errors=True)
class TestFirstRun(_IncrementalTestCase):
def test_all_files_compressed(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
macros = self.ctx.run_compression([rel])
self.assertIn('BLE_HOST', macros)
self.assertGreater(len(macros['BLE_HOST']), 0)
def test_header_generated(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
self.assertTrue(self.ctx.header_path.exists())
class TestRerunUnchanged(_IncrementalTestCase):
def test_file_hashes_tracked(self) -> None:
"""After first run, is_file_processed returns True for unchanged files."""
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
# Simulate new build: close and reopen DB
self.ctx.db_manager.close()
db_path = self.ctx.build_dir / self.ctx.compressor.config.get('db_path', 'ble_log/ble_log_database')
self.ctx.db_manager = LogDBManager(
data_dir=str(db_path),
sources={s: str(c) for s, c in self.ctx.compressor.module_info.items()},
)
self.ctx.compressor.db_manager = self.ctx.db_manager
src_path = self.ctx.code_base / rel
temp_path = str(self.ctx.compressed_srcs / rel) + '.tmp'
self.assertTrue(self.ctx.db_manager.is_file_processed('BLE_HOST', src_path, temp_path))
def test_log_ids_unchanged_on_rerun(self) -> None:
"""Re-running compression on the same file produces same IDs."""
rel = self.ctx.copy_fixture('simple_logs.c')
macros1 = self.ctx.run_compression([rel])
ids1 = sorted(log_id for log_id, _ in macros1['BLE_HOST'])
# Re-run: same file, same content — macros for already-existing logs
macros2 = self.ctx.run_compression([rel])
ids2 = sorted(log_id for log_id, _ in macros2['BLE_HOST'])
self.assertEqual(ids1, ids2)
class TestRerunModifiedFile(_IncrementalTestCase):
def test_modified_file_reprocessed(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
# Modify the source file
src_path = self.ctx.code_base / rel
content = src_path.read_text()
content += '\nvoid new_func(int v) { APPL_TRACE_DEBUG("new %d", v); }\n'
src_path.write_text(content)
macros = self.ctx.run_compression([rel])
self.assertIn('BLE_HOST', macros)
self.assertGreater(len(macros['BLE_HOST']), 0)
def test_existing_ids_stable(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
macros1 = self.ctx.run_compression([rel])
ids1 = sorted(log_id for log_id, _ in macros1['BLE_HOST'])
# Modify source: add a new log
src_path = self.ctx.code_base / rel
content = src_path.read_text()
content += '\nvoid extra(int v) { APPL_TRACE_DEBUG("extra %d", v); }\n'
src_path.write_text(content)
# Re-run — original IDs should still appear
macros2 = self.ctx.run_compression([rel])
ids2 = sorted(log_id for log_id, _ in macros2['BLE_HOST'])
for old_id in ids1:
self.assertIn(old_id, ids2)
class TestAddNewFile(_IncrementalTestCase):
def test_new_file_ids_continue(self) -> None:
rel1 = self.ctx.copy_fixture('simple_logs.c')
macros1 = self.ctx.run_compression([rel1])
ids1 = set(log_id for log_id, _ in macros1['BLE_HOST'])
max_id_1 = max(ids1)
# Simulate new build
self.ctx.db_manager.close()
db_path = self.ctx.build_dir / self.ctx.compressor.config.get('db_path', 'ble_log/ble_log_database')
self.ctx.db_manager = LogDBManager(
data_dir=str(db_path),
sources={s: str(c) for s, c in self.ctx.compressor.module_info.items()},
)
self.ctx.compressor.db_manager = self.ctx.db_manager
rel2 = self.ctx.copy_fixture('multi_level.c')
macros2 = self.ctx.run_compression([rel1, rel2])
if 'BLE_HOST' in macros2 and macros2['BLE_HOST']:
all_ids_2 = set(log_id for log_id, _ in macros2['BLE_HOST'])
new_ids = all_ids_2 - ids1
self.assertGreater(len(new_ids), 0)
for new_id in new_ids:
self.assertGreater(new_id, max_id_1)
class TestConfigChange(_IncrementalTestCase):
def test_config_change_triggers_full_regen(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
# Change config
with open(self.ctx.config_path) as f:
config = yaml.safe_load(f)
config['log_config']['modules']['BLE_HOST']['tags'].append('NEW_TAG_ADDED')
with open(self.ctx.config_path, 'w') as f:
yaml.safe_dump(config, f)
# Reload
self.ctx.compressor.module_info.clear()
self.ctx.compressor.module_mod.clear()
self.ctx.compressor.load_config(str(self.ctx.config_path), ['BLE_HOST'])
self.ctx.db_manager.close()
db_path = self.ctx.build_dir / self.ctx.compressor.config.get('db_path', 'ble_log/ble_log_database')
self.ctx.db_manager = LogDBManager(
data_dir=str(db_path),
sources={s: str(c) for s, c in self.ctx.compressor.module_info.items()},
)
self.ctx.compressor.db_manager = self.ctx.db_manager
self.assertTrue(self.ctx.db_manager.is_config_updated('BLE_HOST'))
self.assertEqual(
self.ctx.db_manager.source_update_state('BLE_HOST'),
LogDBManager.SOURCE_LOG_UPDATE_FULL,
)
if __name__ == '__main__':
unittest.main()

149
components/bt/common/ble_log/extension/log_compression/tests/test_log_extraction.py Normal file
View File

@@ -0,0 +1,149 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Tests for LogCompressor.extract_log_calls: tree-sitter AST extraction."""
import unittest
import test_utils # noqa: F401 — must be first to set up sys.path
from c_format_parse import FormatToken
from test_utils import make_compressor
TAGS = [
'APPL_TRACE_ERROR',
'APPL_TRACE_WARNING',
'APPL_TRACE_API',
'APPL_TRACE_DEBUG',
'APPL_TRACE_EVENT',
'APPL_TRACE_VERBOSE',
]
class TestExtractNoLogs(unittest.TestCase):
def setUp(self) -> None:
self.comp = make_compressor()
def test_no_matching_tags(self) -> None:
code = b'void foo(void) { printf("hello"); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(logs, [])
def test_file_with_no_functions(self) -> None:
code = b'int global_var = 42;'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(logs, [])
class TestExtractBasicLogs(unittest.TestCase):
def setUp(self) -> None:
self.comp = make_compressor()
def test_single_log(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("msg %d", val); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 1)
self.assertEqual(logs[0]['tag'][0], 'APPL_TRACE_DEBUG')
self.assertEqual(logs[0]['caller_name'], 'foo')
def test_no_arg_log(self) -> None:
code = b'void bar(void) { APPL_TRACE_DEBUG("simple message"); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 1)
tokens = [t for t in logs[0]['argu_tokens'] if isinstance(t, FormatToken)]
self.assertEqual(len(tokens), 0)
def test_multiple_tags(self) -> None:
code = b'void foo(void) {\n APPL_TRACE_DEBUG("debug");\n APPL_TRACE_ERROR("error %d", code);\n}\n'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 2)
self.assertEqual(logs[0]['tag'][0], 'APPL_TRACE_DEBUG')
self.assertEqual(logs[1]['tag'][0], 'APPL_TRACE_ERROR')
class TestExtractFormatTokens(unittest.TestCase):
def setUp(self) -> None:
self.comp = make_compressor()
def test_format_tokens_populated(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("val %d str %s", a, b); }'
logs = self.comp.extract_log_calls(code, TAGS)
tokens = [t for t in logs[0]['argu_tokens'] if isinstance(t, FormatToken)]
self.assertEqual(len(tokens), 2)
self.assertEqual(tokens[0].conv_char, 'd')
self.assertEqual(tokens[1].conv_char, 's')
def test_hexify_true_for_normal_formats(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("val %d", a); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertTrue(logs[0]['hexify'])
def test_hexify_false_for_dynamic_width(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("%*d", w, a); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 0)
def test_hexify_false_for_dynamic_precision(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("%.*s", p, s); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 0)
class TestExtractSpecialTokens(unittest.TestCase):
def setUp(self) -> None:
self.comp = make_compressor()
def test_func_macro_replaced(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("%s starting", __func__); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 1)
tokens = [t for t in logs[0]['argu_tokens'] if isinstance(t, FormatToken)]
self.assertTrue(any(t.conv_char == '@func' for t in tokens))
def test_line_macro_replaced(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("at line %d", __LINE__); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 1)
tokens = [t for t in logs[0]['argu_tokens'] if isinstance(t, FormatToken)]
self.assertTrue(any(t.conv_char == '@line' for t in tokens))
def test_mac2str_replaced(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("addr=" MACSTR, MAC2STR(bd_addr)); }'
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 1)
tokens = [t for t in logs[0]['argu_tokens'] if isinstance(t, FormatToken)]
hex_tokens = [t for t in tokens if t.conv_char.startswith('@hex_func')]
self.assertEqual(len(hex_tokens), 1)
self.assertIn('bd_addr', hex_tokens[0].conv_char)
self.assertIn('6', hex_tokens[0].conv_char)
class TestExtractCallerInfo(unittest.TestCase):
def setUp(self) -> None:
self.comp = make_compressor()
def test_correct_caller(self) -> None:
code = (
b'void first_func(void) {\n'
b' APPL_TRACE_DEBUG("in first");\n'
b'}\n'
b'void second_func(void) {\n'
b' APPL_TRACE_DEBUG("in second");\n'
b'}\n'
)
logs = self.comp.extract_log_calls(code, TAGS)
self.assertEqual(len(logs), 2)
self.assertEqual(logs[0]['caller_name'], 'first_func')
self.assertEqual(logs[1]['caller_name'], 'second_func')
class TestExtractArgMismatch(unittest.TestCase):
def setUp(self) -> None:
self.comp = make_compressor()
def test_arg_count_mismatch_raises(self) -> None:
code = b'void foo(void) { APPL_TRACE_DEBUG("msg %d %d", val); }'
with self.assertRaises(SyntaxError):
self.comp.extract_log_calls(code, TAGS)
if __name__ == '__main__':
unittest.main()

159
components/bt/common/ble_log/extension/log_compression/tests/test_macro_generation.py Normal file
View File

@@ -0,0 +1,159 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Tests for compressed macro generation (bluedroid and mesh module scripts)."""
import sys
import types
import unittest
from test_utils import SCRIPTS_DIR
class TestBluedroidMacroGen(unittest.TestCase):
mod: types.ModuleType
@classmethod
def setUpClass(cls) -> None:
p = str(SCRIPTS_DIR / 'module_scripts' / 'bluedroid')
if p not in sys.path:
sys.path.insert(0, p)
import make_bluedroid_log_macro
cls.mod = make_bluedroid_log_macro
def test_header_head_has_declarations(self) -> None:
head = self.mod.gen_header_head()
self.assertIn('ble_log_compressed_hex_print', head)
self.assertIn('ble_log_compressed_hex_print_buf', head)
self.assertIn('extern', head)
def test_zero_args_macro(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=1,
module_id=0,
func_name='APPL_TRACE_DEBUG',
fmt=None,
args=[],
buffer_args=[],
)
self.assertIn('ble_log_compressed_hex_print(0,1, 0)', stmt)
def test_single_arg_macro(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=2,
module_id=0,
func_name='APPL_TRACE_ERROR',
fmt=None,
args=[{'name': 'val', 'size_type': '0'}],
buffer_args=[],
)
self.assertIn('ble_log_compressed_hex_print(0,2, 1, 0, val)', stmt)
def test_multi_arg_with_sizes(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=3,
module_id=0,
func_name='APPL_TRACE_DEBUG',
fmt=None,
args=[
{'name': 'a', 'size_type': '0'},
{'name': 'b', 'size_type': '1'},
{'name': 'c', 'size_type': '2'},
],
buffer_args=[],
)
self.assertIn('3, 0, 1, 2, a, b, c', stmt)
def test_appl_error_level_check(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=1,
module_id=0,
func_name='APPL_TRACE_ERROR',
fmt=None,
args=[],
buffer_args=[],
)
self.assertIn('appl_trace_level >= BT_TRACE_LEVEL_ERROR', stmt)
self.assertIn('BT_LOG_LEVEL_CHECK(APPL, ERROR)', stmt)
def test_appl_debug_level_check(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=1,
module_id=0,
func_name='APPL_TRACE_DEBUG',
fmt=None,
args=[],
buffer_args=[],
)
self.assertIn('appl_trace_level >= BT_TRACE_LEVEL_DEBUG', stmt)
self.assertIn('BT_LOG_LEVEL_CHECK(APPL, DEBUG)', stmt)
def test_buffer_args_generate_buf_print(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=5,
module_id=0,
func_name='APPL_TRACE_DEBUG',
fmt=None,
args=[],
buffer_args=[{'buffer': 'bd_addr', 'length': '6'}],
)
self.assertIn('ble_log_compressed_hex_print_buf', stmt)
self.assertIn('(const uint8_t *)bd_addr, 6', stmt)
def test_multiple_buffer_args(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=6,
module_id=0,
func_name='APPL_TRACE_DEBUG',
fmt=None,
args=[],
buffer_args=[
{'buffer': 'buf1', 'length': '10'},
{'buffer': 'buf2', 'length': '20'},
],
)
self.assertIn('(const uint8_t *)buf1', stmt)
self.assertIn('(const uint8_t *)buf2', stmt)
class TestMeshMacroGen(unittest.TestCase):
mod: types.ModuleType
@classmethod
def setUpClass(cls) -> None:
p = str(SCRIPTS_DIR / 'module_scripts' / 'ble_mesh')
if p not in sys.path:
sys.path.insert(0, p)
import make_mesh_log_macro
cls.mod = make_mesh_log_macro
def test_header_head_has_declarations(self) -> None:
head = self.mod.gen_header_head()
self.assertIn('ble_log_compressed_hex_print', head)
def test_bt_err_level_check(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=1,
module_id=1,
func_name='BT_ERR',
fmt=None,
args=[],
buffer_args=[],
)
self.assertIn('ERROR', stmt)
def test_bt_dbg_level_check(self) -> None:
stmt = self.mod.gen_compressed_stmt(
log_index=2,
module_id=1,
func_name='BT_DBG',
fmt=None,
args=[],
buffer_args=[],
)
self.assertIn('DEBUG', stmt)
if __name__ == '__main__':
unittest.main()

161
components/bt/common/ble_log/extension/log_compression/tests/test_pipeline_e2e.py Normal file
View File

@@ -0,0 +1,161 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""End-to-end pipeline tests: .c input -> .h output with golden file comparison."""
import re
import shutil
import tempfile
import unittest
from test_utils import FIXTURE_DIR
from test_utils import PipelineContext
from test_utils import assert_header_matches_golden
class _E2ETestCase(unittest.TestCase):
"""Base class with pipeline setUp/tearDown."""
def setUp(self) -> None:
self.tmp = tempfile.mkdtemp()
self.ctx = PipelineContext(self.tmp)
def tearDown(self) -> None:
self.ctx.close()
shutil.rmtree(self.tmp, ignore_errors=True)
class TestSimpleLogsE2E(_E2ETestCase):
def test_compresses_and_generates_header(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
macros = self.ctx.run_compression([rel])
self.assertIn('BLE_HOST', macros)
self.assertEqual(len(macros['BLE_HOST']), 4)
self.assertTrue(self.ctx.header_path.exists())
def test_tag_replacement_in_source(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
tmp_file = self.ctx.compressed_srcs / 'test_src' / 'simple_logs.c.tmp'
content = tmp_file.read_text()
self.assertRegex(content, r'APPL_TRACE_DEBUG_\d+')
self.assertRegex(content, r'APPL_TRACE_ERROR_\d+')
def test_header_structure(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
self.assertIn('#ifndef __BLE_HOST_INTERNAL_LOG_INDEX_H', header)
self.assertIn('#define __BLE_HOST_INTERNAL_LOG_INDEX_H', header)
self.assertIn('#endif', header)
self.assertIn('ble_log_compressed_hex_print', header)
def test_macros_sorted_by_id(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
ids = [int(m) for m in re.findall(r'#define \w+_(\d+)\(fmt', header)]
self.assertEqual(ids, sorted(ids))
def test_preserve_tag_appends_original(self) -> None:
rel = self.ctx.copy_fixture('simple_logs.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
self.assertIn('APPL_TRACE_ERROR(fmt, ##__VA_ARGS__)', header)
self.assertIn('APPL_TRACE_WARNING(fmt, ##__VA_ARGS__)', header)
self.assertNotIn('APPL_TRACE_DEBUG(fmt, ##__VA_ARGS__)', header)
self.assertNotIn('APPL_TRACE_API(fmt, ##__VA_ARGS__)', header)
class TestFormatSpecifiersE2E(_E2ETestCase):
def test_all_size_types_in_header(self) -> None:
rel = self.ctx.copy_fixture('format_specifiers.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
self.assertTrue(', 0,' in header or ', 0, ' in header) # U32
self.assertTrue(', 1,' in header or ', 1, ' in header) # STR
self.assertTrue(', 2,' in header or ', 2, ' in header) # U64
class TestSpecialTokensE2E(_E2ETestCase):
def test_func_and_line_in_header(self) -> None:
rel = self.ctx.copy_fixture('special_tokens.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
self.assertIn('ble_log_compressed_hex_print_buf', header)
def test_func_macro_not_in_args(self) -> None:
rel = self.ctx.copy_fixture('special_tokens.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
self.assertNotIn('__func__', header)
class TestNoLogsE2E(_E2ETestCase):
def test_no_macros_generated(self) -> None:
rel = self.ctx.copy_fixture('no_logs.c')
macros = self.ctx.run_compression([rel])
self.assertEqual(len(macros), 0)
class TestMultiLevelE2E(_E2ETestCase):
def test_all_levels_get_unique_ids(self) -> None:
rel = self.ctx.copy_fixture('multi_level.c')
macros = self.ctx.run_compression([rel])
ids = [log_id for log_id, _ in macros['BLE_HOST']]
self.assertEqual(len(ids), len(set(ids)))
def test_different_level_checks(self) -> None:
rel = self.ctx.copy_fixture('multi_level.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
self.assertIn('BT_TRACE_LEVEL_ERROR', header)
self.assertIn('BT_TRACE_LEVEL_WARNING', header)
self.assertIn('BT_TRACE_LEVEL_API', header)
self.assertIn('BT_TRACE_LEVEL_DEBUG', header)
class TestMultiFileE2E(_E2ETestCase):
def test_ids_globally_unique(self) -> None:
rel1 = self.ctx.copy_fixture('simple_logs.c')
rel2 = self.ctx.copy_fixture('multi_level.c')
macros = self.ctx.run_compression([rel1, rel2])
all_ids = [log_id for log_id, _ in macros['BLE_HOST']]
self.assertEqual(len(all_ids), len(set(all_ids)))
self.assertEqual(len(all_ids), 10)
class TestConcatenatedStringsE2E(_E2ETestCase):
def test_pri_macros_produce_u64(self) -> None:
rel = self.ctx.copy_fixture('concatenated_strings.c')
self.ctx.run_compression([rel])
header = self.ctx.header_path.read_text()
self.assertTrue(', 2,' in header or ', 2, ' in header)
class TestGoldenFileComparison(_E2ETestCase):
def _run_and_compare(self, fixture_files: list, golden_name: str) -> None:
rels = [self.ctx.copy_fixture(f) for f in fixture_files]
self.ctx.run_compression(rels)
golden = FIXTURE_DIR / 'expected' / golden_name
if not golden.exists():
self.skipTest(f'Golden file {golden_name} not found. Run update_golden.py first.')
assert_header_matches_golden(self, self.ctx.header_path, golden)
def test_simple_logs_golden(self) -> None:
self._run_and_compare(['simple_logs.c'], 'simple_logs_index.h')
def test_format_specifiers_golden(self) -> None:
self._run_and_compare(['format_specifiers.c'], 'format_specifiers_index.h')
def test_special_tokens_golden(self) -> None:
self._run_and_compare(['special_tokens.c'], 'special_tokens_index.h')
def test_multi_level_golden(self) -> None:
self._run_and_compare(['multi_level.c'], 'multi_level_index.h')
def test_multi_file_golden(self) -> None:
self._run_and_compare(['simple_logs.c', 'multi_level.c'], 'multi_file_index.h')
if __name__ == '__main__':
unittest.main()

203
components/bt/common/ble_log/extension/log_compression/tests/test_utils.py Normal file
View File

@@ -0,0 +1,203 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Shared utilities for BLE log compression tests (unittest-based, no third-party deps)."""
from __future__ import annotations
import difflib
import os
import re
import shutil
import sys
import unittest
from pathlib import Path
from typing import Any
# Add scripts/ to sys.path so test modules can import the production code
SCRIPTS_DIR = Path(__file__).resolve().parent.parent / 'scripts'
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
FIXTURE_DIR = Path(__file__).resolve().parent / 'fixtures'
BLUEDROID_SCRIPT = SCRIPTS_DIR / 'module_scripts' / 'bluedroid' / 'make_bluedroid_log_macro.py'
MESH_SCRIPT = SCRIPTS_DIR / 'module_scripts' / 'ble_mesh' / 'make_mesh_log_macro.py'
BLUEDROID_TAGS = [
'APPL_TRACE_ERROR',
'APPL_TRACE_WARNING',
'APPL_TRACE_API',
'APPL_TRACE_DEBUG',
'APPL_TRACE_EVENT',
'APPL_TRACE_VERBOSE',
]
MESH_TAGS = [
'BT_ERR',
'BT_WARN',
'BT_INFO',
'BT_DBG',
'NET_BUF_ERR',
'NET_BUF_WARN',
'NET_BUF_DBG',
]
def make_db_manager(tmp_dir: str) -> Any:
"""Create a fresh LogDBManager in tmp_dir."""
from LogDBManager import LogDBManager
db_path = os.path.join(tmp_dir, 'db')
return LogDBManager(data_dir=db_path, sources={'BLE_HOST': 'test_config_v1'})
def make_compressor() -> Any:
"""Create a LogCompressor with parser initialized."""
from ble_log_compress import LogCompressor
c = LogCompressor()
c.init_parser()
return c
def write_yaml_config(
tmp_dir: str,
tags: list[str],
script_path: str | Path,
module_name: str = 'BLE_HOST',
log_index_file: str = 'test_log_index.h',
tags_with_preserve: list[str] | None = None,
) -> Path:
"""Write a module_info.yml and return its path."""
import yaml
tags_with_preserve = tags_with_preserve or []
config = {
'log_config': {
'db_path': 'ble_log/ble_log_database',
'modules': {
module_name: {
'description': f'{module_name} Test',
'code_path': ['test_src'],
'log_index_file': log_index_file,
'script': str(script_path),
'tags': tags,
'tags_with_preserve': tags_with_preserve,
}
},
}
}
config_dir = Path(tmp_dir) / 'build' / 'ble_log'
config_dir.mkdir(parents=True, exist_ok=True)
config_path = config_dir / 'module_info.yml'
with open(config_path, 'w') as f:
yaml.safe_dump(config, f)
return config_path
class PipelineContext:
"""Full pipeline context: directory structure + config + DB + compressor."""
def __init__(self, tmp_dir: str) -> None:
from ble_log_compress import LogCompressor
from LogDBManager import LogDBManager
self.tmp_dir = Path(tmp_dir)
self.code_base = self.tmp_dir / 'code_base'
self.build_dir = self.tmp_dir / 'build'
self.compressed_srcs = self.build_dir / 'ble_log' / '.compressed_srcs'
(self.build_dir / 'ble_log' / 'include').mkdir(parents=True, exist_ok=True)
self.compressed_srcs.mkdir(parents=True, exist_ok=True)
self.code_base.mkdir(parents=True, exist_ok=True)
self.config_path = write_yaml_config(
tmp_dir,
tags=BLUEDROID_TAGS,
script_path=BLUEDROID_SCRIPT,
tags_with_preserve=['APPL_TRACE_ERROR', 'APPL_TRACE_WARNING'],
)
self.compressor = LogCompressor()
self.compressor.code_base_path = self.code_base
self.compressor.build_dir = self.build_dir
self.compressor.bt_compressed_srcs_path = self.compressed_srcs
self.compressor.load_config(str(self.config_path), ['BLE_HOST'])
db_path = self.build_dir / self.compressor.config.get('db_path', 'ble_log/ble_log_database')
self.db_manager = LogDBManager(
data_dir=str(db_path),
sources={s: str(c) for s, c in self.compressor.module_info.items()},
)
self.compressor.db_manager = self.db_manager
self.module = 'BLE_HOST'
def copy_fixture(self, filename: str) -> str:
"""Copy a fixture C file into the code base and return its relative path."""
src = FIXTURE_DIR / 'c_sources' / filename
rel_path = Path('test_src') / filename
dst = self.code_base / rel_path
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
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."""
self.compressor.prepare_source_files(src_list)
files_to_process = []
for module, info in self.compressor.module_info.items():
files_to_process.extend([(module, path) for path in info['files_to_process']])
files_to_process.sort(key=lambda x: x[1])
all_macros: dict[str, list[tuple[int, str]]] = {}
for file_info in files_to_process:
file_macros = self.compressor.compress_file(file_info)
for module, log_id, macro in file_macros:
all_macros.setdefault(module, []).append((log_id, macro))
for module, macros in all_macros.items():
self.compressor.generate_log_index_header(module, macros)
# Mark files as processed
for module, info in self.compressor.module_info.items():
for temp_path in info['files_to_process']:
src_path = self.code_base / os.path.relpath(temp_path[:-4], self.compressed_srcs)
self.db_manager.mark_file_processed(module, src_path, temp_path)
# Copy .tmp to final
for root, _, files in os.walk(self.compressed_srcs):
for name in files:
if name.endswith('.tmp'):
file_src = os.path.join(root, name)
dst_path = os.path.join(root, name[:-4])
shutil.copy2(file_src, dst_path)
self.db_manager.save_all()
return all_macros
@property
def header_path(self) -> Path:
return self.build_dir / 'ble_log' / 'include' / 'test_log_index.h'
def close(self) -> None:
self.db_manager.close()
def assert_header_matches_golden(
test_case: unittest.TestCase,
generated_path: str | Path,
golden_path: str | Path,
) -> None:
"""Compare generated header to golden file, normalizing copyright year."""
gen_text = Path(generated_path).read_text()
gold_text = Path(golden_path).read_text()
gen_norm = re.sub(r'20\d{2}', 'YYYY', gen_text)
gold_norm = re.sub(r'20\d{2}', 'YYYY', gold_text)
if gen_norm != gold_norm:
diff = difflib.unified_diff(
gold_norm.splitlines(keepends=True),
gen_norm.splitlines(keepends=True),
fromfile='expected (golden)',
tofile='generated',
)
test_case.fail(f'Header mismatch:\n{"".join(diff)}')

146
components/bt/common/ble_log/extension/log_compression/tests/update_golden.py Normal file
View File

@@ -0,0 +1,146 @@
#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""
Regenerate golden expected output files from fixture C sources.
Usage:
python3 update_golden.py
This runs the compression pipeline on each fixture C source file and copies
the generated header to fixtures/expected/. The output should be reviewed
for correctness before committing.
"""
import shutil
import sys
import tempfile
from pathlib import Path
from typing import List
from typing import Optional
TESTS_DIR = Path(__file__).resolve().parent
SCRIPTS_DIR = TESTS_DIR.parent / 'scripts'
FIXTURE_DIR = TESTS_DIR / 'fixtures'
EXPECTED_DIR = FIXTURE_DIR / 'expected'
sys.path.insert(0, str(SCRIPTS_DIR))
import yaml # noqa: E402
from ble_log_compress import LogCompressor # noqa: E402
from LogDBManager import LogDBManager # noqa: E402
BLUEDROID_SCRIPT = SCRIPTS_DIR / 'module_scripts' / 'bluedroid' / 'make_bluedroid_log_macro.py'
BLUEDROID_TAGS = [
'APPL_TRACE_ERROR',
'APPL_TRACE_WARNING',
'APPL_TRACE_API',
'APPL_TRACE_DEBUG',
'APPL_TRACE_EVENT',
'APPL_TRACE_VERBOSE',
]
def run_pipeline(fixture_files: List[str], output_name: str, tags_with_preserve: Optional[List[str]] = None) -> None:
"""Run compression on fixture files and save the generated header as a golden file."""
tags_with_preserve = tags_with_preserve or ['APPL_TRACE_ERROR', 'APPL_TRACE_WARNING']
with tempfile.TemporaryDirectory() as tmp_str:
tmp = Path(tmp_str)
code_base = tmp / 'code_base'
build_dir = tmp / 'build'
compressed_srcs = build_dir / 'ble_log' / '.compressed_srcs'
(build_dir / 'ble_log' / 'include').mkdir(parents=True)
compressed_srcs.mkdir(parents=True)
# Write config
config = {
'log_config': {
'db_path': 'ble_log/ble_log_database',
'modules': {
'BLE_HOST': {
'description': 'BLE Host Test',
'code_path': ['test_src'],
'log_index_file': 'test_log_index.h',
'script': str(BLUEDROID_SCRIPT),
'tags': BLUEDROID_TAGS,
'tags_with_preserve': tags_with_preserve,
}
},
}
}
config_path = build_dir / 'ble_log' / 'module_info.yml'
with open(config_path, 'w') as f:
yaml.safe_dump(config, f)
# Copy fixtures
rel_paths = []
for fname in fixture_files:
src = FIXTURE_DIR / 'c_sources' / fname
rel = Path('test_src') / fname
dst = code_base / rel
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
rel_paths.append(str(rel))
# Init compressor
comp = LogCompressor()
comp.code_base_path = code_base
comp.build_dir = build_dir
comp.bt_compressed_srcs_path = compressed_srcs
comp.load_config(str(config_path), ['BLE_HOST'])
db_path = build_dir / comp.config.get('db_path', 'ble_log/ble_log_database')
db = LogDBManager(
data_dir=str(db_path),
sources={s: str(c) for s, c in comp.module_info.items()},
)
comp.db_manager = db
# Run compression
comp.prepare_source_files(rel_paths)
files_to_process = []
for module, info in comp.module_info.items():
files_to_process.extend([(module, p) for p in info['files_to_process']])
files_to_process.sort(key=lambda x: x[1])
all_macros: dict[str, list[tuple[int, str]]] = {}
for fi in files_to_process:
for module, log_id, macro in comp.compress_file(fi):
all_macros.setdefault(module, []).append((log_id, macro))
for module, macros in all_macros.items():
comp.generate_log_index_header(module, macros)
db.close()
# Copy generated header to expected
header = build_dir / 'ble_log' / 'include' / 'test_log_index.h'
if header.exists():
dest = EXPECTED_DIR / output_name
shutil.copy2(header, dest)
print(f'Generated: {dest}')
else:
print(f'No header generated for {fixture_files}')
def main() -> None:
EXPECTED_DIR.mkdir(parents=True, exist_ok=True)
scenarios = [
(['simple_logs.c'], 'simple_logs_index.h'),
(['format_specifiers.c'], 'format_specifiers_index.h'),
(['special_tokens.c'], 'special_tokens_index.h'),
(['multi_level.c'], 'multi_level_index.h'),
(['simple_logs.c', 'multi_level.c'], 'multi_file_index.h'),
]
for files, output in scenarios:
run_pipeline(files, output)
print(f'\nAll golden files generated in {EXPECTED_DIR}')
print('Review the output for correctness before committing.')
if __name__ == '__main__':
main()