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