tracer/tracer/parser_test.py

   1 import sys
   2 import unittest
   3
   4 from tracer.parser import Parser
   5
   6 TESTFILE = './logs/trace.latency.log'
   7
   8 TEXT_DATA = ['first line', 'second line']
   9
  10 TRACER_LOG_DATA = [
  11     '0:00:00.079422574  7664      0x238ac70 TRACE             GST_TRACER :0:: thread-rusage, thread-id=(guint64)37268592, ts=(guint64)79416000, average-cpuload=(uint)1000, current-cpuload=(uint)1000, time=(guint64)79418045;'
  12 ]
  13 TRACER_CLASS_LOG_DATA = [
  14     '0:00:00.041536066  1788      0x14b2150 TRACE             GST_TRACER gsttracerrecord.c:110:gst_tracer_record_build_format: latency.class, src=(structure)"scope\,\ type\=\(type\)gchararray\,\ related-to\=\(GstTracerValueScope\)GST_TRACER_VALUE_SCOPE_PAD\;", sink=(structure)"scope\,\ type\=\(type\)gchararray\,\ related-to\=\(GstTracerValueScope\)GST_TRACER_VALUE_SCOPE_PAD\;", time=(structure)"value\,\ type\=\(type\)guint64\,\ description\=\(string\)\"time\\\ it\\\ took\\\ for\\\ the\\\ buffer\\\ to\\\ go\\\ from\\\ src\\\ to\\\ sink\\\ ns\"\,\ flags\=\(GstTracerValueFlags\)GST_TRACER_VALUE_FLAGS_AGGREGATED\,\ min\=\(guint64\)0\,\ max\=\(guint64\)18446744073709551615\;";'
  15 ]
  16
  17
  18 class TestParser(unittest.TestCase):
  19
  20     def test___init__(self):
  21         log = Parser(TESTFILE)
  22         self.assertIsNone(log.file)
  23
  24     def test___enter___with_file(self):
  25         with Parser(TESTFILE) as log:
  26             self.assertIsNotNone(log.file)
  27
  28     def test___enter___with_stdin(self):
  29         sys.stdin = iter(TEXT_DATA)
  30         with Parser('-') as log:
  31             self.assertIsNotNone(log.file)
  32
  33     def test_random_text_reports_none(self):
  34         sys.stdin = iter(TEXT_DATA)
  35         with Parser('-') as log:
  36             with self.assertRaises(StopIteration):
  37                 next(log)
  38
  39     def test_log_file_reports_trace_log(self):
  40         with Parser(TESTFILE) as log:
  41             self.assertIsNotNone(next(log))
  42
  43     def test_trace_log_parsed(self):
  44         sys.stdin = iter(TRACER_LOG_DATA)
  45         with Parser('-') as log:
  46             event = next(log)
  47             self.assertEqual(len(event), 10)