text: #!/usr/bin/env python
  # Copyright 2014 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 contextlib
  import json
  import logging
  import math
  import multiprocessing
  import os
  import time
  import unittest
  import sys
  
  from py_trace_event import trace_event
  from py_trace_event import trace_time
  from py_trace_event.trace_event_impl import log
  from py_trace_event.trace_event_impl import multiprocessing_shim
  from py_utils import tempfile_ext
  
  
  class TraceEventTests(unittest.TestCase):
  
    @contextlib.contextmanager
    def _test_trace(self, disable=True, format=None):
      with tempfile_ext.TemporaryFileName() as filename:
        self._log_path = filename
        try:
          trace_event.trace_enable(self._log_path, format=format)
          yield
        finally:
          if disable:
            trace_event.trace_disable()
  
    def testNoImpl(self):
      orig_impl = trace_event.trace_event_impl
      try:
        trace_event.trace_event_impl = None
        self.assertFalse(trace_event.trace_can_enable())
      finally:
        trace_event.trace_event_impl = orig_impl
  
    def testImpl(self):
      self.assertTrue(trace_event.trace_can_enable())
  
    def testIsEnabledFalse(self):
      self.assertFalse(trace_event.trace_is_enabled())
  
    def testIsEnabledTrue(self):
      with self._test_trace():
        self.assertTrue(trace_event.trace_is_enabled())
  
    def testEnable(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 1)
          self.assertTrue(trace_event.trace_is_enabled())
          log_output = log_output.pop()
          self.assertEquals(log_output['category'], 'process_argv')
          self.assertEquals(log_output['name'], 'process_argv')
          self.assertTrue(log_output['args']['argv'])
          self.assertEquals(log_output['ph'], 'M')
  
    def testDoubleEnable(self):
      try:
        with self._test_trace():
          with self._test_trace():
            pass
      except log.TraceException:
        return
      assert False
  
    def testDisable(self):
      _old_multiprocessing_process = multiprocessing.Process
      with self._test_trace(disable=False):
        with open(self._log_path, 'r') as f:
          self.assertTrue(trace_event.trace_is_enabled())
          self.assertEqual(
              multiprocessing.Process, multiprocessing_shim.ProcessShim)
          trace_event.trace_disable()
          self.assertEqual(
              multiprocessing.Process, _old_multiprocessing_process)
          self.assertEquals(len(json.loads(f.read() + ']')), 1)
          self.assertFalse(trace_event.trace_is_enabled())
  
    def testDoubleDisable(self):
      with self._test_trace():
        pass
      trace_event.trace_disable()
  
    def testFlushChanges(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          trace_event.clock_sync('1')
          self.assertEquals(len(json.loads(f.read() + ']')), 1)
          f.seek(0)
          trace_event.trace_flush()
          self.assertEquals(len(json.loads(f.read() + ']')), 2)
  
    def testFlushNoChanges(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          self.assertEquals(len(json.loads(f.read() + ']')),1)
          f.seek(0)
          trace_event.trace_flush()
          self.assertEquals(len(json.loads(f.read() + ']')), 1)
  
    def testDoubleFlush(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          trace_event.clock_sync('1')
          self.assertEquals(len(json.loads(f.read() + ']')), 1)
          f.seek(0)
          trace_event.trace_flush()
          trace_event.trace_flush()
          self.assertEquals(len(json.loads(f.read() + ']')), 2)
  
    def testTraceBegin(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          trace_event.trace_begin('test_event', this='that')
          trace_event.trace_flush()
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 2)
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'process_argv')
          self.assertEquals(current_entry['name'], 'process_argv')
          self.assertTrue( current_entry['args']['argv'])
          self.assertEquals( current_entry['ph'], 'M')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], 'test_event')
          self.assertEquals(current_entry['args']['this'], '\'that\'')
          self.assertEquals(current_entry['ph'], 'B')
  
    def testTraceEnd(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          trace_event.trace_end('test_event')
          trace_event.trace_flush()
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 2)
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'process_argv')
          self.assertEquals(current_entry['name'], 'process_argv')
          self.assertTrue(current_entry['args']['argv'])
          self.assertEquals(current_entry['ph'], 'M')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], 'test_event')
          self.assertEquals(current_entry['args'], {})
          self.assertEquals(current_entry['ph'], 'E')
  
    def testTrace(self):
     with self._test_trace():
        with trace_event.trace('test_event', this='that'):
          pass
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 3)
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'process_argv')
          self.assertEquals(current_entry['name'], 'process_argv')
          self.assertTrue(current_entry['args']['argv'])
          self.assertEquals(current_entry['ph'], 'M')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], 'test_event')
          self.assertEquals(current_entry['args']['this'], '\'that\'')
          self.assertEquals(current_entry['ph'], 'B')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], 'test_event')
          self.assertEquals(current_entry['args'], {})
          self.assertEquals(current_entry['ph'], 'E')
  
    def testTracedDecorator(self):
      @trace_event.traced("this")
      def test_decorator(this="that"):
        pass
  
      with self._test_trace():
        test_decorator()
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 3)
          expected_name = __name__ + '.test_decorator'
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'process_argv')
          self.assertEquals(current_entry['name'], 'process_argv')
          self.assertTrue(current_entry['args']['argv'])
          self.assertEquals(current_entry['ph'], 'M')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], expected_name)
          self.assertEquals(current_entry['args']['this'], '\'that\'')
          self.assertEquals(current_entry['ph'], 'B')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], expected_name)
          self.assertEquals(current_entry['args'], {})
          self.assertEquals(current_entry['ph'], 'E')
  
    def testClockSyncWithTs(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          trace_event.clock_sync('id', issue_ts=trace_time.Now())
          trace_event.trace_flush()
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 2)
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'process_argv')
          self.assertEquals(current_entry['name'], 'process_argv')
          self.assertTrue(current_entry['args']['argv'])
          self.assertEquals(current_entry['ph'], 'M')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], 'clock_sync')
          self.assertTrue(current_entry['args']['issue_ts'])
          self.assertEquals(current_entry['ph'], 'c')
  
    def testClockSyncWithoutTs(self):
      with self._test_trace():
        with open(self._log_path, 'r') as f:
          trace_event.clock_sync('id')
          trace_event.trace_flush()
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 2)
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'process_argv')
          self.assertEquals(current_entry['name'], 'process_argv')
          self.assertTrue(current_entry['args']['argv'])
          self.assertEquals(current_entry['ph'], 'M')
          current_entry = log_output.pop(0)
          self.assertEquals(current_entry['category'], 'python')
          self.assertEquals(current_entry['name'], 'clock_sync')
          self.assertFalse(current_entry['args'].get('issue_ts'))
          self.assertEquals(current_entry['ph'], 'c')
  
    def testTime(self):
      actual_diff = []
      def func1():
        trace_begin("func1")
        start = time.time()
        time.sleep(0.25)
        end = time.time()
        actual_diff.append(end-start) # Pass via array because of Python scoping
        trace_end("func1")
  
      with self._test_trace():
        start_ts = time.time()
        trace_event.trace_begin('test')
        end_ts = time.time()
        trace_event.trace_end('test')
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 3)
          meta_data = log_output[0]
          open_data = log_output[1]
          close_data = log_output[2]
          self.assertEquals(meta_data['category'], 'process_argv')
          self.assertEquals(meta_data['name'], 'process_argv')
          self.assertTrue(meta_data['args']['argv'])
          self.assertEquals(meta_data['ph'], 'M')
          self.assertEquals(open_data['category'], 'python')
          self.assertEquals(open_data['name'], 'test')
          self.assertEquals(open_data['ph'], 'B')
          self.assertEquals(close_data['category'], 'python')
          self.assertEquals(close_data['name'], 'test')
          self.assertEquals(close_data['ph'], 'E')
          event_time_diff = close_data['ts'] - open_data['ts']
          recorded_time_diff = (end_ts - start_ts) * 1000000
          self.assertLess(math.fabs(event_time_diff - recorded_time_diff), 1000)
  
    def testNestedCalls(self):
      with self._test_trace():
        trace_event.trace_begin('one')
        trace_event.trace_begin('two')
        trace_event.trace_end('two')
        trace_event.trace_end('one')
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 5)
          meta_data = log_output[0]
          one_open = log_output[1]
          two_open = log_output[2]
          two_close = log_output[3]
          one_close = log_output[4]
          self.assertEquals(meta_data['category'], 'process_argv')
          self.assertEquals(meta_data['name'], 'process_argv')
          self.assertTrue(meta_data['args']['argv'])
          self.assertEquals(meta_data['ph'], 'M')
  
          self.assertEquals(one_open['category'], 'python')
          self.assertEquals(one_open['name'], 'one')
          self.assertEquals(one_open['ph'], 'B')
          self.assertEquals(one_close['category'], 'python')
          self.assertEquals(one_close['name'], 'one')
          self.assertEquals(one_close['ph'], 'E')
  
          self.assertEquals(two_open['category'], 'python')
          self.assertEquals(two_open['name'], 'two')
          self.assertEquals(two_open['ph'], 'B')
          self.assertEquals(two_close['category'], 'python')
          self.assertEquals(two_close['name'], 'two')
          self.assertEquals(two_close['ph'], 'E')
  
          self.assertLessEqual(one_open['ts'], two_open['ts'])
          self.assertGreaterEqual(one_close['ts'], two_close['ts'])
  
    def testInterleavedCalls(self):
      with self._test_trace():
        trace_event.trace_begin('one')
        trace_event.trace_begin('two')
        trace_event.trace_end('one')
        trace_event.trace_end('two')
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 5)
          meta_data = log_output[0]
          one_open = log_output[1]
          two_open = log_output[2]
          two_close = log_output[4]
          one_close = log_output[3]
          self.assertEquals(meta_data['category'], 'process_argv')
          self.assertEquals(meta_data['name'], 'process_argv')
          self.assertTrue(meta_data['args']['argv'])
          self.assertEquals(meta_data['ph'], 'M')
  
          self.assertEquals(one_open['category'], 'python')
          self.assertEquals(one_open['name'], 'one')
          self.assertEquals(one_open['ph'], 'B')
          self.assertEquals(one_close['category'], 'python')
          self.assertEquals(one_close['name'], 'one')
          self.assertEquals(one_close['ph'], 'E')
  
          self.assertEquals(two_open['category'], 'python')
          self.assertEquals(two_open['name'], 'two')
          self.assertEquals(two_open['ph'], 'B')
          self.assertEquals(two_close['category'], 'python')
          self.assertEquals(two_close['name'], 'two')
          self.assertEquals(two_close['ph'], 'E')
  
          self.assertLessEqual(one_open['ts'], two_open['ts'])
          self.assertLessEqual(one_close['ts'], two_close['ts'])
  
    # TODO(khokhlov): Fix this test on Windows. See crbug.com/945819 for details.
    def disabled_testMultiprocess(self):
      def child_function():
        with trace_event.trace('child_event'):
          pass
  
      with self._test_trace():
        trace_event.trace_begin('parent_event')
        trace_event.trace_flush()
        p = multiprocessing.Process(target=child_function)
        p.start()
        self.assertTrue(hasattr(p, "_shimmed_by_trace_event"))
        p.join()
        trace_event.trace_end('parent_event')
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 5)
          meta_data = log_output[0]
          parent_open = log_output[1]
          child_open = log_output[2]
          child_close = log_output[3]
          parent_close = log_output[4]
          self.assertEquals(meta_data['category'], 'process_argv')
          self.assertEquals(meta_data['name'], 'process_argv')
          self.assertTrue(meta_data['args']['argv'])
          self.assertEquals(meta_data['ph'], 'M')
  
          self.assertEquals(parent_open['category'], 'python')
          self.assertEquals(parent_open['name'], 'parent_event')
          self.assertEquals(parent_open['ph'], 'B')
  
          self.assertEquals(child_open['category'], 'python')
          self.assertEquals(child_open['name'], 'child_event')
          self.assertEquals(child_open['ph'], 'B')
  
          self.assertEquals(child_close['category'], 'python')
          self.assertEquals(child_close['name'], 'child_event')
          self.assertEquals(child_close['ph'], 'E')
  
          self.assertEquals(parent_close['category'], 'python')
          self.assertEquals(parent_close['name'], 'parent_event')
          self.assertEquals(parent_close['ph'], 'E')
  
    @unittest.skipIf(sys.platform == 'win32', 'crbug.com/945819')
    def testTracingControlDisabledInChildButNotInParent(self):
      def child(resp):
        # test tracing is not controllable in the child
        resp.put(trace_event.is_tracing_controllable())
  
      with self._test_trace():
        q = multiprocessing.Queue()
        p = multiprocessing.Process(target=child, args=[q])
        p.start()
        # test tracing is controllable in the parent
        self.assertTrue(trace_event.is_tracing_controllable())
        self.assertFalse(q.get())
        p.join()
  
    def testMultiprocessExceptionInChild(self):
      def bad_child():
        trace_event.trace_disable()
  
      with self._test_trace():
        p = multiprocessing.Pool(1)
        trace_event.trace_begin('parent')
        self.assertRaises(Exception, lambda: p.apply(bad_child, ()))
        p.close()
        p.terminate()
        p.join()
        trace_event.trace_end('parent')
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
          self.assertEquals(len(log_output), 3)
          meta_data = log_output[0]
          parent_open = log_output[1]
          parent_close = log_output[2]
          self.assertEquals(parent_open['category'], 'python')
          self.assertEquals(parent_open['name'], 'parent')
          self.assertEquals(parent_open['ph'], 'B')
          self.assertEquals(parent_close['category'], 'python')
          self.assertEquals(parent_close['name'], 'parent')
          self.assertEquals(parent_close['ph'], 'E')
  
    def testFormatJson(self):
      with self._test_trace(format=trace_event.JSON):
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          log_output = json.loads(f.read() + ']')
      self.assertEquals(len(log_output), 1)
      self.assertEquals(log_output[0]['ph'], 'M')
  
    def testFormatJsonWithMetadata(self):
      with self._test_trace(format=trace_event.JSON_WITH_METADATA):
        trace_event.trace_disable()
        with open(self._log_path, 'r') as f:
          log_output = json.load(f)
      self.assertEquals(len(log_output), 2)
      events = log_output['traceEvents']
      self.assertEquals(len(events), 1)
      self.assertEquals(events[0]['ph'], 'M')
  
    def testFormatProtobuf(self):
      with self._test_trace(format=trace_event.PROTOBUF):
        trace_event.trace_flush()
        with open(self._log_path, 'r') as f:
          self.assertGreater(len(f.read()), 0)
  
    def testAddMetadata(self):
      with self._test_trace(format=trace_event.JSON_WITH_METADATA):
        trace_event.trace_add_benchmark_metadata(
            benchmark_start_time_us=1000,
            story_run_time_us=2000,
            benchmark_name='benchmark',
            benchmark_description='desc',
            story_name='story',
            story_tags=['tag1', 'tag2'],
            story_run_index=0,
        )
        trace_event.trace_disable()
        with open(self._log_path, 'r') as f:
          log_output = json.load(f)
      self.assertEquals(len(log_output), 2)
      telemetry_metadata = log_output['metadata']['telemetry']
      self.assertEquals(len(telemetry_metadata), 7)
      self.assertEquals(telemetry_metadata['benchmarkStart'], 1)
      self.assertEquals(telemetry_metadata['traceStart'], 2)
      self.assertEquals(telemetry_metadata['benchmarks'], ['benchmark'])
      self.assertEquals(telemetry_metadata['benchmarkDescriptions'], ['desc'])
      self.assertEquals(telemetry_metadata['stories'], ['story'])
      self.assertEquals(telemetry_metadata['storyTags'], ['tag1', 'tag2'])
      self.assertEquals(telemetry_metadata['storysetRepeats'], [0])
  
    def testAddMetadataProtobuf(self):
      with self._test_trace(format=trace_event.PROTOBUF):
        trace_event.trace_add_benchmark_metadata(
            benchmark_start_time_us=1000,
            story_run_time_us=2000,
            benchmark_name='benchmark',
            benchmark_description='desc',
            story_name='story',
            story_tags=['tag1', 'tag2'],
            story_run_index=0,
        )
        trace_event.trace_disable()
        with open(self._log_path, 'r') as f:
          self.assertGreater(len(f.read()), 0)
  
    def testAddMetadataInJsonFormatRaises(self):
      with self._test_trace(format=trace_event.JSON):
        with self.assertRaises(log.TraceException):
          trace_event.trace_add_benchmark_metadata(
              benchmark_start_time_us=1000,
              story_run_time_us=2000,
              benchmark_name='benchmark',
              benchmark_description='description',
              story_name='story',
              story_tags=['tag1', 'tag2'],
              story_run_index=0,
          )
  
  
  if __name__ == '__main__':
    logging.getLogger().setLevel(logging.DEBUG)
    unittest.main(verbosity=2)
  