tests/tests-python.py

   1 #!/usr/bin/env python3
   2 import unittest
   3 import sys
   4 from babeltrace import *
   5
   6 class TestBabeltracePythonModule(unittest.TestCase):
   7
   8         def test_handle_decl(self):
   9                 #Context creation, adding trace
  10                 ctx = Context()
  11                 trace_handle = ctx.add_trace(
  12                         "ctf-traces/succeed/lttng-modules-2.0-pre5",
  13                         "ctf")
  14                 self.assertIsNotNone(trace_handle, "Error adding trace")
  15
  16                 #TraceHandle test
  17                 ts_begin = trace_handle.get_timestamp_begin(ctx, CLOCK_REAL)
  18                 ts_end = trace_handle.get_timestamp_end(ctx, CLOCK_REAL)
  19                 self.assertGreater(ts_end, ts_begin, "Error get_timestamp from trace_handle")
  20
  21                 lst = ctf.get_event_decl_list(trace_handle, ctx)
  22                 self.assertIsNotNone(lst, "Error get_event_decl_list")
  23
  24                 name = lst[0].get_name()
  25                 self.assertIsNotNone(name)
  26
  27                 fields = lst[0].get_decl_fields(ctf.scope.EVENT_FIELDS)
  28                 self.assertIsNotNone(fields, "Error getting FieldDecl list")
  29
  30                 if len(fields) > 0:
  31                         self.assertIsNotNone(fields[0].get_name(),
  32                                 "Error getting name from FieldDecl")
  33
  34                 #Remove trace
  35                 ctx.remove_trace(trace_handle)
  36                 del ctx
  37                 del trace_handle
  38
  39
  40         def test_iterator_event(self):
  41                 #Context creation, adding trace
  42                 ctx = Context()
  43                 trace_handle = ctx.add_trace(
  44                         "ctf-traces/succeed/lttng-modules-2.0-pre5",
  45                         "ctf")
  46                 self.assertIsNotNone(trace_handle, "Error adding trace")
  47
  48                 begin_pos = IterPos(SEEK_BEGIN)
  49                 it = ctf.Iterator(ctx, begin_pos)
  50                 self.assertIsNotNone(it, "Error creating iterator")
  51
  52                 event = it.read_event()
  53                 self.assertIsNotNone(event, "Error reading event")
  54
  55                 handle = event.get_handle()
  56                 self.assertIsNotNone(handle, "Error getting handle")
  57
  58                 context = event.get_context()
  59                 self.assertIsNotNone(context, "Error getting context")
  60
  61                 name = ""
  62                 while event is not None and name != "sched_switch":
  63                         name = event.get_name()
  64                         self.assertIsNotNone(name, "Error getting event name")
  65
  66                         ts = event.get_timestamp()
  67                         self.assertGreaterEqual(ts, 0, "Error getting timestamp")
  68
  69                         cycles = event.get_cycles()
  70                         self.assertGreaterEqual(cycles, 0, "Error getting cycles")
  71
  72                         if name == "sched_switch":
  73                                 scope = event.get_top_level_scope(ctf.scope.STREAM_PACKET_CONTEXT)
  74                                 self.assertIsNotNone(scope, "Error getting scope definition")
  75
  76                                 field = event.get_field(scope, "cpu_id")
  77                                 prev_comm_str = field.get_uint64()
  78                                 self.assertEqual(ctf.field_error(), 0, "Error getting vec info")
  79
  80                                 field_lst = event.get_field_list(scope)
  81                                 self.assertIsNotNone(field_lst, "Error getting field list")
  82
  83                                 fname = field.field_name()
  84                                 self.assertIsNotNone(fname, "Error getting field name")
  85
  86                                 ftype = field.field_type()
  87                                 self.assertIsNot(ftype, -1, "Error getting field type (unknown)")
  88
  89                         ret = it.next()
  90                         self.assertGreaterEqual(ret, 0, "Error moving iterator")
  91
  92                         event = it.read_event()
  93
  94                 pos = it.get_pos()
  95                 self.assertIsNotNone(pos._pos, "Error getting iterator position")
  96
  97                 ret = it.set_pos(pos)
  98                 self.assertEqual(ret, 0, "Error setting iterator position")
  99
 100                 pos = it.create_time_pos(ts)
 101                 self.assertIsNotNone(pos._pos, "Error creating time-based iterator position")
 102
 103                 del it, pos
 104                 set_pos = IterPos(SEEK_TIME, ts)
 105                 it = ctf.Iterator(ctx, begin_pos, set_pos)
 106                 self.assertIsNotNone(it, "Error creating iterator with end_pos")
 107
 108
 109         def test_file_class(self):
 110                 f = File("test-bitfield.c")
 111                 self.assertIsNotNone(f, "Error opening file")
 112                 f.close()
 113
 114
 115 if __name__ == "__main__":
 116         unittest.main()