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 17 import os 18 import sys 19 20 import utils_tests 21 import trappy 22 23 sys.path.append(os.path.join(utils_tests.TESTS_DIRECTORY, "..", "trappy")) 24 25 class 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 32 class 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 43 class 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 58 class 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 69 class 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 80 class 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 91 class 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 102 class 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 115 class 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 129 class 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 151 class 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 164 class 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
