tests/lib/writer/test_ctf_writer_empty_packet.py.in

   1 #!/usr/bin/env python3
   2 #
   3 # The MIT License (MIT)
   4 #
   5 # Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   6 #
   7 # Permission is hereby granted, free of charge, to any person obtaining a copy
   8 # of this software and associated documentation files (the "Software"), to deal
   9 # in the Software without restriction, including without limitation the rights
  10 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11 # copies of the Software, and to permit persons to whom the Software is
  12 # furnished to do so, subject to the following conditions:
  13 #
  14 # The above copyright notice and this permission notice shall be included in
  15 # all copies or substantial portions of the Software.
  16 #
  17 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  23 # SOFTWARE.
  24
  25 import bt_python_helper
  26 import tempfile
  27 import babeltrace.writer as btw
  28 import babeltrace.reader as btr
  29 import shutil
  30 import uuid
  31
  32 TEST_COUNT = 4
  33 EXPECTED_EVENT_COUNT = 10
  34 trace_path = tempfile.mkdtemp()
  35
  36
  37 def print_test_result(test_number, result, description):
  38         result_string = 'ok' if result else 'not ok'
  39         result_string += ' {} - {}'.format(test_number, description)
  40         print(result_string)
  41
  42 def create_trace(path):
  43         trace = btw.Writer(path)
  44         clock = btw.Clock('test_clock')
  45         trace.add_clock(clock)
  46
  47         integer_field_type = btw.IntegerFieldDeclaration(32)
  48
  49         event_class = btw.EventClass('simple_event')
  50         event_class.add_field(integer_field_type, 'int_field')
  51
  52         stream_class = btw.StreamClass('empty_packet_stream')
  53         stream_class.add_event_class(event_class)
  54         stream_class.clock = clock
  55
  56         stream = trace.create_stream(stream_class)
  57
  58         for i in range(EXPECTED_EVENT_COUNT):
  59                 event = btw.Event(event_class)
  60                 event.payload('int_field').value = i
  61                 stream.append_event(event)
  62         stream.flush()
  63         print_test_result(1, True,
  64                           'Flush a packet containing {} events'.format(EXPECTED_EVENT_COUNT))
  65
  66         # The CTF writer will not be able to populate the packet context's
  67         # timestamp_begin and timestamp_end fields if it is asked to flush
  68         # without any queued events.
  69         try:
  70             stream.flush()
  71             empty_flush_succeeded = True
  72         except ValueError:
  73             empty_flush_succeeded = False
  74
  75         print_test_result(2, not empty_flush_succeeded,
  76                           'Flushing an empty packet without explicitly setting packet time bounds should fail')
  77         packet_context = stream.packet_context
  78         packet_context.field('timestamp_begin').value = 1
  79         packet_context.field('timestamp_end').value = 123456
  80         try:
  81             stream.flush()
  82             empty_flush_succeeded = True
  83         except ValueError:
  84             empty_flush_succeeded = False
  85
  86         print_test_result(3, empty_flush_succeeded,
  87                           'Flushing an empty packet after explicitly setting packet time bounds should succeed')
  88
  89
  90 # TAP plan
  91 print('1..{}'.format(TEST_COUNT))
  92 print('# Creating trace at {}'.format(trace_path))
  93 create_trace(trace_path)
  94
  95 traces = btr.TraceCollection()
  96 trace_handle = traces.add_trace(trace_path, 'ctf')
  97 if trace_handle is None:
  98         print('Failed to open trace at {}'.format(trace_path))
  99
 100 event_count = 0
 101 for event in traces.events:
 102         event_count += 1
 103
 104 print_test_result(4, EXPECTED_EVENT_COUNT == event_count,
 105                   'Trace was found to contain {} events, expected {}'.format(
 106                    event_count, EXPECTED_EVENT_COUNT))
 107
 108 shutil.rmtree(trace_path)