Rename *create_iterator*() -> *create_notification_iterator*()
[babeltrace.git] / tests / lib / writer / test_ctf_writer_empty_packet.py
CommitLineData
4c3453a1
JG
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
25import bt_python_helper
26import tempfile
27import babeltrace.writer as btw
28import babeltrace.reader as btr
29import shutil
30import uuid
31
32TEST_COUNT = 4
33EXPECTED_EVENT_COUNT = 10
34trace_path = tempfile.mkdtemp()
35
36
37def 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
42def 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('test_stream')
53 stream_class.add_event_class(event_class)
54 stream_class.clock = clock
55
56 stream_class.packet_context_type = None
57 print_test_result(1, stream_class.packet_context_type is None,
58 'Set stream class packet context type to None')
59
60 stream = trace.create_stream(stream_class)
61
62 for i in range(EXPECTED_EVENT_COUNT):
63 event = btw.Event(event_class)
64 event.payload('int_field').value = i
65 stream.append_event(event)
66 stream.flush()
67 print_test_result(2, True,
68 'Flush a packet')
69
70 # It is not valid for a stream to contain more than one packet
71 # if it does not have content_size/packet_size info in its packet
72 # context
73 event = btw.Event(event_class)
74 event.payload('int_field').value = 42
75 stream.append_event(event)
76 try:
77 stream.flush()
78 second_flush_succeeded = True
79 except ValueError:
80 second_flush_succeeded = False
81
82 print_test_result(3, not second_flush_succeeded,
83 'Flushing a second packet in a stream without a defined packet context fails')
84
85
86# TAP plan
87print('1..{}'.format(TEST_COUNT))
88print('# Creating trace at {}'.format(trace_path))
89create_trace(trace_path)
90
91traces = btr.TraceCollection()
92trace_handle = traces.add_trace(trace_path, 'ctf')
93if trace_handle is None:
94 print('Failed to open trace at {}'.format(trace_path))
95
96event_count = 0
97for event in traces.events:
98 event_count += 1
99
100print_test_result(4, EXPECTED_EVENT_COUNT == event_count,
101 'Trace was found to contain {} events, expected {}'.format(
102 event_count, EXPECTED_EVENT_COUNT))
65c53793
JG
103
104shutil.rmtree(trace_path)
This page took 0.027346 seconds and 4 git commands to generate.