SIENTIAPDE-1030

Add tests for loading OPC slots without decoding and connector configurations

tests/orchestrator/activities/test_slot_manager.py

@@ -51,6 +51,28 @@ async def test_load_opc_slots(slot_manager):
     }
 
 
+@mark.asyncio
+async def test_load_opc_slots_no_decode(slot_manager):
+    slot_manager.redis_client.keys.return_value = [
+        "slot:opc_tags:1", "slot:opc_tags:2", "slot:opc_tags:3"]
+
+    slot_manager.get = MagicMock(
+        side_effect=[
+            "value1",
+            "value2",
+            None
+        ]
+    )
+
+    response = await slot_manager.load_opc_slots()
+
+    assert response == {
+        "slot:opc_tags:1": "value1",
+        "slot:opc_tags:2": "value2",
+        "slot:opc_tags:3": None
+    }
+
+
 @mark.asyncio
 async def test_load_active_ingestors(slot_manager):
     slot_manager.redis_client.keys.return_value = [

tests/orchestrator/utils/test_connectors_config.py

@@ -0,0 +1,51 @@
+from os import environ
+from orchestrator.utils.connectors_config import (build_redis_config,
+                                                  build_couchbase_config)
+
+
+def test_build_redis_config_with_env_vars():
+    environ['REDIS_HOST'] = 'localhost'
+    environ['REDIS_PORT'] = '6379'
+    environ['REDIS_USERNAME'] = 'sientia'
+    environ['REDIS_PASSWORD'] = 'sientia'
+    assert build_redis_config() == {
+        'host': 'localhost',
+        'port': 6379,
+        'username': 'sientia',
+        'password': 'sientia'
+    }
+
+
+def test_build_couchbase_config_with_env_vars():
+    environ['COUCHBASE_CONNECTION_STRING'] = 'couchbase://localhost'
+    environ['COUCHBASE_USERNAME'] = 'sientia'
+    environ['COUCHBASE_PASSWORD'] = 'sientia'
+    assert build_couchbase_config() == {
+        'connection_string': 'couchbase://localhost',
+        'username': 'sientia',
+        'password': 'sientia'
+    }
+
+
+def test_build_redis_config_with_defaults():
+    environ.pop('REDIS_HOST', None)
+    environ.pop('REDIS_PORT', None)
+    environ.pop('REDIS_USERNAME', None)
+    environ.pop('REDIS_PASSWORD', None)
+    assert build_redis_config() == {
+        'host': 'localhost',
+        'port': 6379,
+        'username': None,
+        'password': None
+    }
+
+
+def test_build_couchbase_config_with_defaults():
+    environ.pop('COUCHBASE_CONNECTION_STRING', None)
+    environ.pop('COUCHBASE_USERNAME', None)
+    environ.pop('COUCHBASE_PASSWORD', None)
+    assert build_couchbase_config() == {
+        'connection_string': 'couchbase://localhost',
+        'username': 'sientia',
+        'password': 'sientia'
+    }

tests/orchestrator/utils/test_converters.py

@@ -0,0 +1,15 @@
+from orchestrator.utils.converters import parse_frequency
+
+
+def test_parse_frequency():
+    assert parse_frequency("1s") == 1
+    assert parse_frequency("1m") == 60
+    assert parse_frequency("1h") == 60 * 60
+    assert parse_frequency("1d") == 60 * 60 * 24
+
+    try:
+        parse_frequency("1")
+    except ValueError as e:
+        assert str(e) == "Invalid frequency"
+    else:
+        assert False