tests/test_plotter.py

#    Copyright 2015-2015 ARM Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#


import unittest
import matplotlib
import pandas as pd
import tempfile
import os
import warnings

from test_thermal import BaseTestThermal
import trappy


class TestPlotter(BaseTestThermal):

    """No Bombing testcases for plotter"""

    def __init__(self, *args, **kwargs):
        super(TestPlotter, self).__init__(*args, **kwargs)

    def test_plot_no_pivot(self):
        """Tests LinePlot with no pivot"""
        trace1 = trappy.FTrace(name="first")
        l = trappy.LinePlot(trace1, trappy.thermal.Thermal, column="temp")
        l.view(test=True)

    def test_plot_multi_trace(self):
        """Tests LinePlot with no Pivot multi traces"""
        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")
        l = trappy.LinePlot(
            [trace1, trace2], trappy.thermal.Thermal, column="temp")
        l.view(test=True)

    def test_plot_multi(self):
        """Tests LinePlot with no Pivot multi attrs"""
        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")
        l = trappy.LinePlot([trace1,
                          trace2],
                         [trappy.thermal.Thermal,
                          trappy.thermal.ThermalGovernor],
                         column=["temp",
                                 "power_range"])
        l.view(test=True)

    def test_plot_filter(self):
        """Tests LinePlot with no Pivot with filters"""
        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")
        l = trappy.LinePlot([trace1,
                          trace2],
                         [trappy.cpu_power.CpuOutPower],
                         column=["power"],
                         filters={"cdev_state": [1]})
        l.view(test=True)

    def test_plot_pivot(self):
        """Tests LinePlot with Pivot"""
        trace1 = trappy.FTrace(name="first")
        l = trappy.LinePlot(
            trace1,
            trappy.thermal.Thermal,
            column="temp",
            pivot="thermal_zone")
        l.view(test=True)

    def test_plot_multi_trace_pivot(self):
        """Tests LinePlot with Pivot multi traces"""
        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")
        l = trappy.LinePlot(
            [trace1, trace2], trappy.cpu_power.CpuOutPower, column="power", pivot="cpus")
        l.view(test=True)

    def test_plot_multi_pivot(self):
        """Tests LinePlot with Pivot with multi attrs"""
        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")
        l = trappy.LinePlot([trace1,
                          trace2],
                         [trappy.cpu_power.CpuInPower,
                          trappy.cpu_power.CpuOutPower],
                         column=["dynamic_power",
                                 "power"],
                         pivot="cpus")
        l.view(test=True)

    def test_plot_multi_pivot_filter(self):
        """Tests LinePlot with Pivot and filters"""
        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")
        l = trappy.LinePlot(
            trace1,
            trappy.cpu_power.CpuInPower,
            column=[
                "dynamic_power",
                "load1"],
            filters={
                "cdev_state": [
                    1,
                    0]},
            pivot="cpus")
        l.view(test=True)

    def test_plot_savefig(self):
        """Tests plotter: savefig"""
        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")
        l = trappy.LinePlot(
            trace1,
            trappy.cpu_power.CpuInPower,
            column=[
                "dynamic_power",
                "load1"],
            filters={
                "cdev_state": [
                    1,
                    0]},
            pivot="cpus")
        png_file = tempfile.mktemp(dir="/tmp", suffix=".png")
        l.savefig(png_file)
        self.assertTrue(os.path.isfile(png_file))
        os.remove(png_file)


    def test_signals(self):
        """Test signals input for LinePlot"""

        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")

        l = trappy.LinePlot([trace1,
                          trace2],
                         signals=["cpu_in_power:dynamic_power",
                                 "cpu_out_power:power"],
                         pivot="cpus")

        l.view(test=True)


    def test_signals_exceptions(self):
        """Test incorrect input combinations: signals"""

        trace1 = trappy.FTrace(name="first")
        trace2 = trappy.FTrace(name="second")

        with self.assertRaises(ValueError):
            l = trappy.LinePlot([trace1, trace2],
                            column=[
                                "dynamic_power",
                                "load1"],
                            signals=["cpu_in_power:dynamic_power",
                                 "cpu_out_power:power"],
                            pivot="cpus")

        with self.assertRaises(ValueError):
            l = trappy.LinePlot([trace1, trace2],
                            trappy.cpu_power.CpuInPower,
                            signals=["cpu_in_power:dynamic_power",
                                 "cpu_out_power:power"],
                            pivot="cpus")

        with self.assertRaises(ValueError):
            l = trappy.LinePlot([trace1, trace2],
                            trappy.cpu_power.CpuInPower,
                            column=[
                                "dynamic_power",
                                "load1"],
                            signals=["cpu_in_power:dynamic_power",
                                 "cpu_out_power:power"],
                            pivot="cpus")

    def test_get_trace_event_data_corrupted_trace(self):
        """get_trace_event_data() works with a corrupted trace"""
        from trappy.plotter.Utils import get_trace_event_data

        trace = trappy.FTrace()

        # We create this trace:
        #
        # 1 15414 -> 15411
        # 2 15411 -> 15414
        # 3 15414 -> 15411 (corrupted, should be dropped)
        # 4 15413 -> 15411
        # 5 15411 -> 15413
        #
        # Which should plot like:
        #
        # CPU
        #    +-------+-------+
        #  0 | 15411 | 15414 |
        #    +-------+-------+       +-------+
        #  1                         | 15411 |
        #                            +-------+
        #    +-------+-------+-------+-------+
        #   0.1     0.2     0.3     0.4     0.5

        broken_trace = pd.DataFrame({
            '__comm': ["task2", "task1", "task2", "task3", "task1"],
            '__cpu':  [0, 0, 0, 1, 1],
            '__pid':  [15414, 15411, 15414, 15413, 15411],
            'next_comm': ["task1", "task2", "task1", "task1", "task3"],
            'next_pid':  [15411, 15414, 15411, 15411, 15413],
            'prev_comm': ["task2", "task1", "task2", "task3", "task1"],
            'prev_pid':  [15414, 15411, 15414, 15413, 15411],
            'prev_state': ["S", "R", "S", "S", "S"]},
            index=pd.Series(range(1, 6), name="Time"))

        trace.sched_switch.data_frame = broken_trace

        with warnings.catch_warnings(record=True) as warn:
            data, procs, window = get_trace_event_data(trace)
            self.assertEquals(len(warn), 1)

            warn_str = str(warn[-1])
            self.assertTrue("15411" in warn_str)
            self.assertTrue("4" in warn_str)

        zipped_comms = zip(broken_trace["next_comm"], broken_trace["next_pid"])
        expected_procs = set("-".join([comm, str(pid)]) for comm, pid in zipped_comms)

        self.assertTrue([1, 2, 0] in data["task1-15411"])
        self.assertTrue([2, 3, 0] in data["task2-15414"])
        self.assertTrue([4, 5, 1] in data["task1-15411"])
        self.assertEquals(procs, expected_procs)
        self.assertEquals(window, [1, 5])