• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#    Copyright 2015-2017 ARM Limited
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14#
15
16
17import os
18import sys
19
20import utils_tests
21import trappy
22
23sys.path.append(os.path.join(utils_tests.TESTS_DIRECTORY, "..", "trappy"))
24
25class BaseTestSched(utils_tests.SetupDirectory):
26    def __init__(self, *args, **kwargs):
27        super(BaseTestSched, self).__init__(
28             [("trace_sched.txt", "trace.txt"),],
29             *args,
30             **kwargs)
31
32class TestSchedLoadAvgSchedGroup(BaseTestSched):
33
34    def test_get_dataframe(self):
35        """Test that SchedLoadAvgSchedGroup creates a proper data_frame"""
36        dfr = trappy.FTrace().sched_load_avg_sg.data_frame
37
38        self.assertTrue(len(dfr) == 1)
39        self.assertEquals(dfr["cpus"].iloc[0], "00000002")
40        self.assertEquals(dfr["load"].iloc[0], 0)
41        self.assertEquals(dfr["utilization"].iloc[0], 0)
42
43class TestSchedLoadAvgTask(BaseTestSched):
44
45    def test_get_dataframe(self):
46        """Test that SchedLoadAvgTask creates a proper data_frame"""
47        dfr = trappy.FTrace().sched_load_avg_task.data_frame
48
49        self.assertTrue(len(dfr) == 1)
50        self.assertEquals(dfr["comm"].iloc[0], "sshd")
51        self.assertEquals(dfr["pid"].iloc[0], 2962)
52        self.assertEquals(dfr["load"].iloc[0], 0)
53        self.assertEquals(dfr["utilization"].iloc[0], 0)
54        self.assertEquals(dfr["runnable_avg_sum"].iloc[0], 0)
55        self.assertEquals(dfr["running_avg_sum"].iloc[0], 0)
56        self.assertEquals(dfr["avg_period"].iloc[0], 48595)
57
58class TestSchedLoadAvgCpu(BaseTestSched):
59
60    def test_get_dataframe(self):
61        """Test that SchedLoadAvgCpu creates a proper data_frame"""
62        dfr = trappy.FTrace().sched_load_avg_cpu.data_frame
63
64        self.assertTrue(len(dfr) == 1)
65        self.assertEquals(dfr["cpu"].iloc[0], 0)
66        self.assertEquals(dfr["load"].iloc[0], 13)
67        self.assertEquals(dfr["utilization"].iloc[0], 18)
68
69class TestSchedContribScaleFactor(BaseTestSched):
70
71    def test_get_dataframe(self):
72        """Test that SchedContribScaleFactor creates a proper data_frame"""
73        dfr = trappy.FTrace().sched_contrib_scale_factor.data_frame
74
75        self.assertTrue(len(dfr) == 1)
76        self.assertEquals(dfr["cpu"].iloc[0], 0)
77        self.assertEquals(dfr["freq_scale_factor"].iloc[0], 426)
78        self.assertEquals(dfr["cpu_scale_factor"].iloc[0], 1024)
79
80class TestSchedCpuCapacity(BaseTestSched):
81
82    def test_get_dataframe(self):
83        """Test that SchedCpuCapacity creates a proper data_frame"""
84        dfr = trappy.FTrace().cpu_capacity.data_frame
85
86        self.assertTrue(len(dfr) == 1)
87        self.assertEquals(dfr["cpu"].iloc[0], 3)
88        self.assertEquals(dfr["capacity"].iloc[0], 430)
89        self.assertEquals(dfr["rt_capacity"].iloc[0], 1024)
90
91class TestSchedCpuFrequency(BaseTestSched):
92
93    def test_get_dataframe(self):
94        """Test that SchedCpuFrequency creates a proper data_frame"""
95        dfr = trappy.FTrace().cpu_frequency.data_frame
96
97        self.assertTrue(len(dfr) == 1)
98        self.assertEquals(dfr["cpu"].iloc[0], 0)
99        self.assertEquals(dfr["frequency"].iloc[0], 600000)
100        self.assertFalse("cpu_id" in dfr.columns)
101
102class TestSchedWakeup(BaseTestSched):
103
104    def test_get_dataframe(self):
105        """Test that SchedWakeup creates a proper data_frame"""
106        dfr = trappy.FTrace().sched_wakeup.data_frame
107
108        self.assertTrue(len(dfr) == 2)
109        self.assertEquals(dfr["comm"].iloc[0], "rcu_preempt")
110        self.assertEquals(dfr["pid"].iloc[0], 7)
111        self.assertEquals(dfr["prio"].iloc[0], 120)
112        self.assertEquals(dfr["success"].iloc[0], 1)
113        self.assertEquals(dfr["target_cpu"].iloc[0], 1)
114
115class TestSchedWakeupNew(BaseTestSched):
116
117    def test_get_dataframe(self):
118        """Test that SchedWakeupNew creates a proper data_frame"""
119        dfr = trappy.FTrace().sched_wakeup_new.data_frame
120
121        self.assertTrue(len(dfr) == 2)
122        self.assertEquals(dfr["comm"].iloc[0], "shutils")
123        self.assertEquals(dfr["pid"].iloc[0], 19428)
124        self.assertEquals(dfr["prio"].iloc[0], 120)
125        self.assertEquals(dfr["success"].iloc[0], 1)
126        self.assertEquals(dfr["target_cpu"].iloc[0], 2)
127
128
129class TestGetFilters(BaseTestSched):
130
131    def test_get_filters(self):
132        """Test that FTrace::get_filters returns correct list of filters"""
133
134        trace = trappy.FTrace()
135        classes = trace.class_definitions
136        filters = trace.get_filters()
137        self.assertTrue(len(classes) == len(filters))
138        self.assertTrue(sorted(classes) == sorted(filters))
139
140        sched_classes = trace.sched_classes.copy()
141        sched_filters = trace.get_filters("sched")
142
143        # cpu_capacity and cpu_frequency are in the sched scope but they should
144        # not be captured by get_filters("sched")
145        del sched_classes["cpu_capacity"]
146        del sched_classes["cpu_frequency"]
147
148        self.assertTrue(len(sched_classes) == len(sched_filters))
149        self.assertTrue(sorted(sched_classes) == sorted(sched_filters))
150
151class TestSpacedValueAttributes(BaseTestSched):
152
153    def test_spaced_value_attr(self):
154        """Test that FTrace object parses spaced value attributes correctly"""
155
156        with open("trace.txt", "a") as fout:
157            fout.write("       <...>-2971  [004]  6550.056871: sched_load_avg_task:  comm=AsyncTask #2 pid=6163 ")
158
159        dfr = trappy.FTrace().sched_load_avg_task.data_frame
160        self.assertTrue(len(dfr) == 2)
161        self.assertEquals(dfr["comm"].iloc[1], "AsyncTask #2")
162        self.assertEquals(dfr["pid"].iloc[1], 6163)
163
164class TestNoSchedTraces(utils_tests.SetupDirectory):
165
166    def __init__(self, *args, **kwargs):
167        super(TestNoSchedTraces, self).__init__(
168             [("trace_empty.txt", "trace.txt")],
169             *args,
170             **kwargs)
171
172    def test_empty_trace_txt(self):
173        """Test that empty objects are created with empty trace file"""
174
175        trace = trappy.FTrace()
176
177        for attr in trace.sched_classes.iterkeys():
178            self.assertTrue(len(getattr(trace, attr).data_frame) == 0)
179