text: # Copyright 2016 The Chromium Authors. All rights reserved.
  # Use of this source code is governed by a BSD-style license that can be
  # found in the LICENSE file.
  
  import collections
  import copy
  import logging
  import operator
  import unittest
  
  import devtools_monitor
  
  from tracing import (Event, TracingTrack, _IntervalTree)
  
  
  class TracingTrackTestCase(unittest.TestCase):
    _MIXED_EVENTS = [
        {'ts': 3, 'ph': 'N', 'id': 1, 'args': {'name': 'A'}},
        {'ts': 5, 'ph': 'X', 'dur': 1, 'args': {'name': 'B'}},
        {'ts': 7, 'ph': 'D', 'id': 1},
        {'ts': 10, 'ph': 'B', 'args': {'name': 'D'}},
        {'ts': 10, 'ph': 'b', 'cat': 'X', 'id': 1, 'args': {'name': 'C'}},
        {'ts': 11, 'ph': 'e', 'cat': 'X', 'id': 1},
        {'ts': 12, 'ph': 'E'},
        {'ts': 12, 'ph': 'N', 'id': 1, 'args': {'name': 'E'}},
        {'ts': 13, 'ph': 'b', 'cat': 'X', 'id': 2, 'args': {'name': 'F'}},
        {'ts': 14, 'ph': 'e', 'cat': 'X', 'id': 2},
        {'ts': 15, 'ph': 'D', 'id': 1}]
  
    _EVENTS = [
        {'ts': 5, 'ph': 'X', 'dur': 1, 'pid': 2, 'tid': 1, 'args': {'name': 'B'}},
        {'ts': 3, 'ph': 'X', 'dur': 4, 'pid': 2, 'tid': 1, 'args': {'name': 'A'}},
        {'ts': 10, 'ph': 'X', 'dur': 1, 'pid': 2, 'tid': 2,
         'args': {'name': 'C'}},
        {'ts': 10, 'ph': 'X', 'dur': 2, 'pid': 2, 'tid': 2,
         'args': {'name': 'D'}},
        {'ts': 13, 'ph': 'X', 'dur': 1, 'pid': 2, 'tid': 1,
         'args': {'name': 'F'}},
        {'ts': 12, 'ph': 'X', 'dur': 3, 'pid': 2, 'tid': 1,
         'args': {'name': 'E'}}]
  
    def setUp(self):
      self.tree_threshold = _IntervalTree._TRESHOLD
      _IntervalTree._TRESHOLD = 2  # Expose more edge cases in the tree.
      self.track = TracingTrack(None)
  
    def tearDown(self):
      _IntervalTree._TRESHOLD = self.tree_threshold
  
    def EventToMicroseconds(self, event):
      result = copy.deepcopy(event)
      if 'ts' in result:
        result['ts'] *= 1000
      if 'dur' in result:
        result['dur'] *= 1000
      return result
  
    def CheckTrack(self, timestamp, names):
      self.track._IndexEvents(strict=True)
      self.assertEqual(
          set((e.args['name'] for e in self.track.EventsAt(timestamp))),
          set(names))
  
    def CheckIntervals(self, events):
      """All tests should produce the following sequence of intervals, each
      identified by a 'name' in the event args.
  
      Timestamp
      3    |      A
      4    |
      5    | |    B
      6    |
      7
      ..
      10   | |    C, D
      11     |
      12   |      E
      13   | |    F
      14   |
      """
      self.track.Handle('Tracing.dataCollected',
                        {'params': {'value': [self.EventToMicroseconds(e)
                                              for e in events]}})
      self.CheckTrack(0, '')
      self.CheckTrack(2, '')
      self.CheckTrack(3, 'A')
      self.CheckTrack(4, 'A')
      self.CheckTrack(5, 'AB')
      self.CheckTrack(6, 'A')
      self.CheckTrack(7, '')
      self.CheckTrack(9, '')
      self.CheckTrack(10, 'CD')
      self.CheckTrack(11, 'D')
      self.CheckTrack(12, 'E')
      self.CheckTrack(13, 'EF')
      self.CheckTrack(14, 'E')
      self.CheckTrack(15, '')
      self.CheckTrack(100, '')
  
    def testComplete(self):
      # These are deliberately out of order.
      self.CheckIntervals([
          {'ts': 5, 'ph': 'X', 'dur': 1, 'args': {'name': 'B'}},
          {'ts': 3, 'ph': 'X', 'dur': 4, 'args': {'name': 'A'}},
          {'ts': 10, 'ph': 'X', 'dur': 1, 'args': {'name': 'C'}},
          {'ts': 10, 'ph': 'X', 'dur': 2, 'args': {'name': 'D'}},
          {'ts': 13, 'ph': 'X', 'dur': 1, 'args': {'name': 'F'}},
          {'ts': 12, 'ph': 'X', 'dur': 3, 'args': {'name': 'E'}}])
  
    def testDuration(self):
      self.CheckIntervals([
          {'ts': 3, 'ph': 'B', 'args': {'name': 'A'}},
          {'ts': 5, 'ph': 'B', 'args': {'name': 'B'}},
          {'ts': 6, 'ph': 'E'},
          {'ts': 7, 'ph': 'E'},
          # Since async intervals aren't named and must be nested, we fudge the
          # beginning of D by a tenth to ensure it's consistently detected as the
          # outermost event.
          {'ts': 9.9, 'ph': 'B', 'args': {'name': 'D'}},
          {'ts': 10, 'ph': 'B', 'args': {'name': 'C'}},
          {'ts': 11, 'ph': 'E'},
          # End of D. As end times are exclusive this should not conflict with the
          # start of E.
          {'ts': 12, 'ph': 'E'},
          {'ts': 12, 'ph': 'B', 'args': {'name': 'E'}},
          {'ts': 13, 'ph': 'B', 'args': {'name': 'F'}},
          {'ts': 14, 'ph': 'E'},
          {'ts': 15, 'ph': 'E'}])
  
    def testBadDurationExtraBegin(self):
      self.assertRaises(devtools_monitor.DevToolsConnectionException,
                        self.CheckIntervals,
                        [{'ts': 3, 'ph': 'B'},
                         {'ts': 4, 'ph': 'B'},
                         {'ts': 5, 'ph': 'E'}])
  
    def testBadDurationExtraEnd(self):
      self.assertRaises(devtools_monitor.DevToolsConnectionException,
                        self.CheckIntervals,
                        [{'ts': 3, 'ph': 'B'},
                         {'ts': 4, 'ph': 'E'},
                         {'ts': 5, 'ph': 'E'}])
  
    def testAsync(self):
      self.CheckIntervals([
          # A, B and F have the same category/id (so that A & B nest); C-E do not.
          {'ts': 3, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'A'}},
          {'ts': 5, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'B'}},
          # Not indexable.
          {'ts': 4, 'ph': 'n', 'cat': 'A', 'id': 1, 'args': {'name': 'A'}},
          {'ts': 6, 'ph': 'e', 'cat': 'A', 'id': 1},
          {'ts': 7, 'ph': 'e', 'cat': 'A', 'id': 1},
          {'ts': 10, 'ph': 'b', 'cat': 'B', 'id': 2, 'args': {'name': 'D'}},
          {'ts': 10, 'ph': 'b', 'cat': 'B', 'id': 3, 'args': {'name': 'C'}},
          {'ts': 11, 'ph': 'e', 'cat': 'B', 'id': 3},
          {'ts': 12, 'ph': 'e', 'cat': 'B', 'id': 2},
          {'ts': 12, 'ph': 'b', 'cat': 'A', 'id': 2, 'args': {'name': 'E'}},
          {'ts': 13, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'F'}},
          {'ts': 14, 'ph': 'e', 'cat': 'A', 'id': 1},
          {'ts': 15, 'ph': 'e', 'cat': 'A', 'id': 2}])
  
    def testBadAsyncIdMismatch(self):
      self.assertRaises(
          devtools_monitor.DevToolsConnectionException,
          self.CheckIntervals,
          [{'ts': 3, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'A'}},
           {'ts': 5, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'B'}},
           {'ts': 6, 'ph': 'e', 'cat': 'A', 'id': 2},
           {'ts': 7, 'ph': 'e', 'cat': 'A', 'id': 1}])
  
    def testBadAsyncExtraBegin(self):
      self.assertRaises(
          devtools_monitor.DevToolsConnectionException,
          self.CheckIntervals,
          [{'ts': 3, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'A'}},
           {'ts': 5, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'B'}},
           {'ts': 6, 'ph': 'e', 'cat': 'A', 'id': 1}])
  
    def testBadAsyncExtraEnd(self):
      self.assertRaises(
          devtools_monitor.DevToolsConnectionException,
          self.CheckIntervals,
          [{'ts': 3, 'ph': 'b', 'cat': 'A', 'id': 1, 'args': {'name': 'A'}},
           {'ts': 5, 'ph': 'e', 'cat': 'A', 'id': 1},
           {'ts': 6, 'ph': 'e', 'cat': 'A', 'id': 1}])
  
    def testObject(self):
      # A and E share ids, which is okay as their scopes are disjoint.
      self.CheckIntervals([
          {'ts': 3, 'ph': 'N', 'id': 1, 'args': {'name': 'A'}},
          {'ts': 5, 'ph': 'N', 'id': 2, 'args': {'name': 'B'}},
          {'ts': 6, 'ph': 'D', 'id': 2},
          {'ts': 6, 'ph': 'O', 'id': 2},  #  Ignored.
          {'ts': 7, 'ph': 'D', 'id': 1},
          {'ts': 10, 'ph': 'N', 'id': 3, 'args': {'name': 'D'}},
          {'ts': 10, 'ph': 'N', 'id': 4, 'args': {'name': 'C'}},
          {'ts': 11, 'ph': 'D', 'id': 4},
          {'ts': 12, 'ph': 'D', 'id': 3},
          {'ts': 12, 'ph': 'N', 'id': 1, 'args': {'name': 'E'}},
          {'ts': 13, 'ph': 'N', 'id': 5, 'args': {'name': 'F'}},
          {'ts': 14, 'ph': 'D', 'id': 5},
          {'ts': 15, 'ph': 'D', 'id': 1}])
  
    def testMixed(self):
      # A and E are objects, B complete, D a duration, and C and F async.
      self.CheckIntervals(self._MIXED_EVENTS)
  
    def testEventSerialization(self):
      for e in self._MIXED_EVENTS:
        event = Event(e)
        json_dict = event.ToJsonDict()
        deserialized_event = Event.FromJsonDict(json_dict)
        self.assertEquals(
            event.tracing_event, deserialized_event.tracing_event)
  
    def testTracingTrackSerialization(self):
      self._HandleEvents(self._MIXED_EVENTS)
      json_dict = self.track.ToJsonDict()
      self.assertTrue('events' in json_dict)
      deserialized_track = TracingTrack.FromJsonDict(json_dict)
      self.assertEquals(
          len(self.track._events), len(deserialized_track._events))
      for (e1, e2) in zip(self.track._events, deserialized_track._events):
        self.assertEquals(e1.tracing_event, e2.tracing_event)
  
    def testEventsEndingBetween(self):
      self._HandleEvents(self._EVENTS)
      self.assertEqual(set('ABCDEF'),
                       set([e.args['name']
                            for e in self.track.EventsEndingBetween(0, 100)]))
      self.assertFalse([e.args['name']
                        for e in self.track.EventsEndingBetween(3, 5)])
      self.assertTrue('B' in set([e.args['name']
                            for e in self.track.EventsEndingBetween(3, 6)]))
      self.assertEqual(set('B'),
                       set([e.args['name']
                            for e in self.track.EventsEndingBetween(3, 6)]))
  
    def testOverlappingEvents(self):
      self._HandleEvents(self._EVENTS)
      self.assertEqual(set('ABCDEF'),
                       set([e.args['name']
                            for e in self.track.OverlappingEvents(0, 100)]))
      self.assertFalse([e.args['name']
                        for e in self.track.OverlappingEvents(0, 2)])
      self.assertEqual(set('BA'),
                       set([e.args['name']
                            for e in self.track.OverlappingEvents(4, 5.1)]))
      self.assertEqual(set('ACD'),
                       set([e.args['name']
                            for e in self.track.OverlappingEvents(6, 10.1)]))
  
    def testEventFromStep(self):
      events = [
          {'ts': 5, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'id': '0x123',
           'name': 'B'},
          {'ts': 5, 'ph': 'X', 'dur': 2, 'pid': 2, 'tid': 1, 'id': '0x12343',
          'name': 'A'}]
      step_events = [{'ts': 6, 'ph': 'T', 'pid': 2, 'tid': 1, 'id': '0x123',
                      'name': 'B', 'args': {'step': 'Bla'}},
                     {'ts': 4, 'ph': 'T', 'pid': 2, 'tid': 1, 'id': '0x123',
                      'name': 'B', 'args': {'step': 'Bla'}},
                     {'ts': 6, 'ph': 'T', 'pid': 12, 'tid': 1, 'id': '0x123',
                      'name': 'B', 'args': {'step': 'Bla'}},
                     {'ts': 6, 'ph': 'T', 'pid': 2, 'tid': 1, 'id': '0x1234',
                      'name': 'B', 'args': {'step': 'Bla'}},
                     {'ts': 6, 'ph': 'T', 'pid': 2, 'tid': 1, 'id': '0x123',
                      'name': 'A', 'args': {'step': 'Bla'}},
                     {'ts': 6, 'ph': 'n', 'pid': 2, 'tid': 1, 'id': '0x123',
                      'name': 'B', 'args': {'step': 'Bla'}},
                     {'ts': 6, 'ph': 'n', 'pid': 2, 'tid': 1, 'id': '0x123',
                      'name': 'B', 'args': {}}]
      self._HandleEvents(events + step_events)
      trace_events = self.track.GetEvents()
      self.assertEquals(9, len(trace_events))
      # pylint: disable=unbalanced-tuple-unpacking
      (event, _, step_event, outside, wrong_pid, wrong_id, wrong_name,
       wrong_phase, no_step) = trace_events
      self.assertEquals(event, self.track.EventFromStep(step_event))
      self.assertIsNone(self.track.EventFromStep(outside))
      self.assertIsNone(self.track.EventFromStep(wrong_pid))
      self.assertIsNone(self.track.EventFromStep(wrong_id))
      self.assertIsNone(self.track.EventFromStep(wrong_name))
      # Invalid events
      with self.assertRaises(AssertionError):
        self.track.EventFromStep(wrong_phase)
      with self.assertRaises(AssertionError):
        self.track.EventFromStep(no_step)
  
    def testFilterPidTid(self):
      self._HandleEvents(self._EVENTS)
      tracing_track = self.track.Filter(2, 1)
      self.assertTrue(tracing_track is not self.track)
      self.assertEquals(4, len(tracing_track.GetEvents()))
      tracing_track = self.track.Filter(2, 42)
      self.assertEquals(0, len(tracing_track.GetEvents()))
  
    def testFilterCategories(self):
      events = [
          {'ts': 5, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'cat': 'A'},
          {'ts': 5, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'cat': 'B'},
          {'ts': 5, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'cat': 'C,D'},
          {'ts': 5, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'cat': 'A,B,C,D'}]
      self._HandleEvents(events)
      tracing_events = self.track.GetEvents()
      self.assertEquals(4, len(tracing_events))
      filtered_events = self.track.Filter(categories=None).GetEvents()
      self.assertListEqual(tracing_events, filtered_events)
      filtered_events = self.track.Filter(categories=set(['A'])).GetEvents()
      self.assertEquals(2, len(filtered_events))
      self.assertListEqual([tracing_events[0], tracing_events[3]],
                           filtered_events)
      filtered_events = self.track.Filter(categories=set(['Z'])).GetEvents()
      self.assertEquals(0, len(filtered_events))
      filtered_events = self.track.Filter(categories=set(['B', 'C'])).GetEvents()
      self.assertEquals(3, len(filtered_events))
      self.assertListEqual(tracing_events[1:], filtered_events)
  
    def _HandleEvents(self, events):
      self.track.Handle('Tracing.dataCollected', {'params': {'value': [
          self.EventToMicroseconds(e) for e in events]}})
  
  
  class IntervalTreeTestCase(unittest.TestCase):
    class FakeEvent(object):
      def __init__(self, start_msec, end_msec):
        self.start_msec = start_msec
        self.end_msec = end_msec
  
      def __eq__(self, o):
        return self.start_msec == o.start_msec and self.end_msec == o.end_msec
  
    _COUNT = 1000
  
    def testCreateTree(self):
      events = [self.FakeEvent(100 * i, 100 * (i + 1))
                for i in range(self._COUNT)]
      tree = _IntervalTree.FromEvents(events)
      self.assertEquals(0, tree.start)
      self.assertEquals(100 * self._COUNT, tree.end)
      self.assertFalse(tree._IsLeaf())
  
    def testEventsAt(self):
      events = ([self.FakeEvent(100 * i, 100 * (i + 1))
                 for i in range(self._COUNT)]
                + [self.FakeEvent(100 * i + 50, 100 * i + 150)
                   for i in range(self._COUNT)])
      tree = _IntervalTree.FromEvents(events)
      self.assertEquals(0, tree.start)
      self.assertEquals(100 * self._COUNT + 50, tree.end)
      self.assertFalse(tree._IsLeaf())
      for i in range(self._COUNT):
        self.assertEquals(2, len(tree.EventsAt(100 * i + 75)))
      # Add instant events, check that they are excluded.
      events += [self.FakeEvent(100 * i + 75, 100 * i + 75)
                 for i in range(self._COUNT)]
      tree = _IntervalTree.FromEvents(events)
      self.assertEquals(3 * self._COUNT, len(tree._events))
      for i in range(self._COUNT):
        self.assertEquals(2, len(tree.EventsAt(100 * i + 75)))
  
    def testOverlappingEvents(self):
      events = ([self.FakeEvent(100 * i, 100 * (i + 1))
                 for i in range(self._COUNT)]
                + [self.FakeEvent(100 * i + 50, 100 * i + 150)
                   for i in range(self._COUNT)])
      tree = _IntervalTree.FromEvents(events)
      self.assertEquals(0, tree.start)
      self.assertEquals(100 * self._COUNT + 50, tree.end)
      self.assertFalse(tree._IsLeaf())
      # 400 -> 500, 450 -> 550, 500 -> 600
      self.assertEquals(3, len(tree.OverlappingEvents(450, 550)))
      overlapping = sorted(
          tree.OverlappingEvents(450, 550), key=operator.attrgetter('start_msec'))
      self.assertEquals(self.FakeEvent(400, 500), overlapping[0])
      self.assertEquals(self.FakeEvent(450, 550), overlapping[1])
      self.assertEquals(self.FakeEvent(500, 600), overlapping[2])
      self.assertEquals(8, len(tree.OverlappingEvents(450, 800)))
      # Add instant events, check that they are included.
      events += [self.FakeEvent(500, 500) for i in range(10)]
      tree = _IntervalTree.FromEvents(events)
      self.assertEquals(3 + 10, len(tree.OverlappingEvents(450, 550)))
      self.assertEquals(8 + 10, len(tree.OverlappingEvents(450, 800)))
  
    def testEventMatches(self):
      event = Event({'name': 'foo',
                     'cat': 'bar',
                     'ph': 'X',
                     'ts': 0, 'dur': 0})
      self.assertTrue(event.Matches('bar', 'foo'))
      self.assertFalse(event.Matches('bar', 'biz'))
      self.assertFalse(event.Matches('biz', 'foo'))
  
      event = Event({'name': 'foo',
                     'cat': 'bar,baz,bizbiz',
                     'ph': 'X',
                     'ts': 0, 'dur': 0})
      self.assertTrue(event.Matches('bar', 'foo'))
      self.assertTrue(event.Matches('baz', 'foo'))
      self.assertFalse(event.Matches('bar', 'biz'))
      self.assertFalse(event.Matches('biz', 'foo'))
  
  if __name__ == '__main__':
    unittest.main()
  