From 9065a8598277d3dc3b0ca8897ab09d01cce3c22b Mon Sep 17 00:00:00 2001 From: Xiangquan Xiao Date: Wed, 6 Nov 2019 15:15:46 -0800 Subject: [PATCH] Cyber: Add record_py3 tests. --- cyber/python/test/test_record.py | 32 +++++------ cyber/python/test/test_record_py3.py | 85 ++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 16 deletions(-) mode change 100644 => 100755 cyber/python/test/test_record.py create mode 100755 cyber/python/test/test_record_py3.py diff --git a/cyber/python/test/test_record.py b/cyber/python/test/test_record.py old mode 100644 new mode 100755 index 49392cdfe7a..f80e3591ab3 --- a/cyber/python/test/test_record.py +++ b/cyber/python/test/test_record.py @@ -1,6 +1,8 @@ +#!/usr/bin/env python2 + # **************************************************************************** # Copyright 2018 The Apollo Authors. All Rights Reserved. - +# # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -20,12 +22,12 @@ import unittest sys.path.append("../") -from cyber_py import cyber +from cyber.proto import record_pb2 from cyber_py import record from modules.common.util.testdata.simple_pb2 import SimpleMessage -from cyber.proto import record_pb2 -TEST_RECORD_FILE = "test02.record" + +TEST_RECORD_FILE = "/tmp/test02.record" CHAN_1 = "channel/chatter" MSG_TYPE = "apollo.common.util.test.SimpleMessage" STR_10B = "1234567890" @@ -42,8 +44,6 @@ def test_record_writer_read(self): """ unit test of record. """ - self.assertTrue(cyber.init()) - # writer fwriter = record.RecordWriter() fwriter.set_size_fileseg(0) @@ -63,6 +63,15 @@ def test_record_writer_read(self): channel_list = fread.get_channellist() self.assertEqual(1, len(channel_list)) self.assertEqual(CHAN_1, channel_list[0]) + + header = record_pb2.Header() + header.ParseFromString(fread.get_headerstring()) + self.assertEqual(1, header.major_version) + self.assertEqual(0, header.minor_version) + self.assertEqual(1, header.chunk_number) + self.assertEqual(1, header.channel_number) + self.assertTrue(header.is_complete) + for channelname, msg, datatype, timestamp in fread.read_messages(): self.assertEqual(CHAN_1, channelname) self.assertEqual(STR_10B, msg) @@ -70,16 +79,7 @@ def test_record_writer_read(self): self.assertEqual(1, fread.get_messagenumber(channelname)) self.assertEqual(MSG_TYPE, datatype) self.assertEqual(MSG_TYPE, fread.get_messagetype(channelname)) - msg = record_pb2.Header() - header_msg = fread.get_headerstring() - msg.ParseFromString(header_msg) - self.assertEqual(1, msg.major_version) - self.assertEqual(0, msg.minor_version) - self.assertEqual(1, msg.chunk_number) - self.assertEqual(1, msg.channel_number) - self.assertTrue(msg.is_complete) - - cyber.shutdown() + if __name__ == '__main__': unittest.main() diff --git a/cyber/python/test/test_record_py3.py b/cyber/python/test/test_record_py3.py new file mode 100755 index 00000000000..29540cca88d --- /dev/null +++ b/cyber/python/test/test_record_py3.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 + +# **************************************************************************** +# Copyright 2018 The Apollo Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# **************************************************************************** +# -*- coding: utf-8 -*- +"""Module for test record.""" + +import sys +import unittest + +sys.path.append("../") +from cyber.proto import record_pb2 +from cyber_py import record_py3 as record +from modules.common.util.testdata.simple_pb2 import SimpleMessage + + +TEST_RECORD_FILE = "/tmp/test02.record" +CHAN_1 = "channel/chatter" +MSG_TYPE = "apollo.common.util.test.SimpleMessage" +STR_10B = b"1234567890" +TIME = 999 + + +class TestRecord(unittest.TestCase): + + """ + Class for record unit test. + """ + + def test_record_writer_read(self): + """ + unit test of record. + """ + # writer + fwriter = record.RecordWriter() + fwriter.set_size_fileseg(0) + fwriter.set_intervaltime_fileseg(0) + + self.assertTrue(fwriter.open(TEST_RECORD_FILE)) + fwriter.write_channel(CHAN_1, MSG_TYPE, STR_10B) + fwriter.write_message(CHAN_1, STR_10B, TIME) + + self.assertEqual(1, fwriter.get_messagenumber(CHAN_1)) + self.assertEqual(MSG_TYPE, fwriter.get_messagetype(CHAN_1)) + self.assertEqual(STR_10B, fwriter.get_protodesc(CHAN_1)) + fwriter.close() + + # reader + fread = record.RecordReader(TEST_RECORD_FILE) + channel_list = fread.get_channellist() + self.assertEqual(1, len(channel_list)) + self.assertEqual(CHAN_1, channel_list[0]) + + header = record_pb2.Header() + header.ParseFromString(fread.get_headerstring()) + self.assertEqual(1, header.major_version) + self.assertEqual(0, header.minor_version) + self.assertEqual(1, header.chunk_number) + self.assertEqual(1, header.channel_number) + self.assertTrue(header.is_complete) + + for channelname, msg, datatype, timestamp in fread.read_messages(): + self.assertEqual(CHAN_1, channelname) + self.assertEqual(STR_10B, msg) + self.assertEqual(TIME, timestamp) + self.assertEqual(1, fread.get_messagenumber(channelname)) + self.assertEqual(MSG_TYPE, datatype) + self.assertEqual(MSG_TYPE, fread.get_messagetype(channelname)) + + +if __name__ == '__main__': + unittest.main()