1#
2# Copyright (c) 2024 Intel Corporation.
3#
4# SPDX-License-Identifier: Apache-2.0
5#
6
7'''
8Pytest harness to test the output of the dictionary logging.
9'''
10
11import logging
12import os
13import re
14import shlex
15import subprocess
16
17from twister_harness import DeviceAdapter
18
19ZEPHYR_BASE = os.getenv("ZEPHYR_BASE")
20
21logger = logging.getLogger(__name__)
22
23def process_logs(dut: DeviceAdapter, build_dir):
24    '''
25    This grabs the encoded log from console and parse the log
26    through the dictionary logging parser.
27
28    Returns the decoded log lines.
29    '''
30    # Make sure the log parser script is there...
31    parser_script = os.path.join(ZEPHYR_BASE, "scripts", "logging", "dictionary", "log_parser.py")
32    assert os.path.isfile(parser_script)
33    logger.info(f'Log parser script: {parser_script}')
34
35    # And also the dictionary JSON file is there...
36    dictionary_json = os.path.join(build_dir, "zephyr", "log_dictionary.json")
37    assert os.path.isfile(dictionary_json)
38    logger.info(f'Dictionary JSON: {dictionary_json}')
39
40    # Read the encoded logs and save them to a file
41    # as the log parser requires file as input.
42    # Timeout is intentionally long. Twister will
43    # timeout earlier with per-test timeout.
44    handler_output = dut.readlines_until(regex = '.*##ZLOGV1##[0-9]+', timeout = 600.0)
45
46    # Join all the output lines together
47    handler_output = ''.join(ho.strip() for ho in handler_output)
48
49    # Find the last dictionary logging block and extract it
50    ridx = handler_output.rfind("##ZLOGV1##")
51    encoded_logs = handler_output[ridx:]
52
53    encoded_log_file = os.path.join(build_dir, "encoded.log")
54    with open(encoded_log_file, 'w', encoding='utf-8') as fp:
55        fp.write(encoded_logs)
56
57    # Run the log parser
58    cmd = [parser_script, '--hex', dictionary_json, encoded_log_file]
59    logger.info(f'Running parser script: {shlex.join(cmd)}')
60    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
61    assert result.returncode == 0
62
63    # Grab the decoded log lines from stdout, print a copy and return it
64    decoded_logs = result.stdout
65    logger.info(f'Decoded logs: {decoded_logs}')
66
67    return decoded_logs
68
69
70def expected_regex_common():
71    '''
72    Return an array of compiled regular expression for matching
73    the decoded log lines.
74    '''
75    return [
76    # *** Booting Zephyr OS build <version> ***
77    re.compile(r'.*[*][*][*] Booting Zephyr OS build [0-9a-z.-]+'),
78    # Hello World! <board name>
79    re.compile(r'[\s]+Hello World! [\w-]+'),
80    # [        10] <err> hello_world: error string
81    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <err> hello_world: error string'),
82    # [        10] <dbg> hello_world: main: debug string
83    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: debug string'),
84    # [        10] <inf> hello_world: info string
85    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <inf> hello_world: info string'),
86    # [        10] <dbg> hello_world: main: int8_t 1, uint8_t 2
87    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: int8_t 1, uint8_t 2'),
88    # [        10] <dbg> hello_world: main: int16_t 16, uint16_t 17
89    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: int16_t 16, uint16_t 17'),
90    # [        10] <dbg> hello_world: main: int32_t 32, uint32_t 33
91    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: int32_t 32, uint32_t 33'),
92    # [        10] <dbg> hello_world: main: int64_t 64, uint64_t 65
93    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: int64_t 64, uint64_t 65'),
94    # [        10] <dbg> hello_world: main: char !
95    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: char !'),
96    # [        10] <dbg> hello_world: main: s str static str c str
97    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: s str static str c str'),
98    # [        10] <dbg> hello_world: main: d str dynamic str
99    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: d str dynamic str'),
100    # [        10] <dbg> hello_world: main: mixed str dynamic str --- dynamic str \
101    # --- another dynamic str --- another dynamic str
102    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: mixed str dynamic str '
103                '--- dynamic str --- another dynamic str --- another dynamic str'),
104    # [        10] <dbg> hello_world: main: mixed c/s ! static str dynamic str static str !
105    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: mixed c/s ! static str '
106                'dynamic str static str !'),
107    # [        10] <dbg> hello_world: main: pointer 0x1085f9
108    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: pointer 0x[0-9a-f]+'),
109    # [        10] <dbg> hello_world: main: For HeXdUmP!
110    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: For HeXdUmP!'),
111    # 48 45 58 44 55 4d 50 21  20 48 45 58 44 55 4d 50 |HEXDUMP!  HEXDUMP
112    re.compile(r'[\s]+[ ]+[0-9a-f ]{48,52}[|]HEXDUMP!  HEXDUMP'),
113    # 40 20 48 45 58 44 55 4d  50 23                   |@ HEXDUM P#
114    re.compile(r'[\s]+[ ]+[0-9a-f ]{48,52}[|]@ HEXDUM P#'),
115    ]
116
117
118def expected_regex_fpu():
119    '''
120    Return an array of additional compiled regular expression for matching
121    the decoded log lines for FPU builds.
122    '''
123    return [
124    # [        10] <dbg> hello_world: main: float 66.669998, double 68.690000
125    re.compile(r'[\s]+[\[][0-9,:\. ]+[\]] <dbg> hello_world: main: '
126               r'float 66[\.][0-9-\.]+, double 68[\.][0-9-\.]+'),
127    ]
128
129
130def regex_matching(decoded_logs, expected_regex):
131    '''
132    Given the decoded log lines and an array of compiled regular expression,
133    match all of them and display whether a line is found or not.
134
135    Return True if all regular expressions have corresponding matches,
136    False otherwise.
137    '''
138    regex_results = [ex_re.search(decoded_logs) for ex_re in expected_regex]
139
140    # Using 1:1 mapping between regex_results and expected_regex, so
141    # cannot use enumeration.
142    #
143    # pylint: disable=consider-using-enumerate
144    for idx in range(len(regex_results)):
145        if regex_results[idx]:
146            logger.info(f'Found: {regex_results[idx].group(0).strip()}')
147        else:
148            logger.info(f'NOT FOUND: {expected_regex[idx]}')
149
150    return all(regex_results)
151
152
153def test_logging_dictionary(dut: DeviceAdapter, is_fpu_build):
154    '''
155    Main entrance to setup test result validation.
156    '''
157    build_dir = dut.device_config.app_build_dir
158
159    logger.info(f'FPU build? {is_fpu_build}')
160
161    decoded_logs = process_logs(dut, build_dir)
162
163    assert regex_matching(decoded_logs, expected_regex_common())
164
165    if is_fpu_build:
166        assert regex_matching(decoded_logs, expected_regex_fpu())
167