1from collections import defaultdict
2import lldb
3import json
4from intelpt_testcase import *
5from lldbsuite.test.lldbtest import *
6from lldbsuite.test import lldbutil
7from lldbsuite.test.decorators import *
8import os
9
10class TestTraceExport(TraceIntelPTTestCaseBase):
11
12    mydir = TestBase.compute_mydir(__file__)
13
14    def testErrorMessages(self):
15        ctf_test_file = self.getBuildArtifact("ctf-test.json")
16        # We first check the output when there are no targets
17        self.expect(f"thread trace export ctf --file {ctf_test_file}",
18            substrs=["error: invalid target, create a target using the 'target create' command"],
19            error=True)
20
21        # We now check the output when there's a non-running target
22        self.expect("target create " +
23            os.path.join(self.getSourceDir(), "intelpt-trace", "a.out"))
24
25        self.expect(f"thread trace export ctf --file {ctf_test_file}",
26            substrs=["error: invalid process"],
27            error=True)
28
29        # Now we check the output when there's a running target without a trace
30        self.expect("b main")
31        self.expect("run")
32
33        self.expect(f"thread trace export ctf --file {ctf_test_file}",
34            substrs=["error: Process is not being traced"],
35            error=True)
36
37    def testExportCreatesFile(self):
38        self.expect("trace load -v " +
39            os.path.join(self.getSourceDir(), "intelpt-trace", "trace.json"),
40            substrs=["intel-pt"])
41
42        ctf_test_file = self.getBuildArtifact("ctf-test.json")
43
44        if os.path.exists(ctf_test_file):
45            remove_file(ctf_test_file)
46        self.expect(f"thread trace export ctf --file {ctf_test_file}")
47        self.assertTrue(os.path.exists(ctf_test_file))
48
49
50    def testHtrBasicSuperBlockPass(self):
51        '''
52        Test the BasicSuperBlock pass of HTR
53
54        TODO: Once the "trace save" command is implemented, gather Intel PT
55        trace of this program and load it like the other tests instead of
56        manually executing the commands to trace the program.
57        '''
58        self.expect(f"target create {os.path.join(self.getSourceDir(), 'intelpt-trace', 'export_ctf_test_program.out')}")
59        self.expect("b main")
60        self.expect("r")
61        self.expect("b exit")
62        self.expect("thread trace start")
63        self.expect("c")
64
65        ctf_test_file = self.getBuildArtifact("ctf-test.json")
66
67        if os.path.exists(ctf_test_file):
68            remove_file(ctf_test_file)
69        self.expect(f"thread trace export ctf --file {ctf_test_file}")
70        self.assertTrue(os.path.exists(ctf_test_file))
71
72
73        with open(ctf_test_file) as f:
74            data = json.load(f)
75
76        num_units_by_layer = defaultdict(int)
77        index_of_first_layer_1_block = None
78        for i, event in enumerate(data):
79            layer_id = event.get('pid')
80            if layer_id == 1 and index_of_first_layer_1_block is None:
81                index_of_first_layer_1_block = i
82            if layer_id is not None and event['ph'] == 'B':
83                num_units_by_layer[layer_id] += 1
84
85        # Check that there are two layers
86        self.assertTrue(0 in num_units_by_layer and 1 in num_units_by_layer)
87        # Check that each layer has the correct total number of blocks
88        self.assertTrue(num_units_by_layer[0] == 1630)
89        self.assertTrue(num_units_by_layer[1] == 383)
90
91
92        expected_block_names = [
93                '0x4005f0',
94                '0x4005fe',
95                '0x400606: iterative_handle_request_by_id(int, int)',
96                '0x4005a7',
97                '0x4005af',
98                '0x4005b9: fast_handle_request(int)',
99                '0x4005d5: log_response(int)',
100        ]
101        # There are two events per block, a beginning and an end. This means we must increment data_index by 2, so we only encounter the beginning event of each block.
102        data_index = index_of_first_layer_1_block
103        expected_index = 0
104        while expected_index < len(expected_block_names):
105            self.assertTrue(data[data_index]['name'] == expected_block_names[expected_index])
106            self.assertTrue(data[data_index]['name'] == expected_block_names[expected_index])
107            data_index += 2
108            expected_index += 1
109
110