Add tests for Parser

python/test/test_file_parser.py

@@ -33,18 +33,17 @@
 from novatel_edie import STATUS, ENCODE_FORMAT
 
 
-# -------------------------------------------------------------------------------------------------------
-# Unit Tests
-# -------------------------------------------------------------------------------------------------------
-
 @pytest.fixture(scope="function")
 def fp():
     return ne.FileParser()
 
 
+@pytest.fixture(scope="module")
+def test_gps_file(decoders_test_resources):
+    return decoders_test_resources / "BESTUTMBIN.GPS"
+
+
 def test_logger():
-    assert ne.Logging.get("novatel_file_parser") is None
-    assert ne.Logging.get("novatel_parser") is None
     # FileParser logger
     level = ne.LogLevel.OFF
     file_parser = ne.FileParser()
@@ -78,51 +77,40 @@ def test_unknown_bytes(fp):
     assert not fp.return_unknown_bytes
 
 
-def test_parse_file_with_filter(fp, decoders_test_resources):
+def test_parse_file_with_filter(fp, test_gps_file):
     fp.filter = ne.Filter()
     fp.filter.logger.set_level(ne.LogLevel.DEBUG)
-
-    test_gps_file = decoders_test_resources / "BESTUTMBIN.GPS"
+    fp.encode_format = ENCODE_FORMAT.ASCII
+    assert fp.encode_format == ENCODE_FORMAT.ASCII
     with test_gps_file.open("rb") as f:
         assert fp.set_stream(f)
-
-        success = 0
-        expected_meta_data_length = [213, 195]
-        expected_milliseconds = [270605000, 172189053]
-        expected_message_length = [213, 195]
-
         status = STATUS.UNKNOWN
-        fp.encode_format = ENCODE_FORMAT.ASCII
-        assert fp.encode_format == ENCODE_FORMAT.ASCII
-
+        success = 0
         while status != STATUS.STREAM_EMPTY:
             status, message_data, meta_data = fp.read()
             if status == STATUS.SUCCESS:
-                assert meta_data.length == expected_meta_data_length[success]
-                assert meta_data.milliseconds == pytest.approx(expected_milliseconds[success])
-                assert len(message_data.message) == expected_message_length[success]
+                assert meta_data.length == [213, 195][success]
+                assert meta_data.milliseconds == pytest.approx([270605000, 172189053][success])
+                assert len(message_data.message) == [213, 195][success]
                 success += 1
         assert success == 2
 
 
-def test_file_parser_iterator(fp, decoders_test_resources):
+def test_file_parser_iterator(fp, test_gps_file):
     fp.filter = ne.Filter()
     fp.filter.logger.set_level(ne.LogLevel.DEBUG)
-    test_gps_file = decoders_test_resources / "BESTUTMBIN.GPS"
+    fp.encode_format = ENCODE_FORMAT.ASCII
     with test_gps_file.open("rb") as f:
         assert fp.set_stream(f)
         success = 0
-        expected_meta_data_length = [213, 195]
-        expected_milliseconds = [270605000, 172189053]
-        expected_message_length = [213, 195]
-        fp.encode_format = ENCODE_FORMAT.ASCII
         for status, message_data, meta_data in fp:
             if status == STATUS.SUCCESS:
-                assert meta_data.length == expected_meta_data_length[success]
-                assert meta_data.milliseconds == pytest.approx(expected_milliseconds[success])
-                assert len(message_data.message) == expected_message_length[success]
+                assert meta_data.length == [213, 195][success]
+                assert meta_data.milliseconds == pytest.approx([270605000, 172189053][success])
+                assert len(message_data.message) == [213, 195][success]
                 success += 1
-        assert success == 2
+    assert fp.flush(return_flushed_bytes=True) == b""
+    assert success == 2
 
 
 def test_reset(fp):

python/test/test_parser.py

@@ -0,0 +1,95 @@
+################################################################################
+#
+# COPYRIGHT NovAtel Inc, 2022. All rights reserved.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+################################################################################
+#                            DESCRIPTION
+#
+# \file novateltest.hpp
+# \brief Unit tests for OEM Framer, HeaderDecoder, MessageDecoder,
+# Encoder and Filter.
+################################################################################
+
+import novatel_edie as ne
+import pytest
+from novatel_edie import STATUS, ENCODE_FORMAT
+
+
+@pytest.fixture(scope="function")
+def parser():
+    return ne.Parser()
+
+@pytest.fixture(scope="module")
+def test_gps_file(decoders_test_resources):
+    return decoders_test_resources / "BESTUTMBIN.GPS"
+
+def test_logger():
+    # Parser logger
+    level = ne.LogLevel.OFF
+    parser = ne.Parser()
+    logger = parser.logger
+    logger.set_level(level)
+    assert logger.name == "novatel_parser"
+    assert logger.level == level
+    # Parser logger
+    parser.enable_framer_decoder_logging(level, "novatel_parser.log")
+
+
+@pytest.mark.skip(reason="Slow and redundant")
+def test_parser_instantiation(json_db, json_db_path):
+    parser = ne.Parser()
+    parser.load_json_db(json_db)
+    ne.Parser(json_db_path)
+    ne.Parser(json_db)
+
+
+def test_range_cmp(parser):
+    parser.decompress_range_cmp = True
+    assert parser.decompress_range_cmp
+    parser.decompress_range_cmp = False
+    assert not parser.decompress_range_cmp
+
+
+def test_unknown_bytes(parser):
+    parser.return_unknown_bytes = True
+    assert parser.return_unknown_bytes
+    parser.return_unknown_bytes = False
+    assert not parser.return_unknown_bytes
+
+
+def test_parse_file_with_filter(parser, test_gps_file):
+    parser.filter = ne.Filter()
+    parser.filter.logger.set_level(ne.LogLevel.DEBUG)
+    parser.encode_format = ENCODE_FORMAT.ASCII
+    assert parser.encode_format == ENCODE_FORMAT.ASCII
+    with test_gps_file.open("rb") as f:
+        success = 0
+        while chunk := f.read(32):
+            parser.write(chunk)
+            for status, message_data, meta_data in parser:
+                print(status)
+                if status == STATUS.SUCCESS:
+                    assert meta_data.length == [213, 195][success]
+                    assert meta_data.milliseconds == pytest.approx([270605000, 172189053][success])
+                    assert len(message_data.message) == [213, 195][success]
+                    success += 1
+    assert parser.flush(return_flushed_bytes=True) == b""
+    assert success == 2

python/test/test_rxconfig.py

@@ -15,7 +15,7 @@
 
 
 import novatel_edie as ne
-from novatel_edie import HEADER_FORMAT, STATUS, ENCODE_FORMAT
+from novatel_edie import STATUS, ENCODE_FORMAT
 import pytest
 
 

src/decoders/oem/test/oem_test.cpp

@@ -2869,6 +2869,7 @@ TEST_F(FileParserTest, PARSE_FILE_WITH_FILTER)
         eStatus = pclFp->Read(stMessageData, stMetaData);
     }
 
+    ASSERT_EQ(pclFp->Flush(), 0);
     ASSERT_EQ(numSuccess, 2);
 }
 
@@ -2879,6 +2880,141 @@ TEST_F(FileParserTest, RESET)
     ASSERT_TRUE(pclFp->Reset());
 }
 
+// -------------------------------------------------------------------------------------------------------
+// Parser Unit Tests
+// -------------------------------------------------------------------------------------------------------
+class ParserTest : public ::testing::Test
+{
+  protected:
+    static std::unique_ptr<Parser> pclParser;
+
+    static void SetUpTestSuite()
+    {
+        try
+        {
+            pclParser = std::make_unique<Parser>(std::getenv("TEST_DATABASE_PATH"));
+        }
+        catch (JsonDbReaderFailure& e)
+        {
+            std::cout << e.what() << '\n';
+        }
+    }
+
+    static void TearDownTestSuite() { Logger::Shutdown(); }
+};
+
+std::unique_ptr<Parser> ParserTest::pclParser = nullptr;
+
+TEST_F(ParserTest, LOGGER)
+{
+    spdlog::level::level_enum eLevel = spdlog::level::off;
+    ASSERT_NE(spdlog::get("novatel_parser"), nullptr);
+    std::shared_ptr<spdlog::logger> novatelParser = pclParser->GetLogger();
+    pclParser->SetLoggerLevel(eLevel);
+    ASSERT_EQ(novatelParser->level(), eLevel);
+}
+
+TEST_F(ParserTest, FILEPARSER_INSTANTIATION)
+{
+    ASSERT_NO_THROW(Parser fp1);
+    ASSERT_NO_THROW(Parser fp2(std::getenv("TEST_DATABASE_PATH")));
+
+    std::string sTEST_DATABASE_PATH = std::getenv("TEST_DATABASE_PATH");
+    const std::u32string usTEST_DATABASE_PATH(sTEST_DATABASE_PATH.begin(), sTEST_DATABASE_PATH.end());
+    ASSERT_NO_THROW(Parser fp3 = Parser(usTEST_DATABASE_PATH));
+
+    auto jsonDb = JsonDbReader::LoadFile(std::getenv("TEST_DATABASE_PATH"));
+    ASSERT_NO_THROW(Parser fp4(jsonDb));
+}
+
+TEST_F(ParserTest, LOAD_JSON_DB_STRING)
+{
+    auto pclMyJsonDb = JsonDbReader::LoadFile(std::getenv("TEST_DATABASE_PATH"));
+    ASSERT_NO_THROW(pclParser->LoadJsonDb(pclMyJsonDb));
+    ASSERT_NO_THROW(pclParser->LoadJsonDb(nullptr));
+}
+
+TEST_F(ParserTest, LOAD_JSON_DB_U32STRING)
+{
+    std::wstring_convert<std::codecvt_utf8<char32_t>, char32_t> converter;
+    std::u32string u32str = converter.from_bytes(std::getenv("TEST_DATABASE_PATH"));
+    auto pclMyJsonDb = JsonDbReader::LoadFile(u32str);
+    ASSERT_NO_THROW(pclParser->LoadJsonDb(pclMyJsonDb));
+    ASSERT_NO_THROW(pclParser->LoadJsonDb(nullptr));
+}
+
+TEST_F(ParserTest, LOAD_JSON_DB_CHAR_ARRAY)
+{
+    auto pclMyJsonDb = JsonDbReader::LoadFile(std::getenv("TEST_DATABASE_PATH"));
+    ASSERT_NO_THROW(pclParser->LoadJsonDb(pclMyJsonDb));
+    ASSERT_NO_THROW(pclParser->LoadJsonDb(nullptr));
+}
+
+TEST_F(ParserTest, RANGE_CMP)
+{
+    pclParser->SetDecompressRangeCmp(true);
+    ASSERT_TRUE(pclParser->GetDecompressRangeCmp());
+    pclParser->SetDecompressRangeCmp(false);
+    ASSERT_FALSE(pclParser->GetDecompressRangeCmp());
+}
+
+TEST_F(ParserTest, UNKNOWN_BYTES)
+{
+    pclParser->SetReturnUnknownBytes(true);
+    ASSERT_TRUE(pclParser->GetReturnUnknownBytes());
+    pclParser->SetReturnUnknownBytes(false);
+    ASSERT_FALSE(pclParser->GetReturnUnknownBytes());
+}
+
+TEST_F(ParserTest, PARSE_FILE_WITH_FILTER)
+{
+    // Reset the Parser with the database because a previous test assigns it to the nullptr
+    pclParser = std::make_unique<Parser>(std::getenv("TEST_DATABASE_PATH"));
+    auto clFilter = std::make_shared<Filter>();
+    clFilter->SetLoggerLevel(spdlog::level::debug);
+    pclParser->SetFilter(clFilter);
+    ASSERT_EQ(pclParser->GetFilter(), clFilter);
+
+    std::filesystem::path test_gps_file = std::filesystem::path(std::getenv("TEST_RESOURCE_PATH")) / "BESTUTMBIN.GPS";
+    std::ifstream clInputFileStream{test_gps_file.string().c_str(), std::ios::binary};
+
+    MetaDataStruct stMetaData;
+    MessageDataStruct stMessageData;
+
+    int numSuccess = 0;
+    uint32_t uiExpectedMetaDataLength[2] = {213, 195};
+    double dExpectedMilliseconds[2] = {270605000, 172189053};
+    uint32_t uiExpectedMessageLength[2] = {213, 195};
+
+    pclParser->SetEncodeFormat(ENCODE_FORMAT::ASCII);
+    ASSERT_EQ(pclParser->GetEncodeFormat(), ENCODE_FORMAT::ASCII);
+
+    const std::size_t chunkSize = 32;
+    std::vector<char> buffer(chunkSize);
+    while (clInputFileStream.read(buffer.data(), chunkSize) || clInputFileStream.gcount() > 0) {
+        std::size_t n = clInputFileStream.gcount();
+        pclParser->Write(reinterpret_cast<const uint8_t*>(buffer.data()), n);
+        while (true)
+        {
+            STATUS eStatus = pclParser->Read(stMessageData, stMetaData);
+            if (eStatus == STATUS::BUFFER_EMPTY || eStatus == STATUS::INCOMPLETE || eStatus == STATUS::INCOMPLETE_MORE_DATA)
+            {
+                break;
+            }
+            if (eStatus == STATUS::SUCCESS)
+            {
+                ASSERT_EQ(stMetaData.uiLength, uiExpectedMetaDataLength[numSuccess]);
+                ASSERT_DOUBLE_EQ(stMetaData.dMilliseconds, dExpectedMilliseconds[numSuccess]);
+                ASSERT_EQ(stMessageData.uiMessageLength, uiExpectedMessageLength[numSuccess]);
+                numSuccess++;
+            }
+        }
+    }
+
+    ASSERT_EQ(pclParser->Flush(), 0);
+    ASSERT_EQ(numSuccess, 2);
+}
+
 // -------------------------------------------------------------------------------------------------------
 // Novatel Types Unit Tests
 // -------------------------------------------------------------------------------------------------------