Enhance error handling and logging in writers; refactor tests to use pytest

Author: oto-macenauer-absa

src/writer_eventbridge.py

@@ -18,7 +18,7 @@ def init(logger, CONFIG):
 def write(topicName, message):
     if not EVENT_BUS_ARN:
         _logger.debug("No EventBus Arn - skipping")
-        return True
+        return True, None
 
     try:
         _logger.debug(f"Sending to eventBridge {topicName}")
@@ -33,10 +33,12 @@ def write(topicName, message):
             ]
         )
         if response["FailedEntryCount"] > 0:
-            _logger.error(str(response))
-            return False
+            msg = str(response)
+            _logger.error(msg)
+            return False, msg
     except Exception as e:
-        _logger.error(f"The EventBridge writer failed with unknown error: {str(e)}")
-        return False
+        err_msg = f"The EventBridge writer failed with unknown error: {str(e)}"
+        _logger.error(err_msg)
+        return False, err_msg
 
-    return True
+    return True, None

src/writer_postgres.py

@@ -190,10 +190,10 @@ def write(topicName, message):
     try:
         if not POSTGRES["database"]:
             _logger.debug("No Postgres - skipping")
-            return True
+            return True, None
         if psycopg2 is None:
             _logger.debug("psycopg2 not available - skipping actual Postgres write")
-            return True
+            return True, None
 
         with psycopg2.connect(
             database=POSTGRES["database"],
@@ -210,12 +210,14 @@ def write(topicName, message):
                 elif topicName == "public.cps.za.test":
                     postgres_test_write(cursor, "public_cps_za_test", message)
                 else:
-                    _logger.error(f"unknown topic for postgres {topicName}")
-                    return False
+                    msg = f"unknown topic for postgres {topicName}"
+                    _logger.error(msg)
+                    return False, msg
 
             connection.commit()
-    except Exception as e:
-        _logger.error(f"The Postgres writer with failed unknown error: {str(e)}")
-        return False
+    except Exception as e:  # pragma: no cover - defensive (still tested though)
+        err_msg = f"The Postgres writer with failed unknown error: {str(e)}"
+        _logger.error(err_msg)
+        return False, err_msg
 
-    return True
+    return True, None

tests/test_conf_validation.py

@@ -1,7 +1,7 @@
 import os
 import json
-import unittest
 from glob import glob
+import pytest
 
 CONF_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "conf")
 
@@ -17,50 +17,41 @@ def load_json(path):
     with open(path, "r") as f:
         return json.load(f)
 
-class TestConfigFiles(unittest.TestCase):
-    def test_config_files_have_required_keys(self):
-        # Pick up any config*.json (excluding access and topics)
-        config_files = [
-            f for f in glob(os.path.join(CONF_DIR, "config*.json"))
-            if os.path.basename(f) not in {"access.json"}
-        ]
-        self.assertTrue(config_files, "No config files found matching pattern config*.json")
-        for path in config_files:
-            with self.subTest(config=path):
-                data = load_json(path)
-                missing = REQUIRED_CONFIG_KEYS - data.keys()
-                self.assertFalse(missing, f"Config {path} missing keys: {missing}")
-
-    def test_access_json_structure(self):
-        path = os.path.join(CONF_DIR, "access.json")
+@pytest.fixture(scope="module")
+def config_files():
+    files = [
+        f for f in glob(os.path.join(CONF_DIR, "config*.json"))
+        if os.path.basename(f) not in {"access.json"}
+    ]
+    assert files, "No config files found matching pattern config*.json"
+    return files
+
+@pytest.mark.parametrize("key", sorted(REQUIRED_CONFIG_KEYS))
+def test_config_files_have_required_keys(config_files, key):
+    for path in config_files:
         data = load_json(path)
-        self.assertIsInstance(data, dict, "access.json must contain an object mapping topic -> list[user]")
-        for topic, users in data.items():
-            with self.subTest(topic=topic):
-                self.assertIsInstance(topic, str)
-                self.assertIsInstance(users, list, f"Topic {topic} value must be a list of users")
-                self.assertTrue(all(isinstance(u, str) for u in users), f"All users for topic {topic} must be strings")
-
-    def test_topic_json_schemas_basic(self):
-        topic_files = glob(os.path.join(CONF_DIR, "topic_*.json"))
-        self.assertTrue(topic_files, "No topic_*.json files found")
-        for path in topic_files:
-            with self.subTest(topic_file=path):
-                schema = load_json(path)
-                # Basic required structure
-                self.assertEqual(schema.get("type"), "object", "Schema root 'type' must be 'object'")
-                props = schema.get("properties")
-                self.assertIsInstance(props, dict, "Schema must define 'properties' object")
-                required = schema.get("required")
-                self.assertIsInstance(required, list, "Schema must define 'required' list")
-                # Each required field is present in properties
-                missing_props = [r for r in required if r not in props]
-                self.assertFalse(missing_props, f"Required fields missing in properties: {missing_props}")
-                # Ensure property entries themselves have at least a type
-                for name, definition in props.items():
-                    self.assertIsInstance(definition, dict, f"Property {name} definition must be an object")
-                    self.assertIn("type", definition, f"Property {name} must specify a 'type'")
-
-if __name__ == "__main__":  # pragma: no cover
-    unittest.main()
-
+        assert key in data, f"Config {path} missing key: {key}"
+
+def test_access_json_structure():
+    path = os.path.join(CONF_DIR, "access.json")
+    data = load_json(path)
+    assert isinstance(data, dict), "access.json must contain an object mapping topic -> list[user]"
+    for topic, users in data.items():
+        assert isinstance(topic, str)
+        assert isinstance(users, list), f"Topic {topic} value must be a list of users"
+        assert all(isinstance(u, str) for u in users), f"All users for topic {topic} must be strings"
+
+@pytest.mark.parametrize("topic_file", glob(os.path.join(CONF_DIR, "topic_*.json")))
+def test_topic_json_schemas_basic(topic_file):
+    assert topic_file, "No topic_*.json files found"
+    schema = load_json(topic_file)
+    assert schema.get("type") == "object", "Schema root 'type' must be 'object'"
+    props = schema.get("properties")
+    assert isinstance(props, dict), "Schema must define 'properties' object"
+    required = schema.get("required")
+    assert isinstance(required, list), "Schema must define 'required' list"
+    missing_props = [r for r in required if r not in props]
+    assert not missing_props, f"Required fields missing in properties: {missing_props}"
+    for name, definition in props.items():
+        assert isinstance(definition, dict), f"Property {name} definition must be an object"
+        assert "type" in definition, f"Property {name} must specify a 'type'"