Cyber: Add record_py3 tests.

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,23 +63,23 @@ 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)
             self.assertEqual(TIME, timestamp)
             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()

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()