# 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, ['A', 'B', 'C', 'D'])

  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 testGetMainFrameID(self):
    _MAIN_FRAME_ID = 0xffff
    _SUBFRAME_ID = 0xaaaa
    events = [
        {'ts': 7, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'id': '0x123',
         'name': 'navigationStart', 'cat': 'blink.user_timing',
         'args': {'frame': _SUBFRAME_ID}},
        {'ts': 8, 'ph': 'X', 'dur': 2, 'pid': 2, 'tid': 1, 'id': '0x12343',
        'name': 'A'},
        {'ts': 3, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'id': '0x125',
         'name': 'navigationStart', 'cat': 'blink.user_timing',
         'args': {'frame': _MAIN_FRAME_ID}},
        ]
    self._HandleEvents(events)
    self.assertEquals(_MAIN_FRAME_ID, self.track.GetMainFrameID())

  def testGetMatchingEvents(self):
    _MAIN_FRAME_ID = 0xffff
    _SUBFRAME_ID = 0xaaaa
    events = [
        {'ts': 7, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'id': '0x123',
         'name': 'navigationStart', 'cat': 'blink.user_timing',
         'args': {'frame': _SUBFRAME_ID}},
        {'ts': 8, 'ph': 'X', 'dur': 2, 'pid': 2, 'tid': 1, 'id': '0x12343',
        'name': 'A'},
        {'ts': 3, 'ph': 'X', 'dur': 10, 'pid': 2, 'tid': 1, 'id': '0x125',
         'name': 'navigationStart', 'cat': 'blink.user_timing',
         'args': {'frame': _MAIN_FRAME_ID}},
        ]
    self._HandleEvents(events)
    matching_events = self.track.GetMatchingEvents('blink.user_timing',
                                                   'navigationStart')
    self.assertEquals(2, len(matching_events))
    self.assertListEqual([self.track.GetEvents()[0],
                         self.track.GetEvents()[2]], matching_events)

    matching_main_frame_events = self.track.GetMatchingMainFrameEvents(
        'blink.user_timing', 'navigationStart')
    self.assertEquals(1, len(matching_main_frame_events))
    self.assertListEqual([self.track.GetEvents()[2]],
                         matching_main_frame_events)

  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)
    self.assertSetEqual(
        set('A'), self.track.Filter(categories=set('A')).Categories())

  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()