Feat: unit tests for json_message and log_util (#29)

* add unit tests for json_message utility
* add test_log_util.py
* remove unused path_builder code
* test_path_builder.py: more concise dict definition

Co-authored-by: Geary-Layne <[email protected]>

python/idsse_common/idsse/common/json_message.py

@@ -10,19 +10,23 @@
 # ------------------------------------------------------------------------------
 
 import json
-from typing import Tuple, Union
+from typing import Tuple, Union, Optional
 from uuid import uuid4, UUID
 
 
-def get_corr_id(message: Union[str, dict]) -> Tuple[str, UUID, str]:
+def get_corr_id(
+    message: Union[str, dict]
+) -> Optional[Tuple[Optional[str], Optional[Union[UUID, str]], Optional[str]]]:
     """Extract the correlation id from a json message.
-       The correlation id is made of three part, originator, uuid, issue date/time
+       The correlation id is made of three parts: originator, uuid, issue date/time
 
     Args:
-        message (Union[str, json]): The message to be search as either a string or json obj
+        message (Union[str, json]): The message to be searched as either a string or json obj
 
     Returns:
-        Tuple[str, uuid, str]: A tuple containing originator, uuid, and issue date/time
+        Optional[Tuple[Optional[str], Optional[Union[UUID, str]], Optional[str]]]:
+            A tuple containing originator, uuid, and issue date/time, or None if a given part
+            was not found. Returns simply None if no parts found
     """
     if isinstance(message, str):
         message = json.loads(message)
@@ -42,8 +46,8 @@ def get_corr_id(message: Union[str, dict]) -> Tuple[str, UUID, str]:
 
 def add_corr_id(message: Union[dict, str],
                 originator: str,
-                uuid_: Union[UUID, str] = None,
-                issue_dt: str = None) -> dict:
+                uuid_: Optional[Union[UUID, str]] = None,
+                issue_dt: Optional[str] = None) -> dict:
     """Add (or overwrites) the three part correlation id to a json message
 
     Args:

python/idsse_common/idsse/common/log_util.py

@@ -17,7 +17,7 @@
 import uuid
 from contextvars import ContextVar
 from datetime import datetime
-from typing import Union, Optional
+from typing import Union, Optional, List
 
 from .utils import to_iso
 
@@ -51,21 +51,24 @@ def set_corr_id_context_var(
         corr_id_context_var.set(f'{originator};{key};_')
 
 
-def get_corr_id_context_var_str():
+def get_corr_id_context_var_str() -> str:
     """Getter for correlation ID ContextVar name"""
     return corr_id_context_var.get()
 
 
-def get_corr_id_context_var_parts():
+def get_corr_id_context_var_parts() -> List[str]:
     """Split correlation ID ContextVar into its parts, such as [originator, key, issue_datetime]"""
     return corr_id_context_var.get().split(';')
 
 
 class AddCorrelationIdFilter(logging.Filter):
     """"Provides correlation id parameter for the logger"""
-    def filter(self, record):
-        record.corr_id = corr_id_context_var.get()
-        return True
+    def filter(self, record: logging.LogRecord) -> bool:
+        try:
+            record.corr_id = corr_id_context_var.get()
+            return True
+        except LookupError:  # couldn't add corr_id since it is not set
+            return False
 
 
 class CorrIdFilter(logging.Filter):

python/idsse_common/idsse/common/path_builder.py

@@ -19,7 +19,7 @@
 import os
 import re
 from datetime import datetime, timedelta
-from typing import Dict, Self, Tuple, Union
+from typing import Dict, Self, Union
 
 from .utils import TimeDelta
 
@@ -185,7 +185,6 @@ def parse_dir(self, dir_: str) -> dict:
         Returns:
             dict: Lookup of all information identified and extracted
         """
-        # return self._parse(dir_, self.dir_fmt)
         return self._parse_times(dir_, self.dir_fmt)
 
     def parse_filename(self, filename: str) -> dict:
@@ -197,7 +196,6 @@ def parse_filename(self, filename: str) -> dict:
         Returns:
             dict: Lookup of all information identified and extracted
         """
-        # return self._parse(filename, self.filename_fmt)
         filename = os.path.basename(filename)
         return self._parse_times(filename, self.filename_fmt)
 
@@ -210,7 +208,6 @@ def parse_path(self, path: str) -> dict:
         Returns:
             dict: Lookup of all information identified and extracted
         """
-        # return self._parse(path, self.path_fmt)
         return self._parse_times(path, self.path_fmt)
 
     def get_issue(self, path: str) -> datetime:
@@ -365,38 +362,3 @@ def parse_args(key: str, value: str, result: Dict):
                 parse_args(res.group(), vals[i][res.span()[0]:], time_args)
 
         return time_args
-
-    def _parse(self, string: str, format_str: str) -> Dict:
-        def get_between(query_str: str, pre_off: str, post_off: str) -> Tuple[str, str]:
-            idx1 = query_str.index(pre_off) + len(pre_off)
-            idx2 = query_str.index(post_off, idx1)
-            return query_str[idx1:idx2], query_str[idx2:]
-
-        def parse_args(key: str, value: str, result: Dict):
-            for arg in key.split('{')[1:]:
-                var_name, var_size = arg.split(':')
-                var_type = var_size[-2:-1]
-                var_size = int(var_size[:-2])
-                match var_type:
-                    case 'd':
-                        result[var_name] = int(value[:var_size])
-                    case _:
-                        raise ValueError(f'Unknown format type: {var_type}')
-                key = key[var_size:]
-                value = value[var_size:]
-
-        constants = [part for part in re.split(r'{.*?}', format_str) if part]
-
-        arg_lookup = {}
-        for i in range(1, len(constants)):
-            pre = constants[i-1]
-            post = constants[i]
-            format_between, format_str = get_between(format_str, pre, post)
-            string_between, string = get_between(string, pre, post)
-            arg_lookup[format_between] = string_between
-
-        time_args = {}
-        for key, value in arg_lookup.items():
-            parse_args(key, value, time_args)
-
-        return time_args

python/idsse_common/test/test_json_message.py

@@ -0,0 +1,98 @@
+"""Test suite for json_message.py"""
+# ----------------------------------------------------------------------------------
+# Created on Fri Oct 20 2023
+#
+# Copyright (c) 2023 Regents of the University of Colorado. All rights reserved. (1)
+# Copyright (c) 2023 Colorado State University. All rights reserved. (2)
+#
+# Contributors:
+#    Mackenzie Grimes (2)
+#
+# ----------------------------------------------------------------------------------
+# pylint: disable=missing-function-docstring
+
+import json
+from datetime import datetime, UTC
+from uuid import uuid4 as uuid
+
+from idsse.common.json_message import get_corr_id, add_corr_id
+from idsse.common.utils import to_iso
+
+# test data
+EXAMPLE_ORIGINATOR = 'idsse'
+EXAMPLE_UUID = str(uuid())
+EXAMPLE_ISSUE_DT = to_iso(datetime.now(UTC))
+
+EXAMPLE_MESSAGE = {
+    'corrId': {
+        'originator': EXAMPLE_ORIGINATOR, 'uuid': EXAMPLE_UUID, 'issueDt': EXAMPLE_ISSUE_DT
+    }
+}
+
+
+def test_get_corr_id_dict():
+    result = get_corr_id(EXAMPLE_MESSAGE)
+    assert result == (EXAMPLE_ORIGINATOR, EXAMPLE_UUID, EXAMPLE_ISSUE_DT)
+
+
+def test_get_corr_id_str():
+    result = get_corr_id(json.dumps(EXAMPLE_MESSAGE))
+    assert result == (EXAMPLE_ORIGINATOR, EXAMPLE_UUID, EXAMPLE_ISSUE_DT)
+
+
+def test_get_corr_id_empty_corr_id():
+    result = get_corr_id({'other_data': 123})
+    assert result is None
+
+
+def test_get_corr_id_originator():
+    result = get_corr_id({'corrId': {'originator': EXAMPLE_ORIGINATOR}})
+    assert result == (EXAMPLE_ORIGINATOR, None, None)
+
+
+def test_get_corr_id_uuid():
+    result = get_corr_id({'corrId': {'uuid': EXAMPLE_UUID}})
+    assert result == (None, EXAMPLE_UUID, None)
+
+
+def test_get_corr_id_issue_dt():
+    result = get_corr_id({'corrId': {'issueDt': EXAMPLE_ISSUE_DT}})
+    assert result == (None, None, EXAMPLE_ISSUE_DT)
+
+
+def test_json_get_corr_id_failure():
+    bad_message = {'invalid': 'message'}
+    result = get_corr_id(bad_message)
+    assert result is None
+
+
+def test_add_corr_id():
+    new_originator = 'different_app'
+    new_message = add_corr_id(EXAMPLE_MESSAGE, new_originator)
+
+    # blank issueDt and random uuid should have been returned
+    result = new_message['corrId']
+    assert result['originator'] == new_originator
+    assert result['issueDt'] == '_'
+    assert result['uuid'] != EXAMPLE_UUID
+
+
+def test_add_corr_id_str():
+    new_originator = 'different_app'
+    new_message = add_corr_id(json.dumps(EXAMPLE_MESSAGE), new_originator)
+    assert new_message['corrId']['originator'] == new_originator
+
+
+def test_add_corr_id_uuid():
+    new_uuid = uuid()
+    new_message = add_corr_id(EXAMPLE_MESSAGE, EXAMPLE_ORIGINATOR, uuid_=new_uuid)
+
+    # blank issueDt and random uuid should have been returned
+    assert new_message['corrId']['uuid'] == str(new_uuid)
+    assert new_message['corrId']['originator'] == EXAMPLE_ORIGINATOR
+
+
+def test_add_corr_id_issue_dt():
+    new_issue = to_iso(datetime.now(UTC))
+    new_message = add_corr_id(EXAMPLE_MESSAGE, EXAMPLE_ORIGINATOR, issue_dt=new_issue)
+    assert new_message['corrId']['issueDt'] == new_issue

python/idsse_common/test/test_log_util.py

@@ -0,0 +1,96 @@
+"""Test suite for json_message.py"""
+# ----------------------------------------------------------------------------------
+# Created on Fri Oct 20 2023
+#
+# Copyright (c) 2023 Regents of the University of Colorado. All rights reserved. (1)
+# Copyright (c) 2023 Colorado State University. All rights reserved. (2)
+#
+# Contributors:
+#    Mackenzie Grimes (2)
+#
+# ----------------------------------------------------------------------------------
+# pylint: disable=missing-function-docstring,redefined-outer-name,invalid-name,unused-argument
+
+import logging
+import logging.config
+from datetime import datetime, UTC
+from uuid import uuid4 as uuid
+
+from idsse.common.log_util import (
+    set_corr_id_context_var,
+    get_corr_id_context_var_parts,
+    get_corr_id_context_var_str,
+    get_default_log_config,
+    AddCorrelationIdFilter
+)
+from idsse.common.utils import to_iso
+
+logger = logging.getLogger(__name__)
+
+# test data
+EXAMPLE_ORIGINATOR = 'idsse'
+EXAMPLE_UUID = uuid()
+EXAMPLE_ISSUE_DT = datetime.now(UTC)
+EXAMPLE_ISSUE_STR = to_iso(EXAMPLE_ISSUE_DT)
+EXAMPLE_LOG_MESSAGE = 'hello world'
+
+
+def test_add_correlation_id_filter():
+    filter_object = AddCorrelationIdFilter()
+    test_record = logging.LogRecord(
+        name=EXAMPLE_ORIGINATOR,
+        level=logging.INFO,
+        pathname='./some/path/to/file.py',
+        lineno=123,
+        msg='hello world',
+        args=None,
+        exc_info=(None, None, None)
+    )
+
+    # corr_id does not exist yet, so filter fails
+    assert not filter_object.filter(test_record)
+
+    # set corr_id so filter does match
+    set_corr_id_context_var(EXAMPLE_ORIGINATOR)
+    assert filter_object.filter(test_record)
+
+
+def test_set_corr_id():
+    set_corr_id_context_var(EXAMPLE_ORIGINATOR, EXAMPLE_UUID, EXAMPLE_ISSUE_STR)
+
+    expected_result = [EXAMPLE_ORIGINATOR, str(EXAMPLE_UUID), EXAMPLE_ISSUE_STR]
+    assert get_corr_id_context_var_parts() == expected_result
+    assert get_corr_id_context_var_str() == ';'.join(expected_result)
+
+
+def test_set_corr_id_datetime():
+    set_corr_id_context_var(EXAMPLE_ORIGINATOR, key=EXAMPLE_UUID, issue_dt=EXAMPLE_ISSUE_DT)
+
+    assert get_corr_id_context_var_parts() == [
+        EXAMPLE_ORIGINATOR, str(EXAMPLE_UUID), EXAMPLE_ISSUE_STR
+    ]
+
+
+def test_get_default_log_config_with_corr_id(capsys):
+    logging.config.dictConfig(get_default_log_config('INFO'))
+    corr_id = get_corr_id_context_var_str()
+
+    logger.debug(msg=EXAMPLE_LOG_MESSAGE)
+    stdout = capsys.readouterr().out  # capture std output from test run
+
+    # should not be logging DEBUG if default log config handled level correctly
+    assert stdout == ''
+    logger.info(msg=EXAMPLE_LOG_MESSAGE)
+    stdout = capsys.readouterr().out
+
+    assert EXAMPLE_LOG_MESSAGE in stdout
+    assert corr_id in stdout
+
+
+def test_get_default_log_config_no_corr_id(capsys):
+    logging.config.dictConfigClass(get_default_log_config('DEBUG', False))
+    corr_id = get_corr_id_context_var_str()
+
+    logger.debug('hello world')
+    stdout = capsys.readouterr().out
+    assert corr_id not in stdout

python/idsse_common/test/test_path_builder.py

@@ -171,11 +171,23 @@ def test_get_valid_returns_none_if_args_empty():
 def test_get_valid_from_time_args_calculates_based_on_lead(path_builder: PathBuilder):
     parsed_dict = path_builder.parse_path(EXAMPLE_FULL_PATH)
     result_valid: datetime = PathBuilder.get_valid_from_time_args(parsed_args=parsed_dict)
-
     assert result_valid == EXAMPLE_VALID
 
 
 def test_get_lead_from_time_args(path_builder: PathBuilder):
     parsed_dict = path_builder.parse_path(EXAMPLE_FULL_PATH)
     lead_result: timedelta = PathBuilder.get_lead_from_time_args(parsed_dict)
     assert lead_result.seconds == EXAMPLE_LEAD.minute * 60
+
+
+def test_calculate_issue_from_valid_and_lead():
+    parsed_dict = {
+        'valid.year': 1970,
+        'valid.month': 10,
+        'valid.day': 3,
+        'valid.hour': 14,
+        'lead.hour': 2
+    }
+
+    result_issue = PathBuilder.get_issue_from_time_args(parsed_args=parsed_dict)
+    assert result_issue == EXAMPLE_ISSUE