diff --git a/docker-compose.yml b/docker-compose.yml index baab896..e3aba82 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -117,6 +117,19 @@ services: networks: - sientia-network + mongodb: + image: mongo:7.0 + container_name: mongodb + ports: + - "27017:27017" + environment: + MONGO_INITDB_ROOT_USERNAME: sientia + MONGO_INITDB_ROOT_PASSWORD: sientia + volumes: + - mongodb_data:/data/db + networks: + - sientia-network + networks: sientia-network: @@ -130,4 +143,6 @@ volumes: redis_data: driver: local kafka_data: + driver: local + mongodb_data: driver: local \ No newline at end of file diff --git a/input_sample.json b/input_sample.json index 4a89997..e8d3641 100644 --- a/input_sample.json +++ b/input_sample.json @@ -1,5 +1,37 @@ +[ { "schedule_name": "orchestrator-test", "pipelines_query": "SELECT pipelines.*, models FROM `pipelines` JOIN `models` ON KEYS pipelines.model_id;", "opc_servers_query": "select META().id, opc_servers.* from `opc_servers`;" -} \ No newline at end of file +}, +{ + "schedule_name": "orchestrator-test", + "pipelines_query": { + "collection": "pipelines", + "aggregation": [ + { + "$lookup": { + "from": "models", + "localField": "model_id", + "foreignField": "id", + "as": "model_docs" + } + }, + { + "$addFields": { + "models": { "$arrayElemAt": ["$model_docs", 0] } + } + }, + { + "$project": { + "model_docs": 0 + } + } + ] + }, + "opc_servers_query": { + "collection": "opc-servers", + "filters": {} + } +} +] \ No newline at end of file diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py index 1265539..5049e3f 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -1,6 +1,8 @@ from temporalio import activity, workflow from temporalio.client import Client +from orchestrator.activities.mongo_db import MongoDB + with workflow.unsafe.imports_passed_through(): from orchestrator.activities.couchbase import Couchbase from orchestrator.activities.temporal_manager import TemporalManager @@ -11,21 +13,23 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.notifications.handlers import NotificationHandler -class Activities(Couchbase, TemporalManager, SlotManager, Formatters): +class Activities( # Couchbase, + TemporalManager, SlotManager, Formatters, MongoDB): def __init__(self, temporal_client: Client, - couchbase_config: dict[str, Any], + # couchbase_config: dict[str, Any], redis_config: dict[str, Any], + mongodb_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler): # Initialize parent classes - Couchbase.__init__(self, connection_string=couchbase_config['connection_string'], - username=couchbase_config['username'], - password=couchbase_config['password'], - logger=logger, - notification_handler=notification_handler) + # Couchbase.__init__(self, connection_string=couchbase_config['connection_string'], + # username=couchbase_config['username'], + # password=couchbase_config['password'], + # logger=logger, + # notification_handler=notification_handler) TemporalManager.__init__(self, temporal_client=temporal_client, @@ -44,9 +48,15 @@ class Activities(Couchbase, TemporalManager, SlotManager, Formatters): logger=logger, notification_handler=notification_handler) + MongoDB.__init__(self, + connection_string=mongodb_config['connection_string'], + database_name=mongodb_config['database_name'], + logger=logger, + notification_handler=notification_handler) + @activity.defn(name="prepare_activity") async def prepare_activity(self, input_data: dict[str, Any]): await super().prepare_activity(input_data) def shutdown(self): - Couchbase.shutdown(self) + MongoDB.shutdown(self) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py new file mode 100644 index 0000000..e5c663f --- /dev/null +++ b/orchestrator/activities/mongo_db.py @@ -0,0 +1,186 @@ +from temporalio import workflow, activity + +with workflow.unsafe.imports_passed_through(): + from typing import Any + import traceback + from logging import Logger + from pymongo import MongoClient + from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.notifications.models import NotificationLevel + from sientia_do.temporal.activities.base import BaseActivity + + +def clear_mongo_id(docs: list) -> list: + """ + Remove the MongoDB internal `_id` field from the document. + + Args: + docs (list): The document to clear. + + Returns: + list: The documents without the `_id` field. + """ + for doc in docs: + if isinstance(doc, list): + clear_mongo_id(doc) + + elif isinstance(doc, dict): + if "_id" in doc: + del doc["_id"] + + for key, value in doc.items(): + if isinstance(value, list): + clear_mongo_id(value) + elif isinstance(value, dict): + clear_mongo_id([value]) + + return docs + + +class MongoDB(BaseActivity): + def __init__(self, connection_string: str, database_name: str, + logger: Logger, + notification_handler: NotificationHandler): + self.connection_string = connection_string + self.database_name = database_name + + self.client = MongoClient( + self.connection_string, serverSelectionTimeoutMS=5000) + self.client.server_info() # Trigger an exception if connection fails + + self.database = self.client[self.database_name] + + # Initialize MongoDB client here (omitted for brevity) + logger.info("MongoDB connection initialized") + + BaseActivity.__init__(self, + logger=logger, + notification_handler=notification_handler) + + def shutdown(self): + """ + Close the MongoDB client connection. + """ + try: + if self.client: + self.logger.info("Closing MongoDB connection...") + self.client.close() + self.logger.info("MongoDB connection closed successfully") + except Exception as e: + self.logger.error(f"Failed to close MongoDB connection: {e}") + + def __del__(self): + """ + Destructor to ensure MongoDB client is closed when the object is deleted. + """ + self.shutdown() + + @activity.defn(name="find_documents_in_mongodb",) + async def find_documents_in_mongodb(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: + """ + Find documents in a MongoDB collection based on the provided query parameters. + + Args: + - input_data (dict): Input data containing query parameters. Contains: + - query (dict): Query parameters to filter documents. + + Returns: + list[dict]: List of documents matching the query. + """ + + query = input_data.get("query", {}) + + collection_name = query.get("collection") + if not collection_name: + raise ValueError("Collection name must be provided in the query.") + + filters = query.get("filters", {}) + + self.logger.info( + f"Loading documents from collection '{collection_name}' with filters: {filters}") + + try: + collection = self.database[collection_name] + + documents = list(collection.find(filters, {"_id": 0})) + + documents = clear_mongo_id(documents) + + self.logger.info( + f"Loaded {len(documents)} documents from collection '{collection_name}'") + + self.logger.debug( + f"Documents loaded: {documents}" + ) + + return documents + + except Exception as e: + trace = traceback.format_exc() + self.notification_handler.build_and_send_notification( + notification_id="MONGODB_QUERY_ERROR", + message=f"Failed to execute MongoDB query: {e}", + block="load_query_from_mongodb", + level=NotificationLevel.ERROR, + attachment_content=trace + ) + self.logger.error(trace) + + raise e + + @activity.defn(name="aggregate_documents_in_mongodb") + async def aggregate_documents_in_mongodb(self, + input_data: dict[str, Any]) -> list[dict[str, Any]]: + """ + Aggregate documents in a MongoDB collection based on the provided aggregation pipeline. + + Args: + - input_data (dict): Input data containing aggregation parameters. Contains: + - query (dict): Query parameters to filter documents. + + Returns: + list[dict]: List of aggregated documents. + """ + + query = input_data.get("query", {}) + + collection_name = query.get("collection") + if not collection_name: + raise ValueError("Collection name must be provided in the query.") + aggregation = query.get("aggregation") + if not aggregation: + raise ValueError("Aggregation must be provided.") + aggregation.append({"$project": {"_id": 0}}) + + self.logger.info( + f"Aggregating documents from collection '{collection_name}' with aggregation: {aggregation}") + + try: + collection = self.database[collection_name] + + aggregated_documents = list( + collection.aggregate(aggregation)) + + aggregated_documents = clear_mongo_id(aggregated_documents) + + self.logger.info( + f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'") + + self.logger.debug( + f"Aggregation result: {aggregated_documents}" + ) + + return aggregated_documents + + except Exception as e: + trace = traceback.format_exc() + self.notification_handler.build_and_send_notification( + notification_id="MONGODB_AGGREGATION_ERROR", + message=f"Failed to execute MongoDB aggregation: {e}", + block="aggregate_documents_in_mongodb", + level=NotificationLevel.ERROR, + attachment_content=trace + ) + self.logger.error(trace) + + raise e diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 2e5e80f..3808f72 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -10,6 +10,18 @@ def build_redis_config(): } +def build_mongodb_config(): + username = getenv('MONGODB_USERNAME', 'sientia') + password = getenv('MONGODB_PASSWORD', 'sientia') + uri = getenv('MONGODB_URL', 'localhost:27017') + + connection_string = f'mongodb://{username}:{password}@{uri}' + return { + 'connection_string': connection_string, + 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia') + } + + def build_couchbase_config(): return { 'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'), diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 0814b25..383c1be 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -8,8 +8,9 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.workflows.orchestrator import Orchestrator from orchestrator.activities.activities import Activities from orchestrator.utils.connectors_config import ( - build_couchbase_config, + # build_couchbase_config, build_redis_config, + build_mongodb_config ) from sientia_do.notifications.handlers import NotificationHandler from sientia_do.temporal.utils.logger import get_logger @@ -40,8 +41,9 @@ async def main(): activities = Activities( temporal_client=temporal_client, - couchbase_config=build_couchbase_config(), + # couchbase_config=build_couchbase_config(), redis_config=build_redis_config(), + mongodb_config=build_mongodb_config(), logger=logger, notification_handler=notification_handler ) @@ -62,7 +64,10 @@ async def main(): activities.update_slots, activities.delete_slots, # Couchbase - activities.load_query_from_couchbase, + # activities.load_query_from_couchbase, + # MongoDB + activities.aggregate_documents_in_mongodb, + activities.find_documents_in_mongodb, # Temporal activities.load_schedule, activities.create_schedules, diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index 12550d5..b731eeb 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -26,8 +26,17 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) + # pipeline_config_handler = workflow.execute_local_activity_method( + # Activities.load_query_from_couchbase, + # { + # 'query': input_data['pipelines_query'] + # }, + # retry_policy=retry_policy, + # start_to_close_timeout=timedelta(seconds=60) + # ) + pipeline_config_handler = workflow.execute_local_activity_method( - Activities.load_query_from_couchbase, + Activities.aggregate_documents_in_mongodb, { 'query': input_data['pipelines_query'] }, @@ -35,8 +44,17 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) + # opc_servers_handler = workflow.execute_local_activity_method( + # Activities.load_query_from_couchbase, + # { + # 'query': input_data['opc_servers_query'] + # }, + # retry_policy=retry_policy, + # start_to_close_timeout=timedelta(seconds=60) + # ) + opc_servers_handler = workflow.execute_local_activity_method( - Activities.load_query_from_couchbase, + Activities.find_documents_in_mongodb, { 'query': input_data['opc_servers_query'] }, diff --git a/requirements.txt b/requirements.txt index 07d06ae..25595c6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,4 +3,5 @@ psycopg2-binary sqlalchemy redis couchbase +pymongo git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14 diff --git a/tests/orchestrator/activities/test_activities.py b/tests/orchestrator/activities/test_activities.py index 997a55a..7c0fbb3 100644 --- a/tests/orchestrator/activities/test_activities.py +++ b/tests/orchestrator/activities/test_activities.py @@ -1,18 +1,21 @@ from unittest.mock import patch, MagicMock, ANY from pytest import mark +from orchestrator.activities import mongo_db from orchestrator.activities.activities import Activities -from orchestrator.activities.couchbase import Couchbase +from orchestrator.activities.mongo_db import MongoDB from orchestrator.activities.temporal_manager import TemporalManager from orchestrator.activities.slot_manager import SlotManager from orchestrator.activities.formatters import Formatters @patch('orchestrator.activities.couchbase.Couchbase.__init__') +@patch('orchestrator.activities.mongo_db.MongoDB.__init__') @patch('orchestrator.activities.temporal_manager.TemporalManager.__init__') @patch('orchestrator.activities.slot_manager.SlotManager.__init__') @patch('orchestrator.activities.formatters.Formatters.__init__') def test___init__(mock_formatters_init, mock_slot_manager_init, mock_temporal_manager_init, + mock_mongodb_init, mock_couchbase_init): couchbase_config = { @@ -21,6 +24,11 @@ def test___init__(mock_formatters_init, mock_slot_manager_init, 'password': 'password' } + mongo_db_config = { + 'connection_string': 'mongodb://localhost:27017', + 'database_name': 'test_db' + } + redis_config = { 'host': 'localhost', 'port': 6379, @@ -34,14 +42,16 @@ def test___init__(mock_formatters_init, mock_slot_manager_init, activities = Activities( temporal_client=temporal_client, - couchbase_config=couchbase_config, + # couchbase_config=couchbase_config, redis_config=redis_config, + mongodb_config=mongo_db_config, logger=logger, notification_handler=notification_handler ) assert isinstance(activities, Activities) - assert isinstance(activities, Couchbase) + # assert isinstance(activities, Couchbase) + assert isinstance(activities, MongoDB) assert isinstance(activities, TemporalManager) assert isinstance(activities, SlotManager) assert isinstance(activities, Formatters) @@ -56,11 +66,19 @@ def test___init__(mock_formatters_init, mock_slot_manager_init, notification_handler=notification_handler ) - mock_couchbase_init.assert_called_once_with( + # mock_couchbase_init.assert_called_once_with( + # ANY, + # connection_string=couchbase_config['connection_string'], + # username=couchbase_config['username'], + # password=couchbase_config['password'], + # logger=logger, + # notification_handler=notification_handler + # ) + + mock_mongodb_init.assert_called_once_with( ANY, - connection_string=couchbase_config['connection_string'], - username=couchbase_config['username'], - password=couchbase_config['password'], + connection_string='mongodb://localhost:27017', + database_name='test_db', logger=logger, notification_handler=notification_handler ) @@ -80,7 +98,7 @@ def test___init__(mock_formatters_init, mock_slot_manager_init, @mark.asyncio -@patch('orchestrator.activities.couchbase.Cluster') +@patch('orchestrator.activities.mongo_db.MongoClient') async def test_prepare_activity(_mock_cluster): couchbase_config = { 'connection_string': 'couchbase://localhost', @@ -88,6 +106,11 @@ async def test_prepare_activity(_mock_cluster): 'password': 'password' } + mongo_db_config = { + 'connection_string': 'mongodb://localhost:27017', + 'database_name': 'test_db' + } + redis_config = { 'host': 'localhost', 'port': 6379, @@ -101,8 +124,9 @@ async def test_prepare_activity(_mock_cluster): activities = Activities( temporal_client=temporal_client, - couchbase_config=couchbase_config, + # couchbase_config=couchbase_config, redis_config=redis_config, + mongodb_config=mongo_db_config, logger=logger, notification_handler=notification_handler ) diff --git a/tests/orchestrator/activities/test_couchbase.py b/tests/orchestrator/activities/test_couchbase.py index 84f4da7..91b418a 100644 --- a/tests/orchestrator/activities/test_couchbase.py +++ b/tests/orchestrator/activities/test_couchbase.py @@ -16,6 +16,18 @@ def couchbase(_cluster_mock): ) +def test_shutdown_success(couchbase): + couchbase.shutdown() + couchbase.cluster.close.assert_called_once() + + +def test_shutdown_failure(couchbase): + couchbase.cluster.close.side_effect = Exception("Test error") + couchbase.shutdown() + couchbase.logger.error.assert_called_once_with( + "Failed to close Couchbase connection: %s", couchbase.cluster.close.side_effect) + + @mark.asyncio async def test_load_query_from_couchbase_success(couchbase): couchbase.cluster.query.return_value.rows.return_value = [ diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py new file mode 100644 index 0000000..cb89784 --- /dev/null +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -0,0 +1,246 @@ +from unittest.mock import MagicMock, patch, ANY +from pytest import fixture, mark +from orchestrator.activities.mongo_db import clear_mongo_id +from orchestrator.activities.mongo_db import MongoDB +from sientia_do.notifications.models import NotificationLevel + + +def test_clear_mongo_id(): + input_data = [ + [ + { + "name": "test", + "_id": "12345", + } + ], + { + "name": "test", + "_id": "12345", + "nested": { + "_id": "67890", + "value": [1, 2, 3], + "list": [{"_id": "abcde", "item": "value"}] + }, + "nested_list": [ + {"_id": "fghij", "item": "value1"}, + {"_id": "klmno", "item": "value2"} + ] + } + ] + + output = clear_mongo_id(input_data) + + assert output == [ + [ + {"name": "test"} + ], + { + "name": "test", + "nested": { + "value": [1, 2, 3], + "list": [{"item": "value"}] + }, + "nested_list": [ + {"item": "value1"}, + {"item": "value2"} + ] + }] + + +@fixture +@patch("orchestrator.activities.mongo_db.MongoClient") +def mongo_db(mongo_mock): + return ( + MongoDB( + connection_string="mongodb://localhost:27017", + database_name="test_db", + logger=MagicMock(), + notification_handler=MagicMock() + ) + ) + + +@patch("orchestrator.activities.mongo_db.MongoClient") +def test___init__(mongo_mock): + mongo_db = MongoDB( + connection_string="mongodb://localhost:27017", + database_name="test_db", + logger=MagicMock(), + notification_handler=MagicMock() + ) + assert mongo_db.connection_string == "mongodb://localhost:27017" + assert mongo_db.database_name == "test_db" + mongo_mock.assert_called_once_with( + "mongodb://localhost:27017", serverSelectionTimeoutMS=5000 + ) + mongo_db.client.server_info.assert_called_once() + mongo_db.client.__getitem__.assert_called_once_with("test_db") + + +def test_shutdown_success(mongo_db): + mongo_db.shutdown() + mongo_db.client.close.assert_called_once() + mongo_db.logger.info.assert_any_call("Closing MongoDB connection...") + mongo_db.logger.info.assert_any_call( + "MongoDB connection closed successfully") + + +def test_shutdown_failure(mongo_db): + mongo_db.client.close.side_effect = Exception("Close failed") + mongo_db.shutdown() + mongo_db.logger.error.assert_called_once_with( + "Failed to close MongoDB connection: Close failed" + ) + + +@mark.asyncio +async def test_find_documents_in_mongodb_success(mongo_db): + input_data = {"collection": "test_collection", "filters": { + "name": {"$exists": True} + }} + mock_collection = MagicMock() + mock_collection.find.return_value = [ + {"_id": "12345", "name": "test1"}, + {"_id": "67890", "name": "test2"} + ] + mongo_db.database.__getitem__.return_value = mock_collection + + result = await mongo_db.find_documents_in_mongodb( + { + "query": input_data + }) + + assert len(result) == 2 + assert result[0] == {"name": "test1"} + assert result[1] == {"name": "test2"} + mock_collection.find.assert_called_once_with( + {"name": {"$exists": True}}, {"_id": 0} + ) + + +@mark.asyncio +async def test_find_documents_in_mongodb_failure(mongo_db): + input_data = {"collection": "test_collection", "filters": { + "name": {"$exists": True} + }} + mongo_db.database.__getitem__.return_value = MagicMock( + find=MagicMock(side_effect=Exception("Error")) + ) + + try: + await mongo_db.find_documents_in_mongodb( + { + "query": input_data + }) + except Exception as e: + assert str(e) == "Error" + mongo_db.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id="MONGODB_QUERY_ERROR", + message="Failed to execute MongoDB query: Error", + level=NotificationLevel.ERROR, + block="load_query_from_mongodb", + attachment_content=ANY + ) + + else: + assert False, "Expected an exception to be raised" + + +@mark.asyncio +async def test_find_documents_in_mongodb_missing_collection(mongo_db): + input_data = {"query": {"filters": {}}} + + try: + await mongo_db.find_documents_in_mongodb(input_data) + + except ValueError as e: + assert str(e) == "Collection name must be provided in the query." + + else: + assert False, "Expected a ValueError to be raised" + + +@mark.asyncio +async def test_aggregate_documents_in_mongodb_success(mongo_db): + input_data = {"collection": "test_collection", "aggregation": [ + {"$match": {"name": {"$exists": True}}}, + {"$project": {"name": 1}} + ]} + mock_collection = MagicMock() + mock_collection.aggregate.return_value = [ + {"_id": "asdad", "name": "test1"}, + {"_id": "adzx", "name": "test2"} + ] + mongo_db.database.__getitem__.return_value = mock_collection + + result = await mongo_db.aggregate_documents_in_mongodb( + { + "query": input_data + }) + + assert len(result) == 2 + assert result[0] == {"name": "test1"} + assert result[1] == {"name": "test2"} + expected_pipeline = input_data["aggregation"] + expected_pipeline.append({"$project": {"_id": 0}}) + + mock_collection.aggregate.assert_called_once_with( + expected_pipeline + ) + + +@mark.asyncio +async def test_aggregate_documents_in_mongodb_failure(mongo_db): + input_data = {"collection": "test_collection", "aggregation": [ + {"$match": {"name": {"$exists": True}}}, + {"$project": {"name": 1}} + ]} + mongo_db.database.__getitem__.return_value = MagicMock( + aggregate=MagicMock(side_effect=Exception("Error")) + ) + + try: + await mongo_db.aggregate_documents_in_mongodb( + { + "query": input_data + }) + except Exception as e: + assert str(e) == "Error" + mongo_db.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id="MONGODB_AGGREGATION_ERROR", + message="Failed to execute MongoDB aggregation: Error", + level=NotificationLevel.ERROR, + block="aggregate_documents_in_mongodb", + attachment_content=ANY + ) + + else: + assert False, "Expected an exception to be raised" + + +@mark.asyncio +async def test_aggregate_documents_in_mongodb_missing_collection(mongo_db): + input_data = {"query": {"aggregation": []}} + + try: + await mongo_db.aggregate_documents_in_mongodb(input_data) + + except ValueError as e: + assert str(e) == "Collection name must be provided in the query." + + else: + assert False, "Expected a ValueError to be raised" + + +@mark.asyncio +async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db): + input_data = {"query": {"collection": "test_collection"}} + + try: + await mongo_db.aggregate_documents_in_mongodb(input_data) + + except ValueError as e: + assert str(e) == "Aggregation must be provided." + + else: + assert False, "Expected a ValueError to be raised" diff --git a/tests/orchestrator/utils/test_connectors_config.py b/tests/orchestrator/utils/test_connectors_config.py index 3cf4dac..06b1606 100644 --- a/tests/orchestrator/utils/test_connectors_config.py +++ b/tests/orchestrator/utils/test_connectors_config.py @@ -1,6 +1,7 @@ from os import environ from orchestrator.utils.connectors_config import (build_redis_config, - build_couchbase_config) + build_couchbase_config, + build_mongodb_config) def test_build_redis_config_with_env_vars(): @@ -49,3 +50,27 @@ def test_build_couchbase_config_with_defaults(): 'username': 'sientia', 'password': 'sientia' } + + +def test_build_mongo_db_config_with_env_vars(): + environ['MONGODB_USERNAME'] = 'sientia1' + environ['MONGODB_PASSWORD'] = 'sientia1' + environ['MONGODB_URL'] = 'localhost:27018' + environ['MONGODB_DATABASE_NAME'] = 'test_db' + + assert build_mongodb_config() == { + 'connection_string': 'mongodb://sientia1:sientia1@localhost:27018', + 'database_name': 'test_db' + } + + +def test_build_mongo_db_config_with_defaults(): + environ.pop('MONGODB_USERNAME', None) + environ.pop('MONGODB_PASSWORD', None) + environ.pop('MONGODB_DATABASE_NAME', None) + environ.pop('MONGODB_URL', None) + + assert build_mongodb_config() == { + 'connection_string': 'mongodb://sientia:sientia@localhost:27017', + 'database_name': 'sientia' + } diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index 0b1d4f8..3a131a4 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -36,7 +36,7 @@ async def test_run(workflow_mock, orchestrator): workflow_mock.execute_local_activity_method.assert_has_calls([ call( - Activities.load_query_from_couchbase, + Activities.aggregate_documents_in_mongodb, { "query": input_data["pipelines_query"] }, @@ -47,7 +47,7 @@ async def test_run(workflow_mock, orchestrator): workflow_mock.execute_local_activity_method.assert_has_calls([ call( - Activities.load_query_from_couchbase, + Activities.find_documents_in_mongodb, { "query": input_data["opc_servers_query"] }, diff --git a/values.yaml b/values.yaml index 59e971a..d06746b 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.1.0" + tag: "0.2.0" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "main" + value: "SIENTIAPDE-1107-testar-o-uso-do-mongo-db-no-lugar-do-couchbase" - name: PYTHON_APP value: "orchestrator.worker.worker" @@ -150,6 +150,15 @@ env: - name: COUCHBASE_PASSWORD value: "sientia" + - name: MONGODB_USERNAME + value: "root" + - name: MONGODB_PASSWORD + value: "wKZDbMNU1c" + - name: MONGODB_URL + value: "my-release-mongodb.mongodb.svc.cluster.local:27017" + - name: MONGODB_DATABASE + value: "sientia" + - name: KAFKA_BOOTSTRAP_SERVERS value: "kafka.kafka.svc.cluster.local:9092" @@ -171,7 +180,7 @@ ssh: # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp -# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat +# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.2.0-uat # kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \ # --namespace sientia \