diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index d3c9436..2156009 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -3,11 +3,11 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.postgres import Postgres from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.utils.logger import Logger from laborious.activities.mlflow import MLFlow from laborious.activities.gates import Gates from laborious.activities.opc import OPC from typing import Any - from logging import Logger class Activities(Postgres, MLFlow, Gates, OPC): @@ -44,10 +44,6 @@ class Activities(Postgres, MLFlow, Gates, OPC): 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): Postgres.close(self) OPC.shutdown(self) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index d744578..6147382 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -3,10 +3,10 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): import traceback - from logging import Logger from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity + from sientia_do.temporal.utils.logger import Logger from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter from typing import Any from laborious.utils.filters.conditional_filters import ( @@ -64,8 +64,11 @@ class Gates(BaseActivity): tuple[str | None, int, str]: (policy, confidence, comments) based in priority list and filter configuration and functions. """ + metadata = input_data['metadata'] - self.logger.debug("Performing input gate...") + self.debug("Performing input gate...", metadata) + + self.debug(f"Input data: {input_data}", metadata) filters = input_data['filters'] data = DataFrame(input_data['data']) @@ -73,21 +76,22 @@ class Gates(BaseActivity): filter_output = [] - self.logger.debug(f"Input data:\n {data}") - self.logger.debug(f"Filters: {filters}") + self.debug(f"Input data:\n {data}", metadata) + self.debug(f"Filters: {filters}", metadata) for fil, config in filters.items(): if fil not in input_filter_functions: - self.logger.error(f"Filter {fil} not found") + self.error(f"Filter {fil} not found", metadata) continue try: if input_filter_functions[fil](data, config['config']): - self.logger.debug( - f"Data not passed the input filter {fil}:{config}") + self.debug( + f"Data not passed the input filter {fil}:{config}", metadata) filter_output.append(config['policy']) except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id=f"INTPUT_GATE_ERROR__{fil}", message=f"Error in filter {fil}:{config}: \n {e}", block="input_gate", @@ -97,11 +101,11 @@ class Gates(BaseActivity): for path_flag in path_priority: if path_flag in filter_output: - self.logger.debug(f"Input gate result: {path_flag}") + self.debug(f"Input gate result: {path_flag}", metadata) return path_flag, input_filter_functions['path_confidence'][path_flag], \ "Input data with bad quality" - self.logger.debug("Nothing was filtered by the input gate") + self.debug("Nothing was filtered by the input gate", metadata) return None, 0, "" @activity.defn(name="mlflow_response_gate") @@ -121,7 +125,8 @@ class Gates(BaseActivity): and filter configuration and functions. """ - self.logger.debug("Performing mlflow response gate...") + metadata = input_data['metadata'] + self.debug("Performing mlflow response gate...", metadata) filters = input_data['filters'] data = input_data['data'] @@ -130,8 +135,8 @@ class Gates(BaseActivity): filter_output = [] - self.logger.debug(f"Input data:\n {data}") - self.logger.debug(f"Filters: {filters}") + self.debug(f"Input data:\n {data}", metadata) + self.debug(f"Filters: {filters}", metadata) comments = [] for fil, config in filters.items(): @@ -141,7 +146,8 @@ class Gates(BaseActivity): if mlflow_response_filter_functions[fil](data, config): filter_output.append(config['policy']) comments.append(data['content']['message']) - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}", message=data['content']['message'], block="mlflow_gate", @@ -150,7 +156,8 @@ class Gates(BaseActivity): ) except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id=f"MLFLOW_GATE_RESPONSE_FILTER__{fil}", message=f"Error in filter {fil}:{config}: \n {e}", block="mlflow_gate", @@ -160,11 +167,12 @@ class Gates(BaseActivity): for path_flag in path_priority: if path_flag in filter_output: - self.logger.debug(f"Mlflow response gate result: {path_flag}") + self.debug( + f"Mlflow response gate result: {path_flag}", metadata) return path_flag, mlflow_response_filter_functions['path_confidence'][path_flag], \ ", ".join(comments) - self.logger.debug("Nothing was filtered by the mlflow response gate") + self.debug("Nothing was filtered by the mlflow response gate", metadata) return None, 0, "" @activity.defn(name="mlflow_content_gate") @@ -184,7 +192,8 @@ class Gates(BaseActivity): list and filter configuration and functions. """ - self.logger.debug("Performing mlflow content gate...") + metadata = input_data['metadata'] + self.debug("Performing mlflow content gate...", metadata) filters = input_data['filters'] data = DataFrame(input_data['data']) @@ -193,8 +202,8 @@ class Gates(BaseActivity): filter_output = [] - self.logger.debug(f"Input data:\n {data}") - self.logger.debug(f"Filters: {filters}") + self.debug(f"Input data:\n {data}", metadata) + self.debug(f"Filters: {filters}", metadata) for fil, config in filters.items(): if fil not in mlflow_content_filter_functions: @@ -202,7 +211,8 @@ class Gates(BaseActivity): try: if mlflow_content_filter_functions[fil](data, config): filter_output.append(config['policy']) - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id=f"{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}", message=f"Data not passed the content filter {fil}:{config}", block="mlflow_gate", @@ -211,7 +221,8 @@ class Gates(BaseActivity): ) except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id=f"MLFLOW_GATE_CONTENT_FILTER__{fil}", message=f"Error in filter {fil}:{config}: \n {e}", block="mlflow_gate", @@ -221,11 +232,12 @@ class Gates(BaseActivity): for path_flag in path_priority: if path_flag in filter_output: - self.logger.debug(f"Mlflow content gate result: {path_flag}") + self.debug( + f"Mlflow content gate result: {path_flag}", metadata) return path_flag, mlflow_content_filter_functions['path_confidence'][path_flag], \ "Transformed data not passed the content filter" - self.logger.debug("Nothing was filtered by the mlflow content gate") + self.debug("Nothing was filtered by the mlflow content gate", metadata) return None, 0, "" @activity.defn(name="format_prediction") @@ -241,7 +253,8 @@ class Gates(BaseActivity): Returns: dict: The formatted data. """ - self.logger.debug("Formatting prediction...") + metadata = input_data['metadata'] + self.debug("Formatting prediction...", metadata) data = DataFrame(input_data['data']) data['timestamp'] = input_data['timestamp'] @@ -269,7 +282,8 @@ class Gates(BaseActivity): dict: The formatted data. """ - self.logger.debug("Formatting default prediction...") + metadata = input_data['metadata'] + self.debug("Formatting default prediction...", metadata) return DataFrame({ 'prediction': [0], diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 9b8a58a..4b8c129 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -6,9 +6,9 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.utils.logger import Logger from laborious.utils.repository.model_repository import MLFlowRepository from typing import Any - from logging import Logger class MLFlow(BaseActivity): @@ -36,13 +36,19 @@ class MLFlow(BaseActivity): Returns: dict[str, Any]: The transformed data. """ - self.logger.info('Transforming data...') + metadata = input_data['metadata'] + self.debug('Transforming data...', metadata) data = DataFrame(input_data['data']) model_name = input_data['model_name'] model_retention = input_data['model_retention'] - self.logger.debug("Raw input data:") - self.logger.debug(data) + self.debug("Raw input data:", metadata) + self.debug(data, metadata) + + # Sort by created_at in descending order and keep first occurrence of each variable/timestamp pair + data = data.sort_values('created_at', ascending=False).drop_duplicates( + subset=['variable', 'timestamp'], keep='first' + ) data = data.pivot( index='timestamp', columns='variable', @@ -51,14 +57,14 @@ class MLFlow(BaseActivity): data.reset_index(inplace=True) data.columns.name = None - self.logger.debug("Processed input data:") - self.logger.debug(data) + self.debug("Processed input data:", metadata) + self.debug(data, metadata) response_data = self.model_monitoring_repository.transform( model_name, data, model_retention) - self.logger.debug("Response data:") - self.logger.debug(response_data) + self.debug("Response data:", metadata) + self.debug(response_data, metadata) return response_data @@ -74,18 +80,19 @@ class MLFlow(BaseActivity): Returns: dict[str, Any]: The predicted data. """ - self.logger.info('Predicting data...') + metadata = input_data['metadata'] + self.debug('Predicting data...', metadata) data = DataFrame(input_data['data']) model_name = input_data['model_name'] model_retention = input_data['model_retention'] - self.logger.debug(data) + self.debug(data, metadata) data.replace(np.nan, None, inplace=True) response_data = self.model_monitoring_repository.predict( model_name, data, model_retention) - self.logger.debug(response_data) + self.debug(response_data, metadata) return response_data diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 93c6b7e..6b62e22 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -2,10 +2,10 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): - from logging import Logger from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity + from sientia_do.temporal.utils.logger import Logger from laborious.utils.repository.opc_repository import OpcRepository from typing import Any import traceback @@ -23,9 +23,9 @@ class OPC(BaseActivity): self.opc_servers = opc_servers self.opc_repository = {} - for name, server in opc_servers.items(): - self.opc_repository[name] = OpcRepository( - name=name, + for id, server in opc_servers.items(): + self.opc_repository[id] = OpcRepository( + id=server['id'], url=server['url'], logger=self.logger, server_uri=server['server_uri'], @@ -35,17 +35,17 @@ class OPC(BaseActivity): notification_handler=self.notification_handler, reconnection_interval=server['reconnection_interval'], ) - self.opc_repository[name].connect() + self.opc_repository[id].connect() BaseActivity.__init__(self, logger, notification_handler) - def write_data(self, server: str, tag: str, data: Any, - data_type: str, tag_type: str) -> bool: + def write_data(self, server_id: str, tag: str, data: Any, + data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool: """ Write data to OPC server. Args: - - server (str): The name of the OPC server. + - server_id (str): The id of the OPC server. - tag (str): The tag to write to. - data (Any): The data to write. - data_type (str): The data type. @@ -56,12 +56,12 @@ class OPC(BaseActivity): """ try: - return self.opc_repository[server].write_data( - tag, data, data_type) - self.logger.debug(f"Wrote {tag_type} to {tag}") + return self.opc_repository[server_id].write_data( + tag, data, data_type, self.logger, metadata) except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=metadata, notification_id=f"WRITE_OPC_{tag_type.upper()}_ERROR", message=f"Error writing data to OPC server: {e}", block="write_opc_data", @@ -89,41 +89,44 @@ class OPC(BaseActivity): - dict[Any, Any]: The data that was written to the OPC servers. """ - self.logger.debug("Writing data to OPC servers...") + metadata = input_data['metadata'] + self.debug("Writing data to OPC servers...", metadata) data = DataFrame(input_data['data']) opc_output_config = input_data['opc_output_config'] - self.logger.debug(data) + self.debug(data, metadata) success = True - for server, config in opc_output_config.items(): - if self.opc_repository.get(server) is None: - self.logger.error(f"OPC server {server} not found") + for server_id, config in opc_output_config.items(): + if self.opc_repository.get(server_id) is None: + self.error(f"OPC server {server_id} not found", metadata) continue if 'prediction_tags' in config: for tag, tag_config in config['prediction_tags'].items(): success = success and self.write_data( - server=server, + server_id=server_id, tag=tag, data=data.head(1)['prediction'].values[0], data_type=tag_config['data_type'], - tag_type='prediction' + tag_type='prediction', + metadata=metadata ) if 'confidence_tags' in config: for tag, tag_config in config['confidence_tags'].items(): self.write_data( - server=server, + server_id=server_id, tag=tag, data=data.head(1)['prediction_confidence'].values[0], data_type=tag_config['data_type'], - tag_type='confidence' + tag_type='confidence', + metadata=metadata ) - return self.process_confidence(data, success) + return self.process_confidence(data, success, metadata) - def process_confidence(self, data: DataFrame, success: bool) -> dict[Any, Any]: + def process_confidence(self, data: DataFrame, success: bool, metadata: dict[str, Any]) -> dict[Any, Any]: """ Processes the confidence of OPC server write operations and updates the DataFrame accordingly. @@ -137,15 +140,16 @@ class OPC(BaseActivity): Returns: dict[Any, Any]: The processed data as a dictionary. """ + if not success: data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE - self.logger.debug( - "Some data could not be written to OPC servers, setting confidence to " - f"{OPC_WRITTING_ERROR_CONFIDENCE}." + self.debug( + f"Some data could not be written to OPC servers, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.", + metadata ) else: - self.logger.info("Data written to OPC servers successfully.") + self.debug("Data written to OPC servers successfully.", metadata) return data.to_dict() diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 0515cc2..49cce2b 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -34,8 +34,8 @@ def build_opc_config(): return json.loads(opc_raw) return { - 'opc': { - 'name': getenv('OPC_NAME', 'opc'), + getenv('OPC_ID', '1'): { + 'id': getenv('OPC_ID', '1'), 'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'), 'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), 'cert_path': getenv('OPC_CERT_PATH', None), diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index fcf9d11..f0a768e 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -1,14 +1,14 @@ import traceback -from logging import Logger from datetime import datetime from pathlib import Path from typing import Any from asyncua.sync import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from asyncua.ua import DataValue, Variant, VariantType +from asyncua.ua import DataValue, Variant, VariantType, DateTime from regex import F from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel +from sientia_do.temporal.utils.logger import Logger data_type_map = { 'float': { @@ -35,12 +35,12 @@ data_type_map = { class OpcRepository(): - def __init__(self, name: str, url: str, logger: Logger, + def __init__(self, id: str, url: str, logger: Logger, notification_handler: NotificationHandler, reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None, private_key_path: str = None, server_cert_path: str = None): self.url = url - self.name = name + self.id = id self.server_uri = server_uri self.cert_path = cert_path self.private_key_path = private_key_path @@ -104,7 +104,7 @@ class OpcRepository(): self.client = Client(self.url) if self.cert_path: self.set_security() - self.logger.info('Starting connection...') + self.logger.info(f'Starting connection to OPC server {self.id}...') return self.try_connect() def try_connect(self): @@ -114,6 +114,7 @@ class OpcRepository(): Returns: bool: True if the connection was successful, False otherwise. """ + try: self.last_reconnection_time = datetime.now() self.client.connect() @@ -121,7 +122,7 @@ class OpcRepository(): except Exception as e: trace = traceback.format_exc() self.notification_handler.build_and_send_notification( - notification_id=f"OPC_CONNECTION_ERROR_{self.name}", + notification_id=f"OPC_CONNECTION_ERROR_{self.id}", message=f"Failed to connect to OPC server: {e}", block="opc_repository", level=NotificationLevel.ERROR, @@ -136,9 +137,12 @@ class OpcRepository(): """ if self.client is None: return - self.client.disconnect() + try: + self.client.disconnect() + self.logger.info('Disconnected from OPC server') + except Exception as e: + self.logger.error(f"Failed to disconnect from OPC server: {e}") self.client = None - self.logger.info('Disconnected from OPC server') def __del__(self): """ @@ -165,7 +169,7 @@ class OpcRepository(): if self.error_count > 5: self.logger.warning( - f"OPC server {self.name} will be disconnected due to multiple errors") + f"OPC server {self.id} will be disconnected due to multiple errors") try: self.disconnect() except Exception as e: @@ -173,7 +177,7 @@ class OpcRepository(): self.logger.error(f"Failed to disconnect from OPC server: {e}") self.logger.error(trace) self.logger.info( - f"Attempting to reconnect to OPC server {self.name}...") + f"Attempting to reconnect to OPC server {self.id}...") return self.connect() if hasattr(self.client, 'aio_obj') and self.client.aio_obj.uaclient.protocol is None or \ @@ -181,18 +185,20 @@ class OpcRepository(): self.client.aio_obj.uaclient.protocol.state == "closed"): self.logger.error( - f"OPC server {self.name} is not connected") - if (datetime.now() - self.last_reconnection_time).total_seconds( + f"OPC server {self.id} is not connected") + if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds( ) > self.reconnection_interval: + self.disconnect() self.logger.error( - f"Trying to reconnect to OPC server {self.name}...") - return self.try_connect() + f"Trying to reconnect to OPC server {self.id}...") + return self.connect() return False return True - def write_data(self, node: str, value: Any, data_type: str) -> bool: + def write_data(self, node: str, value: Any, data_type: str, + logger: Logger, metadata: dict[str, Any]) -> bool: """ Writes data to the OPC server. If the connection is not established, it attempts to reconnect. @@ -210,42 +216,54 @@ class OpcRepository(): except Exception as e: trace = traceback.format_exc() self.notification_handler.build_and_send_notification( - notification_id=f"OPC_WRITE_GET_NODE_ERROR_{self.name}", - message=f"Failed to get node from OPC server: {e}", + notification_id=f"OPC_WRITE_GET_NODE_ERROR_{self.id}", + message=f"Failed to get node from OPC server: {e} | metadata: {metadata}", block="opc_repository", level=NotificationLevel.ERROR, attachment_content=trace ) - self.logger.error(trace) + logger.custom_error(trace, metadata.get('schedule_name', 'N/A')) self.error_count += 1 return False if data_type not in data_type_map: self.notification_handler.build_and_send_notification( - notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.name}", - message=f"Unsupported data type: {data_type}", + notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}", + message=f"Unsupported data type: {data_type} | metadata: {metadata}", block="opc_repository", level=NotificationLevel.ERROR ) return False data = data_type_map[data_type]['converter'](value) - self.logger.info(f'Writing {data} - {type(data)} to {node}') + logger.custom_info( + f'Writing {data} - {type(data)} to {node}', metadata.get('schedule_name', 'N/A')) + now = datetime.now() ua_data = DataValue( - Variant(data, data_type_map[data_type]['opc_type'])) + Variant(data, data_type_map[data_type]['opc_type']), + SourceTimestamp=DateTime( + now.year, + now.month, + now.day, + now.hour, + now.minute, + now.second, + now.microsecond + ) + ) try: node.write_value(ua_data) except Exception as e: trace = traceback.format_exc() self.notification_handler.build_and_send_notification( - notification_id=f"OPC_WRITE_DATA_ERROR_{self.name}", - message=f"Failed to write data to OPC server: {e}", + notification_id=f"OPC_WRITE_DATA_ERROR_{self.id}", + message=f"Failed to write data to OPC server: {e} | metadata: {metadata}", block="opc_repository", level=NotificationLevel.ERROR, attachment_content=trace ) - self.logger.error(trace) + logger.custom_error(trace, metadata.get('schedule_name', 'N/A')) self.error_count += 1 return False self.error_count = 0 diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index ae5e47c..e847108 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -60,8 +60,6 @@ async def main(): workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction], activities=[ - # Base - activities.prepare_activity, # MLFlow activities.request_predict, activities.request_transform, @@ -93,7 +91,7 @@ async def main(): # If an exception occurs in any of the worker handlers, it will be propagated here. await asyncio.gather(*handlers) except BaseException as e: - logger.error("An unhandled exception occurred: %s", e, exc_info=True) + logger.error(f"An unhandled exception occurred: {e}") finally: if notification_handler: notification_handler.shutdown() diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index d80a9d4..9d3bcd6 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -40,27 +40,28 @@ class PredictionsBatch(): Exception: If any of the required parameters are missing or if the workflow fails. """ - await workflow.execute_local_activity_method( - Activities.prepare_activity, - { + metadata = { + 'metadata': { 'schedule_name': input_data['schedule_name'], 'model_name': input_data['model_name'], 'model_id': input_data['model_id'], 'workflow_name': 'predictions_batch' - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=60) - ) + } + } data = await workflow.execute_local_activity_method( Activities.load_custom_query, - input_data['query'], + { + **metadata, + 'query': input_data['query'], + }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) ) # Prepare input for prediction_process workflow prediction_input = { + 'metadata': metadata, 'data': data, 'schema': input_data['schema'], 'table_name': input_data['table_name'], diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index a33f71f..b021bfd 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -36,6 +36,7 @@ class FormatAndExportPrediction(): Returns: bool: True if the workflow was successful, False otherwise. """ + metadata = input_data['metadata'] path_flag = input_data['path_flag'] data = input_data['data'] prediction_confidence = input_data['prediction_confidence'] @@ -45,6 +46,7 @@ class FormatAndExportPrediction(): prediction = await workflow.execute_local_activity_method( Activities.format_prediction, { + **metadata, 'data': data, 'timestamp': input_data['timestamp'], 'model_id': input_data['model_id'], @@ -59,6 +61,7 @@ class FormatAndExportPrediction(): prediction = await workflow.execute_local_activity_method( Activities.format_default_prediction, { + **metadata, 'timestamp': input_data['timestamp'], 'model_id': input_data['model_id'], 'prediction_confidence': prediction_confidence, @@ -72,6 +75,7 @@ class FormatAndExportPrediction(): prediction = await workflow.execute_activity_method( Activities.write_opc_data, { + **metadata, 'opc_output_config': input_data['opc_output_config'], 'data': prediction }, @@ -83,6 +87,7 @@ class FormatAndExportPrediction(): await workflow.execute_activity_method( Activities.export_data_to_postgres, { + **metadata, 'schema': input_data['schema'], 'table_name': input_data['table_name'], 'data': prediction, diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index e3b38d8..10f820c 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -41,6 +41,7 @@ class PredictionProcess(): Exception: If any of the required parameters are missing or if the workflow fails. """ + metadata = input_data['metadata'] data = input_data['data'] model_id = input_data['model_id'] model_name = input_data['model_name'] @@ -49,19 +50,26 @@ class PredictionProcess(): last_timestamp = await workflow.execute_local_activity_method( Activities.get_last_timestamp, { + **metadata, 'data': data }, retry_policy=retry_policy, start_to_close_timeout=timedelta(minutes=1), ) + gate_input = { + **metadata, + 'filters': input_data['input_filters'], + 'data': data, + 'path_priority': input_data['path_priority'] + } + + print(f"Metadata e input atualizadas {gate_input}") + print(f"Metadata: {metadata}") + path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.input_gate, - { - 'filters': input_data['input_filters'], - 'data': data, - 'path_priority': input_data['path_priority'] - }, + gate_input, retry_policy=retry_policy, start_to_close_timeout=timedelta(minutes=1), ) @@ -74,6 +82,7 @@ class PredictionProcess(): response_data = await workflow.execute_local_activity_method( Activities.request_transform, { + **metadata, 'data': data, 'model_name': model_name, 'model_retention': model_retention @@ -85,6 +94,7 @@ class PredictionProcess(): path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_response_gate, { + **metadata, 'filters': input_data['mlflow_transform_filters'], 'data': response_data, 'type': 'transform', @@ -104,6 +114,7 @@ class PredictionProcess(): path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_content_gate, { + **metadata, 'filters': input_data['mlflow_transform_filters'], 'data': transformed_data, 'type': 'transform', @@ -121,6 +132,7 @@ class PredictionProcess(): response_data = await workflow.execute_local_activity_method( Activities.request_predict, { + **metadata, 'data': transformed_data, 'model_name': model_name, 'model_retention': model_retention @@ -132,6 +144,7 @@ class PredictionProcess(): path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_response_gate, { + **metadata, 'filters': input_data['mlflow_predict_filters'], 'data': response_data, 'type': 'predict', @@ -149,6 +162,7 @@ class PredictionProcess(): await workflow.execute_child_workflow( 'format_and_export_prediction', { + 'metadata': metadata, 'path_flag': path_flag, 'data': response_data['content'], 'prediction_confidence': confidence, @@ -187,13 +201,15 @@ class PredictionProcess(): bool: True if the prediction should be stopped, False otherwise. """ + metadata = input_data['metadata'] + schema = input_data['schema'] table_name = input_data['table_name'] model_id = input_data['model_id'] model_name = input_data['model_name'] model_retention = input_data['model_retention'] - path_flag = path_flag.upper() if path_flag else None + path_flag = path_flag.upper() if path_flag else '' if path_flag == 'STOP': return True @@ -203,6 +219,7 @@ class PredictionProcess(): await workflow.execute_activity_method( Activities.repeat_last_prediction, { + **metadata, 'schema': schema, 'table_name': table_name, 'model': model_id, @@ -218,6 +235,7 @@ class PredictionProcess(): await workflow.execute_child_workflow( 'format_and_export_prediction', { + 'metadata': metadata, 'path_flag': path_flag, 'data': data, 'prediction_confidence': confidence, diff --git a/requirements.txt b/requirements.txt index 599917b..5dfca57 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,5 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.17 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.1 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1 diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index 31f83e8..e482ca8 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -90,65 +90,6 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre ) -@mark.asyncio -@patch('laborious.activities.activities.Postgres.__init__') -@patch('laborious.activities.activities.MLFlow.__init__') -@patch('laborious.activities.activities.OPC.__init__') -async def test_prepare_activity(_mock_opc_init, - _mock_mlflow_init, _mock_postgres_init): - postgres_config = { - 'host': 'localhost', - 'port': 5432, - 'user': 'postgres', - 'password': 'postgres', - 'dbname': 'postgres', - 'min_connections': 1, - 'max_connections': 10 - } - - mlflow_config = { - 'host': 'localhost', - 'port': 5000, - 'username': 'mlflow', - 'password': 'mlflow' - } - - opc_config = { - 'bootstrap_servers': 'localhost:9092', - 'polling_time': 1000, - 'group_id': 'test-group' - } - - logger = MagicMock() - notification_handler = MagicMock() - - activities = Activities( - postgres_config=postgres_config, - mlflow_config=mlflow_config, - opc_config=opc_config, - logger=logger, - notification_handler=notification_handler - ) - - input_data = { - 'workflow_name': 'test-workflow-name', - 'schedule_name': 'test-schedule-name', - 'model_name': 'test-model-name', - 'model_id': 'test-model-id' - } - - await activities.prepare_activity(input_data) - - assert activities.notification_handler.base_notification.pipeline == input_data[ - 'workflow_name'] - assert activities.notification_handler.base_notification.trigger == input_data[ - 'schedule_name'] - assert activities.notification_handler.base_notification.model_name == input_data[ - 'model_name'] - assert activities.notification_handler.base_notification.model_id == input_data[ - 'model_id'] - - @patch('laborious.activities.activities.Postgres', return_value=MagicMock()) @patch('laborious.activities.activities.MLFlow', return_value=MagicMock()) @patch('laborious.activities.activities.OPC', return_value=MagicMock()) diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index 28ec1d2..19d52cd 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -6,16 +6,34 @@ from laborious.activities.gates import Gates @fixture def gates_activity(): - return Gates( + gates = Gates( logger=MagicMock(), notification_handler=MagicMock(), ) + gates.error = MagicMock() + gates.debug = MagicMock() + gates.info = MagicMock() + gates.warning = MagicMock() + gates.critical = MagicMock() + gates.send_notification = MagicMock() + return gates + + +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} @mark.asyncio async def test_input_gate_invalid_filter(gates_activity): # Arrange input_data = { + **metadata, 'filters': { 'INVALID_FILTER': {'POLICY': 'STOP'} }, @@ -28,8 +46,8 @@ async def test_input_gate_invalid_filter(gates_activity): # Assert assert result == (None, 0, "") - gates_activity.logger.error.assert_called_once_with( - "Filter INVALID_FILTER not found" + gates_activity.error.assert_called_once_with( + "Filter INVALID_FILTER not found", metadata['metadata'] ) @@ -41,6 +59,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac mock_input_filter_functions.__getitem__.return_value = MagicMock( side_effect=Exception("Test error")) input_data = { + **metadata, 'filters': { 'EMPTY_DATA': {'policy': 'STOP', 'config': {}} }, @@ -53,7 +72,8 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac # Assert assert result == (None, 0, "") - gates_activity.notification_handler.build_and_send_notification.assert_called_once_with( + gates_activity.send_notification.assert_called_once_with( + metadata=metadata['metadata'], notification_id="INTPUT_GATE_ERROR__EMPTY_DATA", message="Error in filter EMPTY_DATA:{'policy': 'STOP', 'config': {}}: \n Test error", block="input_gate", @@ -66,6 +86,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac async def test_input_gate_no_filters(gates_activity): # Arrange input_data = { + **metadata, 'filters': {}, 'data': {'value': [1, 2, 3]}, 'path_priority': ['CONTINUE', 'STOP', 'REPEAT'] @@ -76,13 +97,14 @@ async def test_input_gate_no_filters(gates_activity): # Assert assert result == (None, 0, "") - gates_activity.logger.debug.assert_called() + gates_activity.debug.assert_called() @mark.asyncio async def test_input_gate_with_filter(gates_activity): # Arrange input_data = { + **metadata, 'filters': { 'EMPTY_DATA': {'policy': 'STOP', 'config': {}} }, @@ -95,13 +117,14 @@ async def test_input_gate_with_filter(gates_activity): # Assert assert result == ('STOP', -1, "Input data with bad quality") - gates_activity.logger.debug.assert_called() + gates_activity.debug.assert_called() @mark.asyncio async def test_mlflow_response_gate_invalid_filter(gates_activity): # Arrange input_data = { + **metadata, 'filters': { 'INVALID_FILTER': {'POLICY': 'STOP'} }, @@ -126,6 +149,7 @@ async def test_mlflow_response_gate_filter_exception(mock_mlflow_response_filter mock_mlflow_response_filter_functions.__getitem__.return_value = MagicMock( side_effect=Exception("Test error")) input_data = { + **metadata, 'filters': { 'INVALID_FILTER': {'POLICY': 'STOP'} }, @@ -139,7 +163,8 @@ async def test_mlflow_response_gate_filter_exception(mock_mlflow_response_filter # Assert assert result == (None, 0, "") - gates_activity.notification_handler.build_and_send_notification.assert_called_once_with( + gates_activity.send_notification.assert_called_once_with( + metadata=metadata['metadata'], notification_id="MLFLOW_GATE_RESPONSE_FILTER__INVALID_FILTER", message="Error in filter INVALID_FILTER:{'POLICY': 'STOP'}: \n Test error", block="mlflow_gate", @@ -152,6 +177,7 @@ async def test_mlflow_response_gate_filter_exception(mock_mlflow_response_filter async def test_mlflow_response_gate_no_filters(gates_activity): # Arrange input_data = { + **metadata, 'filters': {}, 'data': {'content': {'message': 'success'}}, 'type': 'test', @@ -163,13 +189,14 @@ async def test_mlflow_response_gate_no_filters(gates_activity): # Assert assert result == (None, 0, "") - gates_activity.logger.debug.assert_called() + gates_activity.debug.assert_called() @mark.asyncio async def test_mlflow_response_gate_with_filter(gates_activity): # Arrange input_data = { + **metadata, 'filters': { 'API_ERROR': {'policy': 'STOP'} }, @@ -189,14 +216,15 @@ async def test_mlflow_response_gate_with_filter(gates_activity): # Assert assert result == ('STOP', -1, "API error occurred") - gates_activity.logger.debug.assert_called() - gates_activity.notification_handler.build_and_send_notification.assert_called() + gates_activity.debug.assert_called() + gates_activity.send_notification.assert_called() @mark.asyncio async def test_mlflow_content_gate_invalid_filter(gates_activity): # Arrange input_data = { + **metadata, 'filters': { 'INVALID_FILTER': {'POLICY': 'STOP'} }, @@ -221,6 +249,7 @@ async def test_mlflow_content_gate_filter_exception(mock_mlflow_content_filter_f mock_mlflow_content_filter_functions.__getitem__.return_value = MagicMock( side_effect=Exception("Test error")) input_data = { + **metadata, 'filters': { 'API_ERROR': {'POLICY': 'STOP'} }, @@ -240,8 +269,9 @@ async def test_mlflow_content_gate_filter_exception(mock_mlflow_content_filter_f # Assert assert result == (None, 0, "") - gates_activity.logger.debug.assert_called() - gates_activity.notification_handler.build_and_send_notification.assert_called_once_with( + gates_activity.debug.assert_called() + gates_activity.send_notification.assert_called_once_with( + metadata=metadata['metadata'], notification_id="MLFLOW_GATE_CONTENT_FILTER__API_ERROR", message="Error in filter API_ERROR:{'POLICY': 'STOP'}: \n Test error", block="mlflow_gate", @@ -254,6 +284,7 @@ async def test_mlflow_content_gate_filter_exception(mock_mlflow_content_filter_f async def test_mlflow_content_gate_no_filters(gates_activity): # Arrange input_data = { + **metadata, 'filters': {}, 'data': {'value': [1, 2, 3]}, 'type': 'test', @@ -265,13 +296,14 @@ async def test_mlflow_content_gate_no_filters(gates_activity): # Assert assert result == (None, 0, "") - gates_activity.logger.debug.assert_called() + gates_activity.debug.assert_called() @mark.asyncio async def test_mlflow_content_gate_with_filter(gates_activity): # Arrange input_data = { + **metadata, 'filters': { 'NAN_VALUES': {'policy': 'STOP', 'config': {}} }, @@ -286,14 +318,15 @@ async def test_mlflow_content_gate_with_filter(gates_activity): # Assert assert result == ( 'STOP', -1, "Transformed data not passed the content filter") - gates_activity.logger.debug.assert_called() - gates_activity.notification_handler.build_and_send_notification.assert_called() + gates_activity.debug.assert_called() + gates_activity.send_notification.assert_called() @mark.asyncio async def test_format_prediction(gates_activity): # Arrange input_data = { + **metadata, 'data': {'prediction': [1], 'response_time': [0.1]}, 'timestamp': '2023-05-26 11:12:27', 'model_id': 'test_model', @@ -311,13 +344,14 @@ async def test_format_prediction(gates_activity): assert result['prediction_confidence'] == {0: 0.9} assert result['prediction_status'] == {0: 'Good'} assert result['comments'] == {0: ""} - gates_activity.logger.debug.assert_called() + gates_activity.debug.assert_called() @mark.asyncio async def test_format_default_prediction(gates_activity): # Arrange input_data = { + **metadata, 'timestamp': '2023-05-26 11:12:27', 'model_id': 'test_model', 'prediction_confidence': 0.1, @@ -335,13 +369,14 @@ async def test_format_default_prediction(gates_activity): assert result['prediction_confidence'] == {0: 0.1} assert result['prediction_status'] == {0: 'Bad'} assert result['comments'] == {0: 'Test comment'} - gates_activity.logger.debug.assert_called() + gates_activity.debug.assert_called() @mark.asyncio async def test_get_last_timestamp_with_data(gates_activity): # Arrange input_data = { + **metadata, 'data': { 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'] } diff --git a/tests/laborious/activities/test_mlflow.py b/tests/laborious/activities/test_mlflow.py index 5834cb4..a92f3f3 100644 --- a/tests/laborious/activities/test_mlflow.py +++ b/tests/laborious/activities/test_mlflow.py @@ -39,6 +39,16 @@ def mlflow(mock_mlflow_repository): ) +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + + @mark.asyncio @patch("laborious.activities.mlflow.DataFrame") @patch("laborious.activities.mlflow.max") @@ -46,11 +56,20 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow): mock_max.return_value = '2024-01-02' # Mock input data input_data = { + **metadata, 'data': [ - {'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0}, - {'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0}, - {'timestamp': '2024-01-02', 'variable': 'var1', 'value': 3.0}, - {'timestamp': '2024-01-02', 'variable': 'var2', 'value': 4.0} + {'timestamp': '2024-01-01', 'variable': 'var1', + 'value': 1.0, 'created_at': '2024-01-01 12:00:00'}, + {'timestamp': '2024-01-01', 'variable': 'var2', + 'value': 2.0, 'created_at': '2024-01-01 12:00:00'}, + {'timestamp': '2024-01-02', 'variable': 'var1', + 'value': 3.0, 'created_at': '2024-01-02 12:00:00'}, + {'timestamp': '2024-01-02', 'variable': 'var2', + 'value': 4.0, 'created_at': '2024-01-02 12:00:00'}, + {'timestamp': '2024-01-02', 'variable': 'var1', + 'value': 1.0, 'created_at': '2024-01-01 12:00:00'}, + {'timestamp': '2024-01-02', 'variable': 'var2', + 'value': 1.0, 'created_at': '2024-01-01 12:00:00'} ], 'model_name': 'test_model', 'model_retention': 30 @@ -60,6 +79,9 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow): expected_response = {'prediction': [0.5, 0.6]} mlflow.model_monitoring_repository.transform.return_value = expected_response + mock_dataframe.return_value.sort_values.return_value = mock_dataframe.return_value + mock_dataframe.return_value.drop_duplicates.return_value = mock_dataframe.return_value + # Call the method response_data = await mlflow.request_transform(input_data) @@ -89,6 +111,7 @@ async def test_request_predict(mock_max, mock_dataframe, mlflow): mock_max.return_value = '2024-01-02' # Mock input data input_data = { + **metadata, 'data': [ {'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0}, {'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0}, diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index bbdc626..ebce052 100644 --- a/tests/laborious/activities/test_opc.py +++ b/tests/laborious/activities/test_opc.py @@ -5,6 +5,15 @@ from laborious.activities.opc import NotificationLevel from laborious.activities.opc import OPC +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @patch("laborious.activities.opc.OpcRepository") def test___init__(mock_opc_repository): @@ -15,6 +24,7 @@ def test___init__(mock_opc_repository): mock_notification_handler = MagicMock() servers = { 'server1': { + 'id': 'server1', 'url': 'http://localhost:8080', 'server_uri': 'opc.tcp://localhost:4840', 'cert_path': '', @@ -23,6 +33,7 @@ def test___init__(mock_opc_repository): 'reconnection_interval': 60, }, 'server2': { + 'id': 'server2', 'url': 'http://localhost:8080', 'server_uri': 'opc.tcp://localhost:4840', 'cert_path': '', @@ -45,7 +56,7 @@ def test___init__(mock_opc_repository): mock_opc_repository.assert_has_calls([ call( - name="server1", + id="server1", url="http://localhost:8080", logger=mock_logger, server_uri="opc.tcp://localhost:4840", @@ -58,7 +69,7 @@ def test___init__(mock_opc_repository): ]) mock_opc_repository.assert_has_calls([ call( - name="server2", + id="server2", url="http://localhost:8080", logger=mock_logger, server_uri="opc.tcp://localhost:4840", @@ -79,6 +90,7 @@ def test___init__(mock_opc_repository): def opc(mock_opc_repository): servers = { 'server1': { + 'id': 'server1', 'url': 'http://localhost:8080', 'server_uri': 'opc.tcp://localhost:4840', 'cert_path': '', @@ -90,12 +102,15 @@ def opc(mock_opc_repository): mock_opc_repository.write_data = MagicMock( return_value=True ) - return OPC( + opc = OPC( opc_servers=servers, logger=MagicMock(), notification_handler=MagicMock() ) + opc.send_notification = MagicMock() + return opc + WRITE_DATA_CASES = [ ('tag1', 'int', 50), @@ -107,10 +122,10 @@ WRITE_DATA_CASES = [ @mark.parametrize('tag,data_type,data', WRITE_DATA_CASES) def test_write_data_success(opc, tag, data_type, data): - assert opc.write_data(server='server1', tag=tag, data=data, - data_type=data_type, tag_type='prediction') + assert opc.write_data(server_id='server1', tag=tag, data=data, + data_type=data_type, tag_type='prediction', metadata=metadata) opc.opc_repository['server1'].write_data.assert_called_once_with( - tag, data, data_type) + tag, data, data_type, opc.logger, metadata) def test_write_data_exception(opc): @@ -118,11 +133,12 @@ def test_write_data_exception(opc): "Test error") try: - opc.write_data(server='server1', tag='tag1', data=50, - data_type='int', tag_type='prediction') + opc.write_data(server_id='server1', tag='tag1', data=50, + data_type='int', tag_type='prediction', metadata=metadata) except Exception: - opc.notification_handler.build_and_send_notification.assert_called_once_with( + opc.send_notification.assert_called_once_with( + metadata=metadata, notification_id="WRITE_OPC_PREDICTION_ERROR", message="Error writing data to OPC server: Test error", block="write_opc_data", @@ -138,6 +154,7 @@ def test_write_data_exception(opc): async def test_write_opc_data_success(opc): # Arrange input_data = { + **metadata, 'data': { 'prediction': [0.75], 'prediction_confidence': [0.95] @@ -163,19 +180,21 @@ async def test_write_opc_data_success(opc): assert output == {'data': 'data'} opc.write_data.assert_has_calls([ call( - server='server1', + server_id='server1', tag='tag1', data=0.75, data_type='float', - tag_type='prediction' + tag_type='prediction', + metadata=metadata['metadata'] )]) opc.write_data.assert_has_calls([ call( - server='server1', + server_id='server1', tag='tag2', data=0.95, data_type='float', - tag_type='confidence' + tag_type='confidence', + metadata=metadata['metadata'] ) ]) assert opc.write_data.call_count == 2 @@ -185,6 +204,7 @@ async def test_write_opc_data_success(opc): async def test_write_opc_data_empty_config(opc): # Arrange input_data = { + **metadata, 'data': { 'prediction': [0.75], 'prediction_confidence': [0.95] @@ -209,7 +229,7 @@ async def test_write_opc_data_empty_config(opc): ]) def test_process_confidence(opc, data, success, expected): # Act - result = opc.process_confidence(data, success) + result = opc.process_confidence(data, success, metadata) # Assert assert result['prediction_confidence'][0] == expected diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 7369496..1b661d2 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -14,7 +14,7 @@ def mock_logger(): @fixture def opc_repository(mock_logger): return OpcRepository( - name="test_repo", + id="test_repo", url="opc.tcp://localhost:4840", logger=mock_logger, notification_handler=Mock(), @@ -34,8 +34,18 @@ def mock_client(): yield client_instance +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + + def test_init(opc_repository): - assert opc_repository.name == "test_repo" + assert opc_repository.id == "test_repo" assert opc_repository.url == "opc.tcp://localhost:4840" assert opc_repository.server_uri == "urn:test:server" assert opc_repository.cert_path == "/path/to/cert.pem" @@ -110,7 +120,7 @@ def test_try_connect_fail(opc_repository): opc_repository.client.connect.assert_called_once() assert opc_repository.last_reconnection_time is not None opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( - notification_id=f"OPC_CONNECTION_ERROR_{opc_repository.name}", + notification_id=f"OPC_CONNECTION_ERROR_{opc_repository.id}", message="Failed to connect to OPC server: Test error", block="opc_repository", level=NotificationLevel.ERROR, @@ -126,6 +136,17 @@ def test_disconnect(opc_repository, mock_client): assert opc_repository.client is None +def test_disconnect_error(opc_repository, mock_client): + opc_repository.client = mock_client + mock_client.disconnect.side_effect = Exception("Test error") + opc_repository.disconnect() + + opc_repository.logger.error.assert_called_once_with( + "Failed to disconnect from OPC server: Test error" + ) + assert opc_repository.client is None + + def test_validate_connection_none_client(opc_repository): opc_repository.client = None opc_repository.connect = MagicMock() @@ -174,11 +195,11 @@ def test_validate_connection_lost_time_to_reconect(_mock_datetime, opc_repositor opc_repository.client = MagicMock() opc_repository.client.aio_obj.uaclient.protocol = None opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.try_connect = MagicMock() + opc_repository.connect = MagicMock() response = opc_repository.validate_connection() - opc_repository.try_connect.assert_called_once() - assert response == opc_repository.try_connect.return_value + opc_repository.connect.assert_called_once() + assert response == opc_repository.connect.return_value def test_validate_connection_failed(opc_repository): @@ -192,7 +213,8 @@ def test_validate_connection_failed(opc_repository): def test_write_data_validate_connection_do_nothing(opc_repository): opc_repository.validate_connection = MagicMock(return_value=True) opc_repository.client = MagicMock() - opc_repository.write_data("ns=2;s=TestNode", 42.0, "float") + opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata) opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") @@ -201,7 +223,8 @@ def test_write_data_validate_connection_failed(opc_repository): opc_repository.validate_connection = MagicMock(return_value=False) opc_repository.client = MagicMock() opc_repository.error_count = 0 - opc_repository.write_data("ns=2;s=TestNode", 42.0, "float") + opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata) opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_not_called() @@ -211,12 +234,13 @@ def test_write_data_get_node_failed(opc_repository): opc_repository.client = MagicMock() opc_repository.error_count = 0 opc_repository.client.get_node.side_effect = Exception("Test error") - opc_repository.write_data("ns=2;s=TestNode", 42.0, "float") + opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata) opc_repository.validate_connection.assert_called_once() opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( - notification_id=f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.name}", - message="Failed to get node from OPC server: Test error", + notification_id=f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}", + message="Failed to get node from OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}", block="opc_repository", level=NotificationLevel.ERROR, attachment_content=ANY @@ -230,13 +254,14 @@ def test_write_data_invalid_data_type(opc_repository, mock_client): mock_node = MagicMock() mock_client.get_node.return_value = mock_node - opc_repository.write_data("ns=2;s=TestNode", 42.0, "invalid_type") + opc_repository.write_data("ns=2;s=TestNode", 42.0, + "invalid_type", opc_repository.logger, metadata) opc_repository.validate_connection.assert_called_once() mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( - notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.name}", - message="Unsupported data type: invalid_type", + notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}", + message="Unsupported data type: invalid_type | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}", block="opc_repository", level=NotificationLevel.ERROR ) @@ -248,12 +273,11 @@ def test_write_data(opc_repository, mock_client): mock_node = MagicMock() mock_client.get_node.return_value = mock_node - opc_repository.write_data("ns=2;s=TestNode", 42.0, "float") + opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata) mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") mock_node.write_value.assert_called_once() - opc_repository.logger.info.assert_called_once_with( - "Writing 42.0 - to " + str(mock_node)) def test_write_data_write_value_failed(opc_repository, mock_client): @@ -263,13 +287,14 @@ def test_write_data_write_value_failed(opc_repository, mock_client): opc_repository.error_count = 0 mock_client.get_node.return_value = mock_node mock_node.write_value.side_effect = Exception("Test error") - opc_repository.write_data("ns=2;s=TestNode", 42.0, "float") + opc_repository.write_data("ns=2;s=TestNode", 42.0, + "float", opc_repository.logger, metadata) opc_repository.validate_connection.assert_called_once() mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") mock_node.write_value.assert_called_once() opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( - notification_id=f"OPC_WRITE_DATA_ERROR_{opc_repository.name}", - message="Failed to write data to OPC server: Test error", + notification_id=f"OPC_WRITE_DATA_ERROR_{opc_repository.id}", + message="Failed to write data to OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}", block="opc_repository", level=NotificationLevel.ERROR, attachment_content=ANY diff --git a/tests/laborious/utils/test_connectors_config.py b/tests/laborious/utils/test_connectors_config.py index b137bc2..a7b36aa 100644 --- a/tests/laborious/utils/test_connectors_config.py +++ b/tests/laborious/utils/test_connectors_config.py @@ -54,7 +54,7 @@ def test_build_opc_config_with_env_vars(): def test_build_opc_config_with_individual_env_vars(): # Arrange environ.pop('OPC_CONFIG', None) - environ['OPC_NAME'] = 'test-name' + environ['OPC_ID'] = '1' environ['OPC_URL'] = 'opc.tcp://test:4840' environ['OPC_SERVER_URI'] = 'opc.tcp://test:4840' environ['OPC_RECONNECTION_INTERVAL'] = '300' @@ -63,16 +63,16 @@ def test_build_opc_config_with_individual_env_vars(): config = build_opc_config() # Assert - assert config['opc']['name'] == 'test-name' - assert config['opc']['url'] == 'opc.tcp://test:4840' - assert config['opc']['server_uri'] == 'opc.tcp://test:4840' - assert config['opc']['reconnection_interval'] == 300 + assert config['1']['id'] == '1' + assert config['1']['url'] == 'opc.tcp://test:4840' + assert config['1']['server_uri'] == 'opc.tcp://test:4840' + assert config['1']['reconnection_interval'] == 300 def test_build_opc_config_with_defaults(): # Arrange environ.pop('OPC_CONFIG', None) - environ.pop('OPC_NAME', None) + environ.pop('OPC_ID', None) environ.pop('OPC_URL', None) environ.pop('OPC_SERVER_URI', None) environ.pop('OPC_RECONNECTION_INTERVAL', None) @@ -81,10 +81,10 @@ def test_build_opc_config_with_defaults(): config = build_opc_config() # Assert - assert config['opc']['name'] == 'opc' - assert config['opc']['url'] == 'opc.tcp://localhost:4840' - assert config['opc']['server_uri'] == 'opc.tcp://localhost:4840' - assert config['opc']['reconnection_interval'] == 120 + assert config['1']['id'] == '1' + assert config['1']['url'] == 'opc.tcp://localhost:4840' + assert config['1']['server_uri'] == 'opc.tcp://localhost:4840' + assert config['1']['reconnection_interval'] == 120 def test_build_postgres_config_with_env_vars(): diff --git a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py index 85cbd34..7dbf39a 100644 --- a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py +++ b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py @@ -10,11 +10,22 @@ def format_and_export_prediction(): return FormatAndExportPrediction() +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + + @mark.asyncio @patch("laborious.workflows.sub_workflows.format_and_export_prediction.workflow", new_callable=AsyncMock) async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): input_data = { + 'metadata': metadata, "path_flag": None, "data": {"test": "data"}, "timestamp": "2021-01-01", @@ -35,7 +46,8 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): 'data': input_data['data'], 'timestamp': input_data['timestamp'], 'model_id': input_data['model_id'], - 'prediction_confidence': input_data['prediction_confidence'] + 'prediction_confidence': input_data['prediction_confidence'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -46,7 +58,8 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): Activities.write_opc_data, { 'opc_output_config': input_data['opc_output_config'], - 'data': workflow_mock.execute_local_activity_method.return_value + 'data': workflow_mock.execute_local_activity_method.return_value, + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -59,7 +72,8 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): { 'schema': input_data['schema'], 'table_name': input_data['table_name'], - 'data': workflow_mock.execute_activity_method.return_value + 'data': workflow_mock.execute_activity_method.return_value, + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -74,6 +88,7 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): async def test_run_default_path_flag(workflow_mock, format_and_export_prediction): input_data = { + 'metadata': metadata, "path_flag": "default", "data": {"test": "data"}, "timestamp": "2021-01-01", @@ -95,7 +110,8 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction 'timestamp': input_data['timestamp'], 'model_id': input_data['model_id'], 'prediction_confidence': input_data['prediction_confidence'], - 'comment': input_data['comment'] + 'comment': input_data['comment'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -107,7 +123,8 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction Activities.write_opc_data, { 'opc_output_config': input_data['opc_output_config'], - 'data': workflow_mock.execute_local_activity_method.return_value + 'data': workflow_mock.execute_local_activity_method.return_value, + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY @@ -120,7 +137,8 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction { 'schema': input_data['schema'], 'table_name': input_data['table_name'], - 'data': workflow_mock.execute_activity_method.return_value + 'data': workflow_mock.execute_activity_method.return_value, + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 560565e..e76b586 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -9,12 +9,23 @@ def prediction_process(): return PredictionProcess() +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + + @mark.asyncio @patch("laborious.workflows.sub_workflows.prediction_process.workflow", new_callable=AsyncMock) async def test_run(workflow_mock, prediction_process): prediction_process.path_flag_handler = AsyncMock(return_value=False) # Arrange input_data = { + 'metadata': metadata, 'data': {'test': 'data'}, 'schema': 'test_schema', 'table_name': 'test_table', @@ -49,51 +60,61 @@ async def test_run(workflow_mock, prediction_process): assert workflow_mock.execute_local_activity_method.call_count == 7 workflow_mock.execute_local_activity_method.assert_has_calls([ - call(Activities.get_last_timestamp, {'data': input_data['data']}, - retry_policy=ANY, start_to_close_timeout=ANY)]) + call(Activities.get_last_timestamp, { + 'data': input_data['data'], + **metadata + }, + retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.input_gate, { 'filters': input_data['input_filters'], 'data': input_data['data'], - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'] + 'model_retention': input_data['model_retention'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_response_gate, { 'filters': input_data['mlflow_transform_filters'], 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'type': 'transform', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { 'filters': input_data['mlflow_transform_filters'], 'data': 'transformed_data', 'type': 'transform', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_predict, { 'data': 'transformed_data', 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'] + 'model_retention': input_data['model_retention'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_response_gate, { 'filters': input_data['mlflow_predict_filters'], 'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'}, 'type': 'predict', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_child_workflow.assert_called_once_with( 'format_and_export_prediction', { + 'metadata': metadata, 'path_flag': 'continue', 'data': 'predicted_data', 'prediction_confidence': 0.95, @@ -115,6 +136,7 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process): prediction_process.path_flag_handler = AsyncMock(return_value=True) # Arrange input_data = { + 'metadata': metadata, 'data': {'test': 'data'}, 'schema': 'test_schema', 'table_name': 'test_table', @@ -141,11 +163,15 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process): assert workflow_mock.execute_local_activity_method.call_count == 2 workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.get_last_timestamp, { - 'data': input_data['data']}, retry_policy=ANY, start_to_close_timeout=ANY), + 'data': input_data['data'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY), call(Activities.input_gate, { 'filters': input_data['input_filters'], 'data': input_data['data'], - 'path_priority': input_data['path_priority']}, retry_policy=ANY, start_to_close_timeout=ANY) + 'path_priority': input_data['path_priority'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY) ]) workflow_mock.execute_child_workflow.assert_not_called() @@ -156,6 +182,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ prediction_process.path_flag_handler = AsyncMock(side_effect=[False, True]) # Arrange input_data = { + 'metadata': metadata, 'data': {'test': 'data'}, 'schema': 'test_schema', 'table_name': 'test_table', @@ -183,19 +210,26 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ # Assert assert workflow_mock.execute_local_activity_method.call_count == 4 workflow_mock.execute_local_activity_method.assert_has_calls([ - call(Activities.get_last_timestamp, {'data': input_data['data']}, + call(Activities.get_last_timestamp, { + 'data': input_data['data'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.input_gate, { 'filters': input_data['input_filters'], 'data': input_data['data'], - 'path_priority': input_data['path_priority']}, + 'path_priority': input_data['path_priority'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention']}, + 'model_retention': input_data['model_retention'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY) ]) workflow_mock.execute_local_activity_method.assert_has_calls([ @@ -203,7 +237,8 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ 'filters': input_data['mlflow_transform_filters'], 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'type': 'transform', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY) ]) workflow_mock.execute_child_workflow.assert_not_called() @@ -216,6 +251,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process side_effect=[False, False, True]) # Arrange input_data = { + 'metadata': metadata, 'data': {'test': 'data'}, 'schema': 'test_schema', 'table_name': 'test_table', @@ -247,33 +283,41 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process assert workflow_mock.execute_local_activity_method.call_count == 5 workflow_mock.execute_local_activity_method.assert_has_calls([ - call(Activities.get_last_timestamp, {'data': input_data['data']}, + call(Activities.get_last_timestamp, { + 'data': input_data['data'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.input_gate, { 'filters': input_data['input_filters'], 'data': input_data['data'], - 'path_priority': input_data['path_priority']}, + 'path_priority': input_data['path_priority'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'] + 'model_retention': input_data['model_retention'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_response_gate, { 'filters': input_data['mlflow_transform_filters'], 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'type': 'transform', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { 'filters': input_data['mlflow_transform_filters'], 'data': 'transformed_data', 'type': 'transform', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_child_workflow.assert_not_called() @@ -285,6 +329,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p side_effect=[False, False, False, True]) # Arrange input_data = { + 'metadata': metadata, 'data': {'test': 'data'}, 'schema': 'test_schema', 'table_name': 'test_table', @@ -317,46 +362,56 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p # Assert assert workflow_mock.execute_local_activity_method.call_count == 7 workflow_mock.execute_local_activity_method.assert_has_calls([ - call(Activities.get_last_timestamp, {'data': input_data['data']}, + call(Activities.get_last_timestamp, { + 'data': input_data['data'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.input_gate, { 'filters': input_data['input_filters'], 'data': input_data['data'], - 'path_priority': input_data['path_priority']}, + 'path_priority': input_data['path_priority'], + **metadata + }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'] + 'model_retention': input_data['model_retention'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_response_gate, { 'filters': input_data['mlflow_transform_filters'], 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'type': 'transform', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { 'filters': input_data['mlflow_transform_filters'], 'data': 'transformed_data', 'type': 'transform', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_predict, { 'data': 'transformed_data', 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'] + 'model_retention': input_data['model_retention'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_response_gate, { 'filters': input_data['mlflow_predict_filters'], 'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'}, 'type': 'predict', - 'path_priority': input_data['path_priority'] + 'path_priority': input_data['path_priority'], + **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_child_workflow.assert_not_called() @@ -378,6 +433,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process): # Act result = await prediction_process.path_flag_handler( data, path_flag, { + 'metadata': metadata, 'schema': schema, 'table_name': table_name, 'model_id': model, @@ -410,6 +466,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): # Act result = await prediction_process.path_flag_handler( data, path_flag, { + 'metadata': metadata, 'schema': schema, 'table_name': table_name, 'model_id': model, @@ -424,6 +481,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): workflow_mock.execute_activity_method.assert_called_once_with( Activities.repeat_last_prediction, { + **metadata, 'schema': schema, 'table_name': table_name, 'model': model, @@ -452,6 +510,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): # Act result = await prediction_process.path_flag_handler( data, path_flag, { + 'metadata': metadata, 'schema': schema, 'table_name': table_name, 'model_id': model, @@ -468,6 +527,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): workflow_mock.execute_child_workflow.assert_called_once_with( 'format_and_export_prediction', { + 'metadata': metadata, 'path_flag': path_flag, 'data': data, 'prediction_confidence': confidence, @@ -500,6 +560,7 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process): # Act result = await prediction_process.path_flag_handler( data, path_flag, { + **metadata, 'schema': schema, 'table_name': table_name, 'model_id': model, diff --git a/tests/laborious/workflows/test_predictions_batch.py b/tests/laborious/workflows/test_predictions_batch.py index 0ca45e1..3c8fc82 100644 --- a/tests/laborious/workflows/test_predictions_batch.py +++ b/tests/laborious/workflows/test_predictions_batch.py @@ -9,6 +9,16 @@ def predictions_batch() -> PredictionsBatch: return PredictionsBatch() +metadata = { + "metadata": { + "model_id": "test_model_id", + "model_name": "test_model", + "workflow_name": "predictions_batch", + "schedule_name": "test_schedule", + }, +} + + @mark.asyncio @patch('laborious.workflows.predictions_batch.workflow', new_callable=AsyncMock) async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch): @@ -29,27 +39,17 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch workflow_mock.execute_local_activity_method.assert_has_calls([ call( - Activities.prepare_activity, + Activities.load_custom_query, { - 'schedule_name': input_data['schedule_name'], - 'model_name': input_data['model_name'], - 'model_id': input_data['model_id'], - 'workflow_name': 'predictions_batch' + **metadata, + 'query': input_data['query'], }, retry_policy=ANY, start_to_close_timeout=ANY ) ]) - - workflow_mock.execute_local_activity_method.assert_has_calls([ - call( - Activities.load_custom_query, - input_data['query'], - retry_policy=ANY, - start_to_close_timeout=ANY - ) - ]) prediction_input = { + 'metadata': metadata, 'data': {'data': 'test_data'}, 'schema': input_data['schema'], 'table_name': input_data['table_name'], diff --git a/values.yaml b/values.yaml index ba03752..9e5c34a 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.1" + tag: "0.2.4" # 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-laborious_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1097-realizar-testes-basicos-no-cluster-suse-linux" + value: "SIENTIAPDE-1110-criar-testes-e-2-e" - name: PYTHON_APP value: "laborious.worker.worker" @@ -153,9 +153,9 @@ env: value: "aignosi" - name: OPC_ID - value: "server-1" + value: "1" - name: OPC_URL - value: "opc.tcp://sientia-opc-simulator.sientia.svc.cluster.local:4840" + value: "opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840" - name: KAFKA_BOOTSTRAP_SERVERS value: "kafka.kafka.svc.cluster.local:9092" @@ -178,7 +178,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-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat +# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0-uat # kubectl create secret generic git-ssh-key-sientia-laborious-worker \ # --namespace sientia \