gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #7 from Aignosi/SIENTIAPDE-1110-criar-testes-e-2-e

Browse Source 
Sientiapde 1110 criar testes e 2 e
This commit is contained in:
Matheus Demoner
2025-07-08 19:14:17 -03:00
committed by GitHub
parent 2be78b1aea f4610fec5b
commit 3a01a9a244
21 changed files with 473 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
laborious/activities/activities.py
View File

@@ -3,11 +3,11 @@ from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
from sientia_do.temporal.activities.postgres import Postgres from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.temporal.utils.logger import Logger
from laborious.activities.mlflow import MLFlow from laborious.activities.mlflow import MLFlow
from laborious.activities.gates import Gates from laborious.activities.gates import Gates
from laborious.activities.opc import OPC from laborious.activities.opc import OPC
from typing import Any from typing import Any
from logging import Logger
class Activities(Postgres, MLFlow, Gates, OPC): class Activities(Postgres, MLFlow, Gates, OPC):
@@ -44,10 +44,6 @@ class Activities(Postgres, MLFlow, Gates, OPC):
logger=logger, logger=logger,
notification_handler=notification_handler) 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): def shutdown(self):
Postgres.close(self) Postgres.close(self)
OPC.shutdown(self) OPC.shutdown(self)

66
laborious/activities/gates.py
View File

@@ -3,10 +3,10 @@ from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
import traceback import traceback
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity 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 laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter
from typing import Any from typing import Any
from laborious.utils.filters.conditional_filters import ( 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 tuple[str | None, int, str]: (policy, confidence, comments) based in priority
list and filter configuration and functions. 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'] filters = input_data['filters']
data = DataFrame(input_data['data']) data = DataFrame(input_data['data'])
@@ -73,21 +76,22 @@ class Gates(BaseActivity):
filter_output = [] filter_output = []
self.logger.debug(f"Input data:\n {data}") self.debug(f"Input data:\n {data}", metadata)
self.logger.debug(f"Filters: {filters}") self.debug(f"Filters: {filters}", metadata)
for fil, config in filters.items(): for fil, config in filters.items():
if fil not in input_filter_functions: if fil not in input_filter_functions:
self.logger.error(f"Filter {fil} not found") self.error(f"Filter {fil} not found", metadata)
continue continue
try: try:
if input_filter_functions[fil](data, config['config']): if input_filter_functions[fil](data, config['config']):
self.logger.debug( self.debug(
f"Data not passed the input filter {fil}:{config}") f"Data not passed the input filter {fil}:{config}", metadata)
filter_output.append(config['policy']) filter_output.append(config['policy'])
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id=f"INTPUT_GATE_ERROR__{fil}", notification_id=f"INTPUT_GATE_ERROR__{fil}",
message=f"Error in filter {fil}:{config}: \n {e}", message=f"Error in filter {fil}:{config}: \n {e}",
block="input_gate", block="input_gate",
@@ -97,11 +101,11 @@ class Gates(BaseActivity):
for path_flag in path_priority: for path_flag in path_priority:
if path_flag in filter_output: 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], \ return path_flag, input_filter_functions['path_confidence'][path_flag], \
"Input data with bad quality" "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, "" return None, 0, ""
@activity.defn(name="mlflow_response_gate") @activity.defn(name="mlflow_response_gate")
@@ -121,7 +125,8 @@ class Gates(BaseActivity):
and filter configuration and functions. 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'] filters = input_data['filters']
data = input_data['data'] data = input_data['data']
@@ -130,8 +135,8 @@ class Gates(BaseActivity):
filter_output = [] filter_output = []
self.logger.debug(f"Input data:\n {data}") self.debug(f"Input data:\n {data}", metadata)
self.logger.debug(f"Filters: {filters}") self.debug(f"Filters: {filters}", metadata)
comments = [] comments = []
for fil, config in filters.items(): for fil, config in filters.items():
@@ -141,7 +146,8 @@ class Gates(BaseActivity):
if mlflow_response_filter_functions[fil](data, config): if mlflow_response_filter_functions[fil](data, config):
filter_output.append(config['policy']) filter_output.append(config['policy'])
comments.append(data['content']['message']) 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}", notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}",
message=data['content']['message'], message=data['content']['message'],
block="mlflow_gate", block="mlflow_gate",
@@ -150,7 +156,8 @@ class Gates(BaseActivity):
) )
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id=f"MLFLOW_GATE_RESPONSE_FILTER__{fil}", notification_id=f"MLFLOW_GATE_RESPONSE_FILTER__{fil}",
message=f"Error in filter {fil}:{config}: \n {e}", message=f"Error in filter {fil}:{config}: \n {e}",
block="mlflow_gate", block="mlflow_gate",
@@ -160,11 +167,12 @@ class Gates(BaseActivity):
for path_flag in path_priority: for path_flag in path_priority:
if path_flag in filter_output: 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], \ return path_flag, mlflow_response_filter_functions['path_confidence'][path_flag], \
", ".join(comments) ", ".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, "" return None, 0, ""
@activity.defn(name="mlflow_content_gate") @activity.defn(name="mlflow_content_gate")
@@ -184,7 +192,8 @@ class Gates(BaseActivity):
list and filter configuration and functions. 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'] filters = input_data['filters']
data = DataFrame(input_data['data']) data = DataFrame(input_data['data'])
@@ -193,8 +202,8 @@ class Gates(BaseActivity):
filter_output = [] filter_output = []
self.logger.debug(f"Input data:\n {data}") self.debug(f"Input data:\n {data}", metadata)
self.logger.debug(f"Filters: {filters}") self.debug(f"Filters: {filters}", metadata)
for fil, config in filters.items(): for fil, config in filters.items():
if fil not in mlflow_content_filter_functions: if fil not in mlflow_content_filter_functions:
@@ -202,7 +211,8 @@ class Gates(BaseActivity):
try: try:
if mlflow_content_filter_functions[fil](data, config): if mlflow_content_filter_functions[fil](data, config):
filter_output.append(config['policy']) 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}", notification_id=f"{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}",
message=f"Data not passed the content filter {fil}:{config}", message=f"Data not passed the content filter {fil}:{config}",
block="mlflow_gate", block="mlflow_gate",
@@ -211,7 +221,8 @@ class Gates(BaseActivity):
) )
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id=f"MLFLOW_GATE_CONTENT_FILTER__{fil}", notification_id=f"MLFLOW_GATE_CONTENT_FILTER__{fil}",
message=f"Error in filter {fil}:{config}: \n {e}", message=f"Error in filter {fil}:{config}: \n {e}",
block="mlflow_gate", block="mlflow_gate",
@@ -221,11 +232,12 @@ class Gates(BaseActivity):
for path_flag in path_priority: for path_flag in path_priority:
if path_flag in filter_output: 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], \ return path_flag, mlflow_content_filter_functions['path_confidence'][path_flag], \
"Transformed data not passed the content filter" "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, "" return None, 0, ""
@activity.defn(name="format_prediction") @activity.defn(name="format_prediction")
@@ -241,7 +253,8 @@ class Gates(BaseActivity):
Returns: Returns:
dict: The formatted data. dict: The formatted data.
""" """
self.logger.debug("Formatting prediction...") metadata = input_data['metadata']
self.debug("Formatting prediction...", metadata)
data = DataFrame(input_data['data']) data = DataFrame(input_data['data'])
data['timestamp'] = input_data['timestamp'] data['timestamp'] = input_data['timestamp']
@@ -269,7 +282,8 @@ class Gates(BaseActivity):
dict: The formatted data. dict: The formatted data.
""" """
self.logger.debug("Formatting default prediction...") metadata = input_data['metadata']
self.debug("Formatting default prediction...", metadata)
return DataFrame({ return DataFrame({
'prediction': [0], 'prediction': [0],

29
laborious/activities/mlflow.py
View File

@@ -6,9 +6,9 @@ from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
from sientia_do.temporal.activities.base import BaseActivity from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.temporal.utils.logger import Logger
from laborious.utils.repository.model_repository import MLFlowRepository from laborious.utils.repository.model_repository import MLFlowRepository
from typing import Any from typing import Any
from logging import Logger
class MLFlow(BaseActivity): class MLFlow(BaseActivity):
@@ -36,13 +36,19 @@ class MLFlow(BaseActivity):
Returns: Returns:
dict[str, Any]: The transformed data. 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']) data = DataFrame(input_data['data'])
model_name = input_data['model_name'] model_name = input_data['model_name']
model_retention = input_data['model_retention'] model_retention = input_data['model_retention']
self.logger.debug("Raw input data:") self.debug("Raw input data:", metadata)
self.logger.debug(data) 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( data = data.pivot(
index='timestamp', columns='variable', index='timestamp', columns='variable',
@@ -51,14 +57,14 @@ class MLFlow(BaseActivity):
data.reset_index(inplace=True) data.reset_index(inplace=True)
data.columns.name = None data.columns.name = None
self.logger.debug("Processed input data:") self.debug("Processed input data:", metadata)
self.logger.debug(data) self.debug(data, metadata)
response_data = self.model_monitoring_repository.transform( response_data = self.model_monitoring_repository.transform(
model_name, data, model_retention) model_name, data, model_retention)
self.logger.debug("Response data:") self.debug("Response data:", metadata)
self.logger.debug(response_data) self.debug(response_data, metadata)
return response_data return response_data
@@ -74,18 +80,19 @@ class MLFlow(BaseActivity):
Returns: Returns:
dict[str, Any]: The predicted data. 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']) data = DataFrame(input_data['data'])
model_name = input_data['model_name'] model_name = input_data['model_name']
model_retention = input_data['model_retention'] model_retention = input_data['model_retention']
self.logger.debug(data) self.debug(data, metadata)
data.replace(np.nan, None, inplace=True) data.replace(np.nan, None, inplace=True)
response_data = self.model_monitoring_repository.predict( response_data = self.model_monitoring_repository.predict(
model_name, data, model_retention) model_name, data, model_retention)
self.logger.debug(response_data) self.debug(response_data, metadata)
return response_data return response_data

58
laborious/activities/opc.py
View File

@@ -2,10 +2,10 @@ from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity 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 laborious.utils.repository.opc_repository import OpcRepository
from typing import Any from typing import Any
import traceback import traceback
@@ -23,9 +23,9 @@ class OPC(BaseActivity):
self.opc_servers = opc_servers self.opc_servers = opc_servers
self.opc_repository = {} self.opc_repository = {}
for name, server in opc_servers.items(): for id, server in opc_servers.items():
self.opc_repository[name] = OpcRepository( self.opc_repository[id] = OpcRepository(
name=name, id=server['id'],
url=server['url'], url=server['url'],
logger=self.logger, logger=self.logger,
server_uri=server['server_uri'], server_uri=server['server_uri'],
@@ -35,17 +35,17 @@ class OPC(BaseActivity):
notification_handler=self.notification_handler, notification_handler=self.notification_handler,
reconnection_interval=server['reconnection_interval'], reconnection_interval=server['reconnection_interval'],
) )
self.opc_repository[name].connect() self.opc_repository[id].connect()
BaseActivity.__init__(self, logger, notification_handler) BaseActivity.__init__(self, logger, notification_handler)
def write_data(self, server: str, tag: str, data: Any, def write_data(self, server_id: str, tag: str, data: Any,
data_type: str, tag_type: str) -> bool: data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool:
""" """
Write data to OPC server. Write data to OPC server.
Args: 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. - tag (str): The tag to write to.
- data (Any): The data to write. - data (Any): The data to write.
- data_type (str): The data type. - data_type (str): The data type.
@@ -56,12 +56,12 @@ class OPC(BaseActivity):
""" """
try: try:
return self.opc_repository[server].write_data( return self.opc_repository[server_id].write_data(
tag, data, data_type) tag, data, data_type, self.logger, metadata)
self.logger.debug(f"Wrote {tag_type} to {tag}")
except Exception as e: except Exception as e:
trace = traceback.format_exc() 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", notification_id=f"WRITE_OPC_{tag_type.upper()}_ERROR",
message=f"Error writing data to OPC server: {e}", message=f"Error writing data to OPC server: {e}",
block="write_opc_data", block="write_opc_data",
@@ -89,41 +89,44 @@ class OPC(BaseActivity):
- dict[Any, Any]: The data that was written to the OPC servers. - 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']) data = DataFrame(input_data['data'])
opc_output_config = input_data['opc_output_config'] opc_output_config = input_data['opc_output_config']
self.logger.debug(data) self.debug(data, metadata)
success = True success = True
for server, config in opc_output_config.items(): for server_id, config in opc_output_config.items():
if self.opc_repository.get(server) is None: if self.opc_repository.get(server_id) is None:
self.logger.error(f"OPC server {server} not found") self.error(f"OPC server {server_id} not found", metadata)
continue continue
if 'prediction_tags' in config: if 'prediction_tags' in config:
for tag, tag_config in config['prediction_tags'].items(): for tag, tag_config in config['prediction_tags'].items():
success = success and self.write_data( success = success and self.write_data(
server=server, server_id=server_id,
tag=tag, tag=tag,
data=data.head(1)['prediction'].values[0], data=data.head(1)['prediction'].values[0],
data_type=tag_config['data_type'], data_type=tag_config['data_type'],
tag_type='prediction' tag_type='prediction',
metadata=metadata
) )
if 'confidence_tags' in config: if 'confidence_tags' in config:
for tag, tag_config in config['confidence_tags'].items(): for tag, tag_config in config['confidence_tags'].items():
self.write_data( self.write_data(
server=server, server_id=server_id,
tag=tag, tag=tag,
data=data.head(1)['prediction_confidence'].values[0], data=data.head(1)['prediction_confidence'].values[0],
data_type=tag_config['data_type'], 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. Processes the confidence of OPC server write operations and updates the DataFrame accordingly.
@@ -137,15 +140,16 @@ class OPC(BaseActivity):
Returns: Returns:
dict[Any, Any]: The processed data as a dictionary. dict[Any, Any]: The processed data as a dictionary.
""" """
if not success: if not success:
data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE
self.logger.debug( self.debug(
"Some data could not be written to OPC servers, setting confidence to " f"Some data could not be written to OPC servers, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.",
f"{OPC_WRITTING_ERROR_CONFIDENCE}." metadata
) )
else: else:
self.logger.info("Data written to OPC servers successfully.") self.debug("Data written to OPC servers successfully.", metadata)
return data.to_dict() return data.to_dict()

4
laborious/utils/connectors_config.py
View File

@@ -34,8 +34,8 @@ def build_opc_config():
return json.loads(opc_raw) return json.loads(opc_raw)
return { return {
'opc': { getenv('OPC_ID', '1'): {
'name': getenv('OPC_NAME', 'opc'), 'id': getenv('OPC_ID', '1'),
'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'), 'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'),
'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), 'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
'cert_path': getenv('OPC_CERT_PATH', None), 'cert_path': getenv('OPC_CERT_PATH', None),

68
laborious/utils/repository/opc_repository.py
View File

@@ -1,14 +1,14 @@
import traceback import traceback
from logging import Logger
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from asyncua.sync import Client from asyncua.sync import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256 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 regex import F
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.utils.logger import Logger
data_type_map = { data_type_map = {
'float': { 'float': {
@@ -35,12 +35,12 @@ data_type_map = {
class OpcRepository(): class OpcRepository():
def __init__(self, name: str, url: str, logger: Logger, def __init__(self, id: str, url: str, logger: Logger,
notification_handler: NotificationHandler, notification_handler: NotificationHandler,
reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None, reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None,
private_key_path: str = None, server_cert_path: str = None): private_key_path: str = None, server_cert_path: str = None):
self.url = url self.url = url
self.name = name self.id = id
self.server_uri = server_uri self.server_uri = server_uri
self.cert_path = cert_path self.cert_path = cert_path
self.private_key_path = private_key_path self.private_key_path = private_key_path
@@ -104,7 +104,7 @@ class OpcRepository():
self.client = Client(self.url) self.client = Client(self.url)
if self.cert_path: if self.cert_path:
self.set_security() self.set_security()
self.logger.info('Starting connection...') self.logger.info(f'Starting connection to OPC server {self.id}...')
return self.try_connect() return self.try_connect()
def try_connect(self): def try_connect(self):
@@ -114,6 +114,7 @@ class OpcRepository():
Returns: Returns:
bool: True if the connection was successful, False otherwise. bool: True if the connection was successful, False otherwise.
""" """
try: try:
self.last_reconnection_time = datetime.now() self.last_reconnection_time = datetime.now()
self.client.connect() self.client.connect()
@@ -121,7 +122,7 @@ class OpcRepository():
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( 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}", message=f"Failed to connect to OPC server: {e}",
block="opc_repository", block="opc_repository",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
@@ -136,9 +137,12 @@ class OpcRepository():
""" """
if self.client is None: if self.client is None:
return 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.client = None
self.logger.info('Disconnected from OPC server')
def __del__(self): def __del__(self):
""" """
@@ -165,7 +169,7 @@ class OpcRepository():
if self.error_count > 5: if self.error_count > 5:
self.logger.warning( 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: try:
self.disconnect() self.disconnect()
except Exception as e: except Exception as e:
@@ -173,7 +177,7 @@ class OpcRepository():
self.logger.error(f"Failed to disconnect from OPC server: {e}") self.logger.error(f"Failed to disconnect from OPC server: {e}")
self.logger.error(trace) self.logger.error(trace)
self.logger.info( self.logger.info(
f"Attempting to reconnect to OPC server {self.name}...") f"Attempting to reconnect to OPC server {self.id}...")
return self.connect() return self.connect()
if hasattr(self.client, 'aio_obj') and self.client.aio_obj.uaclient.protocol is None or \ 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.client.aio_obj.uaclient.protocol.state == "closed"):
self.logger.error( self.logger.error(
f"OPC server {self.name} is not connected") f"OPC server {self.id} is not connected")
if (datetime.now() - self.last_reconnection_time).total_seconds( if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds(
) > self.reconnection_interval: ) > self.reconnection_interval:
self.disconnect()
self.logger.error( self.logger.error(
f"Trying to reconnect to OPC server {self.name}...") f"Trying to reconnect to OPC server {self.id}...")
return self.try_connect() return self.connect()
return False return False
return True 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. Writes data to the OPC server.
If the connection is not established, it attempts to reconnect. If the connection is not established, it attempts to reconnect.
@@ -210,42 +216,54 @@ class OpcRepository():
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.notification_handler.build_and_send_notification(
notification_id=f"OPC_WRITE_GET_NODE_ERROR_{self.name}", notification_id=f"OPC_WRITE_GET_NODE_ERROR_{self.id}",
message=f"Failed to get node from OPC server: {e}", message=f"Failed to get node from OPC server: {e} | metadata: {metadata}",
block="opc_repository", block="opc_repository",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=trace attachment_content=trace
) )
self.logger.error(trace) logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
self.error_count += 1 self.error_count += 1
return False return False
if data_type not in data_type_map: if data_type not in data_type_map:
self.notification_handler.build_and_send_notification( self.notification_handler.build_and_send_notification(
notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.name}", notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
message=f"Unsupported data type: {data_type}", message=f"Unsupported data type: {data_type} | metadata: {metadata}",
block="opc_repository", block="opc_repository",
level=NotificationLevel.ERROR level=NotificationLevel.ERROR
) )
return False return False
data = data_type_map[data_type]['converter'](value) 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( 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: try:
node.write_value(ua_data) node.write_value(ua_data)
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.notification_handler.build_and_send_notification(
notification_id=f"OPC_WRITE_DATA_ERROR_{self.name}", notification_id=f"OPC_WRITE_DATA_ERROR_{self.id}",
message=f"Failed to write data to OPC server: {e}", message=f"Failed to write data to OPC server: {e} | metadata: {metadata}",
block="opc_repository", block="opc_repository",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=trace attachment_content=trace
) )
self.logger.error(trace) logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
self.error_count += 1 self.error_count += 1
return False return False
self.error_count = 0 self.error_count = 0

4
laborious/worker/worker.py
View File

@@ -60,8 +60,6 @@ async def main():
workflows=[PredictionsBatch, PredictionProcess, workflows=[PredictionsBatch, PredictionProcess,
FormatAndExportPrediction], FormatAndExportPrediction],
activities=[ activities=[
# Base
activities.prepare_activity,
# MLFlow # MLFlow
activities.request_predict, activities.request_predict,
activities.request_transform, 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. # If an exception occurs in any of the worker handlers, it will be propagated here.
await asyncio.gather(*handlers) await asyncio.gather(*handlers)
except BaseException as e: except BaseException as e:
logger.error("An unhandled exception occurred: %s", e, exc_info=True) logger.error(f"An unhandled exception occurred: {e}")
finally: finally:
if notification_handler: if notification_handler:
notification_handler.shutdown() notification_handler.shutdown()

17
laborious/workflows/predictions_batch.py
View File

@@ -40,27 +40,28 @@ class PredictionsBatch():
Exception: If any of the required parameters are missing or if the workflow fails. Exception: If any of the required parameters are missing or if the workflow fails.
""" """
await workflow.execute_local_activity_method( metadata = {
Activities.prepare_activity, 'metadata': {
{
'schedule_name': input_data['schedule_name'], 'schedule_name': input_data['schedule_name'],
'model_name': input_data['model_name'], 'model_name': input_data['model_name'],
'model_id': input_data['model_id'], 'model_id': input_data['model_id'],
'workflow_name': 'predictions_batch' 'workflow_name': 'predictions_batch'
}, }
retry_policy=retry_policy, }
start_to_close_timeout=timedelta(seconds=60)
)
data = await workflow.execute_local_activity_method( data = await workflow.execute_local_activity_method(
Activities.load_custom_query, Activities.load_custom_query,
input_data['query'], {
**metadata,
'query': input_data['query'],
},
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60) start_to_close_timeout=timedelta(seconds=60)
) )
# Prepare input for prediction_process workflow # Prepare input for prediction_process workflow
prediction_input = { prediction_input = {
'metadata': metadata,
'data': data, 'data': data,
'schema': input_data['schema'], 'schema': input_data['schema'],
'table_name': input_data['table_name'], 'table_name': input_data['table_name'],

5
laborious/workflows/sub_workflows/format_and_export_prediction.py
View File

@@ -36,6 +36,7 @@ class FormatAndExportPrediction():
Returns: Returns:
bool: True if the workflow was successful, False otherwise. bool: True if the workflow was successful, False otherwise.
""" """
metadata = input_data['metadata']
path_flag = input_data['path_flag'] path_flag = input_data['path_flag']
data = input_data['data'] data = input_data['data']
prediction_confidence = input_data['prediction_confidence'] prediction_confidence = input_data['prediction_confidence']
@@ -45,6 +46,7 @@ class FormatAndExportPrediction():
prediction = await workflow.execute_local_activity_method( prediction = await workflow.execute_local_activity_method(
Activities.format_prediction, Activities.format_prediction,
{ {
**metadata,
'data': data, 'data': data,
'timestamp': input_data['timestamp'], 'timestamp': input_data['timestamp'],
'model_id': input_data['model_id'], 'model_id': input_data['model_id'],
@@ -59,6 +61,7 @@ class FormatAndExportPrediction():
prediction = await workflow.execute_local_activity_method( prediction = await workflow.execute_local_activity_method(
Activities.format_default_prediction, Activities.format_default_prediction,
{ {
**metadata,
'timestamp': input_data['timestamp'], 'timestamp': input_data['timestamp'],
'model_id': input_data['model_id'], 'model_id': input_data['model_id'],
'prediction_confidence': prediction_confidence, 'prediction_confidence': prediction_confidence,
@@ -72,6 +75,7 @@ class FormatAndExportPrediction():
prediction = await workflow.execute_activity_method( prediction = await workflow.execute_activity_method(
Activities.write_opc_data, Activities.write_opc_data,
{ {
**metadata,
'opc_output_config': input_data['opc_output_config'], 'opc_output_config': input_data['opc_output_config'],
'data': prediction 'data': prediction
}, },
@@ -83,6 +87,7 @@ class FormatAndExportPrediction():
await workflow.execute_activity_method( await workflow.execute_activity_method(
Activities.export_data_to_postgres, Activities.export_data_to_postgres,
{ {
**metadata,
'schema': input_data['schema'], 'schema': input_data['schema'],
'table_name': input_data['table_name'], 'table_name': input_data['table_name'],
'data': prediction, 'data': prediction,

30
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -41,6 +41,7 @@ class PredictionProcess():
Exception: If any of the required parameters are missing or if the workflow fails. Exception: If any of the required parameters are missing or if the workflow fails.
""" """
metadata = input_data['metadata']
data = input_data['data'] data = input_data['data']
model_id = input_data['model_id'] model_id = input_data['model_id']
model_name = input_data['model_name'] model_name = input_data['model_name']
@@ -49,19 +50,26 @@ class PredictionProcess():
last_timestamp = await workflow.execute_local_activity_method( last_timestamp = await workflow.execute_local_activity_method(
Activities.get_last_timestamp, Activities.get_last_timestamp,
{ {
**metadata,
'data': data 'data': data
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(minutes=1), 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( path_flag, confidence, comment = await workflow.execute_local_activity_method(
Activities.input_gate, Activities.input_gate,
{ gate_input,
'filters': input_data['input_filters'],
'data': data,
'path_priority': input_data['path_priority']
},
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(minutes=1), start_to_close_timeout=timedelta(minutes=1),
) )
@@ -74,6 +82,7 @@ class PredictionProcess():
response_data = await workflow.execute_local_activity_method( response_data = await workflow.execute_local_activity_method(
Activities.request_transform, Activities.request_transform,
{ {
**metadata,
'data': data, 'data': data,
'model_name': model_name, 'model_name': model_name,
'model_retention': model_retention 'model_retention': model_retention
@@ -85,6 +94,7 @@ class PredictionProcess():
path_flag, confidence, comment = await workflow.execute_local_activity_method( path_flag, confidence, comment = await workflow.execute_local_activity_method(
Activities.mlflow_response_gate, Activities.mlflow_response_gate,
{ {
**metadata,
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': response_data, 'data': response_data,
'type': 'transform', 'type': 'transform',
@@ -104,6 +114,7 @@ class PredictionProcess():
path_flag, confidence, comment = await workflow.execute_local_activity_method( path_flag, confidence, comment = await workflow.execute_local_activity_method(
Activities.mlflow_content_gate, Activities.mlflow_content_gate,
{ {
**metadata,
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': transformed_data, 'data': transformed_data,
'type': 'transform', 'type': 'transform',
@@ -121,6 +132,7 @@ class PredictionProcess():
response_data = await workflow.execute_local_activity_method( response_data = await workflow.execute_local_activity_method(
Activities.request_predict, Activities.request_predict,
{ {
**metadata,
'data': transformed_data, 'data': transformed_data,
'model_name': model_name, 'model_name': model_name,
'model_retention': model_retention 'model_retention': model_retention
@@ -132,6 +144,7 @@ class PredictionProcess():
path_flag, confidence, comment = await workflow.execute_local_activity_method( path_flag, confidence, comment = await workflow.execute_local_activity_method(
Activities.mlflow_response_gate, Activities.mlflow_response_gate,
{ {
**metadata,
'filters': input_data['mlflow_predict_filters'], 'filters': input_data['mlflow_predict_filters'],
'data': response_data, 'data': response_data,
'type': 'predict', 'type': 'predict',
@@ -149,6 +162,7 @@ class PredictionProcess():
await workflow.execute_child_workflow( await workflow.execute_child_workflow(
'format_and_export_prediction', 'format_and_export_prediction',
{ {
'metadata': metadata,
'path_flag': path_flag, 'path_flag': path_flag,
'data': response_data['content'], 'data': response_data['content'],
'prediction_confidence': confidence, 'prediction_confidence': confidence,
@@ -187,13 +201,15 @@ class PredictionProcess():
bool: True if the prediction should be stopped, False otherwise. bool: True if the prediction should be stopped, False otherwise.
""" """
metadata = input_data['metadata']
schema = input_data['schema'] schema = input_data['schema']
table_name = input_data['table_name'] table_name = input_data['table_name']
model_id = input_data['model_id'] model_id = input_data['model_id']
model_name = input_data['model_name'] model_name = input_data['model_name']
model_retention = input_data['model_retention'] 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': if path_flag == 'STOP':
return True return True
@@ -203,6 +219,7 @@ class PredictionProcess():
await workflow.execute_activity_method( await workflow.execute_activity_method(
Activities.repeat_last_prediction, Activities.repeat_last_prediction,
{ {
**metadata,
'schema': schema, 'schema': schema,
'table_name': table_name, 'table_name': table_name,
'model': model_id, 'model': model_id,
@@ -218,6 +235,7 @@ class PredictionProcess():
await workflow.execute_child_workflow( await workflow.execute_child_workflow(
'format_and_export_prediction', 'format_and_export_prediction',
{ {
'metadata': metadata,
'path_flag': path_flag, 'path_flag': path_flag,
'data': data, 'data': data,
'prediction_confidence': confidence, 'prediction_confidence': confidence,

2
requirements.txt
View File

@@ -3,5 +3,5 @@ psycopg2-binary
sqlalchemy sqlalchemy
asyncua asyncua
redis 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 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1

59
tests/laborious/activities/test_activities.py
View File

@@ -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.Postgres', return_value=MagicMock())
@patch('laborious.activities.activities.MLFlow', return_value=MagicMock()) @patch('laborious.activities.activities.MLFlow', return_value=MagicMock())
@patch('laborious.activities.activities.OPC', return_value=MagicMock()) @patch('laborious.activities.activities.OPC', return_value=MagicMock())

69
tests/laborious/activities/test_gates.py
View File

@@ -6,16 +6,34 @@ from laborious.activities.gates import Gates
@fixture @fixture
def gates_activity(): def gates_activity():
return Gates( gates = Gates(
logger=MagicMock(), logger=MagicMock(),
notification_handler=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 @mark.asyncio
async def test_input_gate_invalid_filter(gates_activity): async def test_input_gate_invalid_filter(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'INVALID_FILTER': {'POLICY': 'STOP'} 'INVALID_FILTER': {'POLICY': 'STOP'}
}, },
@@ -28,8 +46,8 @@ async def test_input_gate_invalid_filter(gates_activity):
# Assert # Assert
assert result == (None, 0, "") assert result == (None, 0, "")
gates_activity.logger.error.assert_called_once_with( gates_activity.error.assert_called_once_with(
"Filter INVALID_FILTER not found" "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( mock_input_filter_functions.__getitem__.return_value = MagicMock(
side_effect=Exception("Test error")) side_effect=Exception("Test error"))
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'EMPTY_DATA': {'policy': 'STOP', 'config': {}} 'EMPTY_DATA': {'policy': 'STOP', 'config': {}}
}, },
@@ -53,7 +72,8 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac
# Assert # Assert
assert result == (None, 0, "") 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", notification_id="INTPUT_GATE_ERROR__EMPTY_DATA",
message="Error in filter EMPTY_DATA:{'policy': 'STOP', 'config': {}}: \n Test error", message="Error in filter EMPTY_DATA:{'policy': 'STOP', 'config': {}}: \n Test error",
block="input_gate", 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): async def test_input_gate_no_filters(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': {}, 'filters': {},
'data': {'value': [1, 2, 3]}, 'data': {'value': [1, 2, 3]},
'path_priority': ['CONTINUE', 'STOP', 'REPEAT'] 'path_priority': ['CONTINUE', 'STOP', 'REPEAT']
@@ -76,13 +97,14 @@ async def test_input_gate_no_filters(gates_activity):
# Assert # Assert
assert result == (None, 0, "") assert result == (None, 0, "")
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
@mark.asyncio @mark.asyncio
async def test_input_gate_with_filter(gates_activity): async def test_input_gate_with_filter(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'EMPTY_DATA': {'policy': 'STOP', 'config': {}} 'EMPTY_DATA': {'policy': 'STOP', 'config': {}}
}, },
@@ -95,13 +117,14 @@ async def test_input_gate_with_filter(gates_activity):
# Assert # Assert
assert result == ('STOP', -1, "Input data with bad quality") assert result == ('STOP', -1, "Input data with bad quality")
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
@mark.asyncio @mark.asyncio
async def test_mlflow_response_gate_invalid_filter(gates_activity): async def test_mlflow_response_gate_invalid_filter(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'INVALID_FILTER': {'POLICY': 'STOP'} '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( mock_mlflow_response_filter_functions.__getitem__.return_value = MagicMock(
side_effect=Exception("Test error")) side_effect=Exception("Test error"))
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'INVALID_FILTER': {'POLICY': 'STOP'} 'INVALID_FILTER': {'POLICY': 'STOP'}
}, },
@@ -139,7 +163,8 @@ async def test_mlflow_response_gate_filter_exception(mock_mlflow_response_filter
# Assert # Assert
assert result == (None, 0, "") 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", notification_id="MLFLOW_GATE_RESPONSE_FILTER__INVALID_FILTER",
message="Error in filter INVALID_FILTER:{'POLICY': 'STOP'}: \n Test error", message="Error in filter INVALID_FILTER:{'POLICY': 'STOP'}: \n Test error",
block="mlflow_gate", 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): async def test_mlflow_response_gate_no_filters(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': {}, 'filters': {},
'data': {'content': {'message': 'success'}}, 'data': {'content': {'message': 'success'}},
'type': 'test', 'type': 'test',
@@ -163,13 +189,14 @@ async def test_mlflow_response_gate_no_filters(gates_activity):
# Assert # Assert
assert result == (None, 0, "") assert result == (None, 0, "")
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
@mark.asyncio @mark.asyncio
async def test_mlflow_response_gate_with_filter(gates_activity): async def test_mlflow_response_gate_with_filter(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'API_ERROR': {'policy': 'STOP'} 'API_ERROR': {'policy': 'STOP'}
}, },
@@ -189,14 +216,15 @@ async def test_mlflow_response_gate_with_filter(gates_activity):
# Assert # Assert
assert result == ('STOP', -1, "API error occurred") assert result == ('STOP', -1, "API error occurred")
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
gates_activity.notification_handler.build_and_send_notification.assert_called() gates_activity.send_notification.assert_called()
@mark.asyncio @mark.asyncio
async def test_mlflow_content_gate_invalid_filter(gates_activity): async def test_mlflow_content_gate_invalid_filter(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'INVALID_FILTER': {'POLICY': 'STOP'} '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( mock_mlflow_content_filter_functions.__getitem__.return_value = MagicMock(
side_effect=Exception("Test error")) side_effect=Exception("Test error"))
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'API_ERROR': {'POLICY': 'STOP'} 'API_ERROR': {'POLICY': 'STOP'}
}, },
@@ -240,8 +269,9 @@ async def test_mlflow_content_gate_filter_exception(mock_mlflow_content_filter_f
# Assert # Assert
assert result == (None, 0, "") assert result == (None, 0, "")
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
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_CONTENT_FILTER__API_ERROR", notification_id="MLFLOW_GATE_CONTENT_FILTER__API_ERROR",
message="Error in filter API_ERROR:{'POLICY': 'STOP'}: \n Test error", message="Error in filter API_ERROR:{'POLICY': 'STOP'}: \n Test error",
block="mlflow_gate", 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): async def test_mlflow_content_gate_no_filters(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': {}, 'filters': {},
'data': {'value': [1, 2, 3]}, 'data': {'value': [1, 2, 3]},
'type': 'test', 'type': 'test',
@@ -265,13 +296,14 @@ async def test_mlflow_content_gate_no_filters(gates_activity):
# Assert # Assert
assert result == (None, 0, "") assert result == (None, 0, "")
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
@mark.asyncio @mark.asyncio
async def test_mlflow_content_gate_with_filter(gates_activity): async def test_mlflow_content_gate_with_filter(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'filters': { 'filters': {
'NAN_VALUES': {'policy': 'STOP', 'config': {}} 'NAN_VALUES': {'policy': 'STOP', 'config': {}}
}, },
@@ -286,14 +318,15 @@ async def test_mlflow_content_gate_with_filter(gates_activity):
# Assert # Assert
assert result == ( assert result == (
'STOP', -1, "Transformed data not passed the content filter") 'STOP', -1, "Transformed data not passed the content filter")
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
gates_activity.notification_handler.build_and_send_notification.assert_called() gates_activity.send_notification.assert_called()
@mark.asyncio @mark.asyncio
async def test_format_prediction(gates_activity): async def test_format_prediction(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'data': {'prediction': [1], 'response_time': [0.1]}, 'data': {'prediction': [1], 'response_time': [0.1]},
'timestamp': '2023-05-26 11:12:27', 'timestamp': '2023-05-26 11:12:27',
'model_id': 'test_model', '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_confidence'] == {0: 0.9}
assert result['prediction_status'] == {0: 'Good'} assert result['prediction_status'] == {0: 'Good'}
assert result['comments'] == {0: ""} assert result['comments'] == {0: ""}
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
@mark.asyncio @mark.asyncio
async def test_format_default_prediction(gates_activity): async def test_format_default_prediction(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'timestamp': '2023-05-26 11:12:27', 'timestamp': '2023-05-26 11:12:27',
'model_id': 'test_model', 'model_id': 'test_model',
'prediction_confidence': 0.1, '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_confidence'] == {0: 0.1}
assert result['prediction_status'] == {0: 'Bad'} assert result['prediction_status'] == {0: 'Bad'}
assert result['comments'] == {0: 'Test comment'} assert result['comments'] == {0: 'Test comment'}
gates_activity.logger.debug.assert_called() gates_activity.debug.assert_called()
@mark.asyncio @mark.asyncio
async def test_get_last_timestamp_with_data(gates_activity): async def test_get_last_timestamp_with_data(gates_activity):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'data': { 'data': {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'] 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28']
} }

31
tests/laborious/activities/test_mlflow.py
View File

@@ -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 @mark.asyncio
@patch("laborious.activities.mlflow.DataFrame") @patch("laborious.activities.mlflow.DataFrame")
@patch("laborious.activities.mlflow.max") @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_max.return_value = '2024-01-02'
# Mock input data # Mock input data
input_data = { input_data = {
**metadata,
'data': [ 'data': [
{'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0}, {'timestamp': '2024-01-01', 'variable': 'var1',
{'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0}, 'value': 1.0, 'created_at': '2024-01-01 12:00:00'},
{'timestamp': '2024-01-02', 'variable': 'var1', 'value': 3.0}, {'timestamp': '2024-01-01', 'variable': 'var2',
{'timestamp': '2024-01-02', 'variable': 'var2', 'value': 4.0} '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_name': 'test_model',
'model_retention': 30 'model_retention': 30
@@ -60,6 +79,9 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow):
expected_response = {'prediction': [0.5, 0.6]} expected_response = {'prediction': [0.5, 0.6]}
mlflow.model_monitoring_repository.transform.return_value = expected_response 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 # Call the method
response_data = await mlflow.request_transform(input_data) 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_max.return_value = '2024-01-02'
# Mock input data # Mock input data
input_data = { input_data = {
**metadata,
'data': [ 'data': [
{'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0}, {'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0},
{'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0}, {'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0},

48
tests/laborious/activities/test_opc.py
View File

@@ -5,6 +5,15 @@ from laborious.activities.opc import NotificationLevel
from laborious.activities.opc import OPC 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") @patch("laborious.activities.opc.OpcRepository")
def test___init__(mock_opc_repository): def test___init__(mock_opc_repository):
@@ -15,6 +24,7 @@ def test___init__(mock_opc_repository):
mock_notification_handler = MagicMock() mock_notification_handler = MagicMock()
servers = { servers = {
'server1': { 'server1': {
'id': 'server1',
'url': 'http://localhost:8080', 'url': 'http://localhost:8080',
'server_uri': 'opc.tcp://localhost:4840', 'server_uri': 'opc.tcp://localhost:4840',
'cert_path': '', 'cert_path': '',
@@ -23,6 +33,7 @@ def test___init__(mock_opc_repository):
'reconnection_interval': 60, 'reconnection_interval': 60,
}, },
'server2': { 'server2': {
'id': 'server2',
'url': 'http://localhost:8080', 'url': 'http://localhost:8080',
'server_uri': 'opc.tcp://localhost:4840', 'server_uri': 'opc.tcp://localhost:4840',
'cert_path': '', 'cert_path': '',
@@ -45,7 +56,7 @@ def test___init__(mock_opc_repository):
mock_opc_repository.assert_has_calls([ mock_opc_repository.assert_has_calls([
call( call(
name="server1", id="server1",
url="http://localhost:8080", url="http://localhost:8080",
logger=mock_logger, logger=mock_logger,
server_uri="opc.tcp://localhost:4840", server_uri="opc.tcp://localhost:4840",
@@ -58,7 +69,7 @@ def test___init__(mock_opc_repository):
]) ])
mock_opc_repository.assert_has_calls([ mock_opc_repository.assert_has_calls([
call( call(
name="server2", id="server2",
url="http://localhost:8080", url="http://localhost:8080",
logger=mock_logger, logger=mock_logger,
server_uri="opc.tcp://localhost:4840", server_uri="opc.tcp://localhost:4840",
@@ -79,6 +90,7 @@ def test___init__(mock_opc_repository):
def opc(mock_opc_repository): def opc(mock_opc_repository):
servers = { servers = {
'server1': { 'server1': {
'id': 'server1',
'url': 'http://localhost:8080', 'url': 'http://localhost:8080',
'server_uri': 'opc.tcp://localhost:4840', 'server_uri': 'opc.tcp://localhost:4840',
'cert_path': '', 'cert_path': '',
@@ -90,12 +102,15 @@ def opc(mock_opc_repository):
mock_opc_repository.write_data = MagicMock( mock_opc_repository.write_data = MagicMock(
return_value=True return_value=True
) )
return OPC( opc = OPC(
opc_servers=servers, opc_servers=servers,
logger=MagicMock(), logger=MagicMock(),
notification_handler=MagicMock() notification_handler=MagicMock()
) )
opc.send_notification = MagicMock()
return opc
WRITE_DATA_CASES = [ WRITE_DATA_CASES = [
('tag1', 'int', 50), ('tag1', 'int', 50),
@@ -107,10 +122,10 @@ WRITE_DATA_CASES = [
@mark.parametrize('tag,data_type,data', WRITE_DATA_CASES) @mark.parametrize('tag,data_type,data', WRITE_DATA_CASES)
def test_write_data_success(opc, tag, data_type, data): def test_write_data_success(opc, tag, data_type, data):
assert opc.write_data(server='server1', tag=tag, data=data, assert opc.write_data(server_id='server1', tag=tag, data=data,
data_type=data_type, tag_type='prediction') data_type=data_type, tag_type='prediction', metadata=metadata)
opc.opc_repository['server1'].write_data.assert_called_once_with( 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): def test_write_data_exception(opc):
@@ -118,11 +133,12 @@ def test_write_data_exception(opc):
"Test error") "Test error")
try: try:
opc.write_data(server='server1', tag='tag1', data=50, opc.write_data(server_id='server1', tag='tag1', data=50,
data_type='int', tag_type='prediction') data_type='int', tag_type='prediction', metadata=metadata)
except Exception: 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", notification_id="WRITE_OPC_PREDICTION_ERROR",
message="Error writing data to OPC server: Test error", message="Error writing data to OPC server: Test error",
block="write_opc_data", block="write_opc_data",
@@ -138,6 +154,7 @@ def test_write_data_exception(opc):
async def test_write_opc_data_success(opc): async def test_write_opc_data_success(opc):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'data': { 'data': {
'prediction': [0.75], 'prediction': [0.75],
'prediction_confidence': [0.95] 'prediction_confidence': [0.95]
@@ -163,19 +180,21 @@ async def test_write_opc_data_success(opc):
assert output == {'data': 'data'} assert output == {'data': 'data'}
opc.write_data.assert_has_calls([ opc.write_data.assert_has_calls([
call( call(
server='server1', server_id='server1',
tag='tag1', tag='tag1',
data=0.75, data=0.75,
data_type='float', data_type='float',
tag_type='prediction' tag_type='prediction',
metadata=metadata['metadata']
)]) )])
opc.write_data.assert_has_calls([ opc.write_data.assert_has_calls([
call( call(
server='server1', server_id='server1',
tag='tag2', tag='tag2',
data=0.95, data=0.95,
data_type='float', data_type='float',
tag_type='confidence' tag_type='confidence',
metadata=metadata['metadata']
) )
]) ])
assert opc.write_data.call_count == 2 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): async def test_write_opc_data_empty_config(opc):
# Arrange # Arrange
input_data = { input_data = {
**metadata,
'data': { 'data': {
'prediction': [0.75], 'prediction': [0.75],
'prediction_confidence': [0.95] '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): def test_process_confidence(opc, data, success, expected):
# Act # Act
result = opc.process_confidence(data, success) result = opc.process_confidence(data, success, metadata)
# Assert # Assert
assert result['prediction_confidence'][0] == expected assert result['prediction_confidence'][0] == expected

65
tests/laborious/utils/repository/test_opc_repository.py
View File

@@ -14,7 +14,7 @@ def mock_logger():
@fixture @fixture
def opc_repository(mock_logger): def opc_repository(mock_logger):
return OpcRepository( return OpcRepository(
name="test_repo", id="test_repo",
url="opc.tcp://localhost:4840", url="opc.tcp://localhost:4840",
logger=mock_logger, logger=mock_logger,
notification_handler=Mock(), notification_handler=Mock(),
@@ -34,8 +34,18 @@ def mock_client():
yield client_instance 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): 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.url == "opc.tcp://localhost:4840"
assert opc_repository.server_uri == "urn:test:server" assert opc_repository.server_uri == "urn:test:server"
assert opc_repository.cert_path == "/path/to/cert.pem" 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() opc_repository.client.connect.assert_called_once()
assert opc_repository.last_reconnection_time is not None assert opc_repository.last_reconnection_time is not None
opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( 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", message="Failed to connect to OPC server: Test error",
block="opc_repository", block="opc_repository",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
@@ -126,6 +136,17 @@ def test_disconnect(opc_repository, mock_client):
assert opc_repository.client is None 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): def test_validate_connection_none_client(opc_repository):
opc_repository.client = None opc_repository.client = None
opc_repository.connect = MagicMock() 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 = MagicMock()
opc_repository.client.aio_obj.uaclient.protocol = None opc_repository.client.aio_obj.uaclient.protocol = None
opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) 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() response = opc_repository.validate_connection()
opc_repository.try_connect.assert_called_once() opc_repository.connect.assert_called_once()
assert response == opc_repository.try_connect.return_value assert response == opc_repository.connect.return_value
def test_validate_connection_failed(opc_repository): 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): def test_write_data_validate_connection_do_nothing(opc_repository):
opc_repository.validate_connection = MagicMock(return_value=True) opc_repository.validate_connection = MagicMock(return_value=True)
opc_repository.client = MagicMock() 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.validate_connection.assert_called_once()
opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") 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.validate_connection = MagicMock(return_value=False)
opc_repository.client = MagicMock() opc_repository.client = MagicMock()
opc_repository.error_count = 0 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.validate_connection.assert_called_once()
opc_repository.client.get_node.assert_not_called() 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.client = MagicMock()
opc_repository.error_count = 0 opc_repository.error_count = 0
opc_repository.client.get_node.side_effect = Exception("Test error") 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.validate_connection.assert_called_once()
opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") 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( opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id=f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.name}", notification_id=f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}",
message="Failed to get node from OPC server: Test error", 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", block="opc_repository",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=ANY attachment_content=ANY
@@ -230,13 +254,14 @@ def test_write_data_invalid_data_type(opc_repository, mock_client):
mock_node = MagicMock() mock_node = MagicMock()
mock_client.get_node.return_value = mock_node 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() opc_repository.validate_connection.assert_called_once()
mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") mock_client.get_node.assert_called_once_with("ns=2;s=TestNode")
opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.name}", notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}",
message="Unsupported data type: invalid_type", 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", block="opc_repository",
level=NotificationLevel.ERROR level=NotificationLevel.ERROR
) )
@@ -248,12 +273,11 @@ def test_write_data(opc_repository, mock_client):
mock_node = MagicMock() mock_node = MagicMock()
mock_client.get_node.return_value = mock_node 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_client.get_node.assert_called_once_with("ns=2;s=TestNode")
mock_node.write_value.assert_called_once() mock_node.write_value.assert_called_once()
opc_repository.logger.info.assert_called_once_with(
"Writing 42.0 - <class 'float'> to " + str(mock_node))
def test_write_data_write_value_failed(opc_repository, mock_client): 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 opc_repository.error_count = 0
mock_client.get_node.return_value = mock_node mock_client.get_node.return_value = mock_node
mock_node.write_value.side_effect = Exception("Test error") 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() opc_repository.validate_connection.assert_called_once()
mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") mock_client.get_node.assert_called_once_with("ns=2;s=TestNode")
mock_node.write_value.assert_called_once() mock_node.write_value.assert_called_once()
opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id=f"OPC_WRITE_DATA_ERROR_{opc_repository.name}", notification_id=f"OPC_WRITE_DATA_ERROR_{opc_repository.id}",
message="Failed to write data to OPC server: Test error", 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", block="opc_repository",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=ANY attachment_content=ANY

20
tests/laborious/utils/test_connectors_config.py
View File

@@ -54,7 +54,7 @@ def test_build_opc_config_with_env_vars():
def test_build_opc_config_with_individual_env_vars(): def test_build_opc_config_with_individual_env_vars():
# Arrange # Arrange
environ.pop('OPC_CONFIG', None) environ.pop('OPC_CONFIG', None)
environ['OPC_NAME'] = 'test-name' environ['OPC_ID'] = '1'
environ['OPC_URL'] = 'opc.tcp://test:4840' environ['OPC_URL'] = 'opc.tcp://test:4840'
environ['OPC_SERVER_URI'] = 'opc.tcp://test:4840' environ['OPC_SERVER_URI'] = 'opc.tcp://test:4840'
environ['OPC_RECONNECTION_INTERVAL'] = '300' environ['OPC_RECONNECTION_INTERVAL'] = '300'
@@ -63,16 +63,16 @@ def test_build_opc_config_with_individual_env_vars():
config = build_opc_config() config = build_opc_config()
# Assert # Assert
assert config['opc']['name'] == 'test-name' assert config['1']['id'] == '1'
assert config['opc']['url'] == 'opc.tcp://test:4840' assert config['1']['url'] == 'opc.tcp://test:4840'
assert config['opc']['server_uri'] == 'opc.tcp://test:4840' assert config['1']['server_uri'] == 'opc.tcp://test:4840'
assert config['opc']['reconnection_interval'] == 300 assert config['1']['reconnection_interval'] == 300
def test_build_opc_config_with_defaults(): def test_build_opc_config_with_defaults():
# Arrange # Arrange
environ.pop('OPC_CONFIG', None) environ.pop('OPC_CONFIG', None)
environ.pop('OPC_NAME', None) environ.pop('OPC_ID', None)
environ.pop('OPC_URL', None) environ.pop('OPC_URL', None)
environ.pop('OPC_SERVER_URI', None) environ.pop('OPC_SERVER_URI', None)
environ.pop('OPC_RECONNECTION_INTERVAL', None) environ.pop('OPC_RECONNECTION_INTERVAL', None)
@@ -81,10 +81,10 @@ def test_build_opc_config_with_defaults():
config = build_opc_config() config = build_opc_config()
# Assert # Assert
assert config['opc']['name'] == 'opc' assert config['1']['id'] == '1'
assert config['opc']['url'] == 'opc.tcp://localhost:4840' assert config['1']['url'] == 'opc.tcp://localhost:4840'
assert config['opc']['server_uri'] == 'opc.tcp://localhost:4840' assert config['1']['server_uri'] == 'opc.tcp://localhost:4840'
assert config['opc']['reconnection_interval'] == 120 assert config['1']['reconnection_interval'] == 120
def test_build_postgres_config_with_env_vars(): def test_build_postgres_config_with_env_vars():

30
tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py
View File

@@ -10,11 +10,22 @@ def format_and_export_prediction():
return FormatAndExportPrediction() return FormatAndExportPrediction()
metadata = {
"metadata": {
"model_id": "test_model",
"model_name": "test_model",
"workflow_name": "test_workflow",
"schema_name": "test_schedule",
},
}
@mark.asyncio @mark.asyncio
@patch("laborious.workflows.sub_workflows.format_and_export_prediction.workflow", new_callable=AsyncMock) @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): async def test_run_none_path_flag(workflow_mock, format_and_export_prediction):
input_data = { input_data = {
'metadata': metadata,
"path_flag": None, "path_flag": None,
"data": {"test": "data"}, "data": {"test": "data"},
"timestamp": "2021-01-01", "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'], 'data': input_data['data'],
'timestamp': input_data['timestamp'], 'timestamp': input_data['timestamp'],
'model_id': input_data['model_id'], 'model_id': input_data['model_id'],
'prediction_confidence': input_data['prediction_confidence'] 'prediction_confidence': input_data['prediction_confidence'],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=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, Activities.write_opc_data,
{ {
'opc_output_config': input_data['opc_output_config'], '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, retry_policy=ANY,
start_to_close_timeout=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'], 'schema': input_data['schema'],
'table_name': input_data['table_name'], '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, retry_policy=ANY,
start_to_close_timeout=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): async def test_run_default_path_flag(workflow_mock, format_and_export_prediction):
input_data = { input_data = {
'metadata': metadata,
"path_flag": "default", "path_flag": "default",
"data": {"test": "data"}, "data": {"test": "data"},
"timestamp": "2021-01-01", "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'], 'timestamp': input_data['timestamp'],
'model_id': input_data['model_id'], 'model_id': input_data['model_id'],
'prediction_confidence': input_data['prediction_confidence'], 'prediction_confidence': input_data['prediction_confidence'],
'comment': input_data['comment'] 'comment': input_data['comment'],
**metadata
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=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, Activities.write_opc_data,
{ {
'opc_output_config': input_data['opc_output_config'], '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, retry_policy=ANY,
start_to_close_timeout=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'], 'schema': input_data['schema'],
'table_name': input_data['table_name'], '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, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY

113
tests/laborious/workflows/subworkflows/test_prediction_process.py
View File

@@ -9,12 +9,23 @@ def prediction_process():
return PredictionProcess() return PredictionProcess()
metadata = {
"metadata": {
"model_id": "test_model",
"model_name": "test_model",
"workflow_name": "test_workflow",
"schema_name": "test_schedule",
},
}
@mark.asyncio @mark.asyncio
@patch("laborious.workflows.sub_workflows.prediction_process.workflow", new_callable=AsyncMock) @patch("laborious.workflows.sub_workflows.prediction_process.workflow", new_callable=AsyncMock)
async def test_run(workflow_mock, prediction_process): async def test_run(workflow_mock, prediction_process):
prediction_process.path_flag_handler = AsyncMock(return_value=False) prediction_process.path_flag_handler = AsyncMock(return_value=False)
# Arrange # Arrange
input_data = { input_data = {
'metadata': metadata,
'data': {'test': 'data'}, 'data': {'test': 'data'},
'schema': 'test_schema', 'schema': 'test_schema',
'table_name': 'test_table', '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 assert workflow_mock.execute_local_activity_method.call_count == 7
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.get_last_timestamp, {'data': input_data['data']}, call(Activities.get_last_timestamp, {
retry_policy=ANY, start_to_close_timeout=ANY)]) 'data': input_data['data'],
**metadata
},
retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.input_gate, { call(Activities.input_gate, {
'filters': input_data['input_filters'], 'filters': input_data['input_filters'],
'data': input_data['data'], '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)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.request_transform, { call(Activities.request_transform, {
'data': input_data['data'], 'data': input_data['data'],
'model_name': input_data['model_name'], '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)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_response_gate, { call(Activities.mlflow_response_gate, {
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'},
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_content_gate, { call(Activities.mlflow_content_gate, {
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': 'transformed_data', 'data': 'transformed_data',
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.request_predict, { call(Activities.request_predict, {
'data': 'transformed_data', 'data': 'transformed_data',
'model_name': input_data['model_name'], '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)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_response_gate, { call(Activities.mlflow_response_gate, {
'filters': input_data['mlflow_predict_filters'], 'filters': input_data['mlflow_predict_filters'],
'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'}, 'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'},
'type': 'predict', 'type': 'predict',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_child_workflow.assert_called_once_with( workflow_mock.execute_child_workflow.assert_called_once_with(
'format_and_export_prediction', 'format_and_export_prediction',
{ {
'metadata': metadata,
'path_flag': 'continue', 'path_flag': 'continue',
'data': 'predicted_data', 'data': 'predicted_data',
'prediction_confidence': 0.95, '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) prediction_process.path_flag_handler = AsyncMock(return_value=True)
# Arrange # Arrange
input_data = { input_data = {
'metadata': metadata,
'data': {'test': 'data'}, 'data': {'test': 'data'},
'schema': 'test_schema', 'schema': 'test_schema',
'table_name': 'test_table', '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 assert workflow_mock.execute_local_activity_method.call_count == 2
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.get_last_timestamp, { 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, { call(Activities.input_gate, {
'filters': input_data['input_filters'], 'filters': input_data['input_filters'],
'data': input_data['data'], '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() 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]) prediction_process.path_flag_handler = AsyncMock(side_effect=[False, True])
# Arrange # Arrange
input_data = { input_data = {
'metadata': metadata,
'data': {'test': 'data'}, 'data': {'test': 'data'},
'schema': 'test_schema', 'schema': 'test_schema',
'table_name': 'test_table', 'table_name': 'test_table',
@@ -183,19 +210,26 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
# Assert # Assert
assert workflow_mock.execute_local_activity_method.call_count == 4 assert workflow_mock.execute_local_activity_method.call_count == 4
workflow_mock.execute_local_activity_method.assert_has_calls([ 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)]) retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.input_gate, { call(Activities.input_gate, {
'filters': input_data['input_filters'], 'filters': input_data['input_filters'],
'data': input_data['data'], '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)]) retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.request_transform, { call(Activities.request_transform, {
'data': input_data['data'], 'data': input_data['data'],
'model_name': input_data['model_name'], '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) retry_policy=ANY, start_to_close_timeout=ANY)
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ 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'], 'filters': input_data['mlflow_transform_filters'],
'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'},
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY) }, retry_policy=ANY, start_to_close_timeout=ANY)
]) ])
workflow_mock.execute_child_workflow.assert_not_called() 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]) side_effect=[False, False, True])
# Arrange # Arrange
input_data = { input_data = {
'metadata': metadata,
'data': {'test': 'data'}, 'data': {'test': 'data'},
'schema': 'test_schema', 'schema': 'test_schema',
'table_name': 'test_table', '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 assert workflow_mock.execute_local_activity_method.call_count == 5
workflow_mock.execute_local_activity_method.assert_has_calls([ 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)]) retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.input_gate, { call(Activities.input_gate, {
'filters': input_data['input_filters'], 'filters': input_data['input_filters'],
'data': input_data['data'], '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)]) retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.request_transform, { call(Activities.request_transform, {
'data': input_data['data'], 'data': input_data['data'],
'model_name': input_data['model_name'], '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)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_response_gate, { call(Activities.mlflow_response_gate, {
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'},
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_content_gate, { call(Activities.mlflow_content_gate, {
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': 'transformed_data', 'data': 'transformed_data',
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_child_workflow.assert_not_called() 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]) side_effect=[False, False, False, True])
# Arrange # Arrange
input_data = { input_data = {
'metadata': metadata,
'data': {'test': 'data'}, 'data': {'test': 'data'},
'schema': 'test_schema', 'schema': 'test_schema',
'table_name': 'test_table', 'table_name': 'test_table',
@@ -317,46 +362,56 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
# Assert # Assert
assert workflow_mock.execute_local_activity_method.call_count == 7 assert workflow_mock.execute_local_activity_method.call_count == 7
workflow_mock.execute_local_activity_method.assert_has_calls([ 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)]) retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.input_gate, { call(Activities.input_gate, {
'filters': input_data['input_filters'], 'filters': input_data['input_filters'],
'data': input_data['data'], '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)]) retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.request_transform, { call(Activities.request_transform, {
'data': input_data['data'], 'data': input_data['data'],
'model_name': input_data['model_name'], '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)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_response_gate, { call(Activities.mlflow_response_gate, {
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'},
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_content_gate, { call(Activities.mlflow_content_gate, {
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': 'transformed_data', 'data': 'transformed_data',
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.request_predict, { call(Activities.request_predict, {
'data': 'transformed_data', 'data': 'transformed_data',
'model_name': input_data['model_name'], '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)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.execute_local_activity_method.assert_has_calls([
call(Activities.mlflow_response_gate, { call(Activities.mlflow_response_gate, {
'filters': input_data['mlflow_predict_filters'], 'filters': input_data['mlflow_predict_filters'],
'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'}, 'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'},
'type': 'predict', 'type': 'predict',
'path_priority': input_data['path_priority'] 'path_priority': input_data['path_priority'],
**metadata
}, retry_policy=ANY, start_to_close_timeout=ANY)]) }, retry_policy=ANY, start_to_close_timeout=ANY)])
workflow_mock.execute_child_workflow.assert_not_called() workflow_mock.execute_child_workflow.assert_not_called()
@@ -378,6 +433,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
# Act # Act
result = await prediction_process.path_flag_handler( result = await prediction_process.path_flag_handler(
data, path_flag, { data, path_flag, {
'metadata': metadata,
'schema': schema, 'schema': schema,
'table_name': table_name, 'table_name': table_name,
'model_id': model, 'model_id': model,
@@ -410,6 +466,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
# Act # Act
result = await prediction_process.path_flag_handler( result = await prediction_process.path_flag_handler(
data, path_flag, { data, path_flag, {
'metadata': metadata,
'schema': schema, 'schema': schema,
'table_name': table_name, 'table_name': table_name,
'model_id': model, '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( workflow_mock.execute_activity_method.assert_called_once_with(
Activities.repeat_last_prediction, Activities.repeat_last_prediction,
{ {
**metadata,
'schema': schema, 'schema': schema,
'table_name': table_name, 'table_name': table_name,
'model': model, 'model': model,
@@ -452,6 +510,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
# Act # Act
result = await prediction_process.path_flag_handler( result = await prediction_process.path_flag_handler(
data, path_flag, { data, path_flag, {
'metadata': metadata,
'schema': schema, 'schema': schema,
'table_name': table_name, 'table_name': table_name,
'model_id': model, '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( workflow_mock.execute_child_workflow.assert_called_once_with(
'format_and_export_prediction', 'format_and_export_prediction',
{ {
'metadata': metadata,
'path_flag': path_flag, 'path_flag': path_flag,
'data': data, 'data': data,
'prediction_confidence': confidence, 'prediction_confidence': confidence,
@@ -500,6 +560,7 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
# Act # Act
result = await prediction_process.path_flag_handler( result = await prediction_process.path_flag_handler(
data, path_flag, { data, path_flag, {
**metadata,
'schema': schema, 'schema': schema,
'table_name': table_name, 'table_name': table_name,
'model_id': model, 'model_id': model,

28
tests/laborious/workflows/test_predictions_batch.py
View File

@@ -9,6 +9,16 @@ def predictions_batch() -> PredictionsBatch:
return PredictionsBatch() return PredictionsBatch()
metadata = {
"metadata": {
"model_id": "test_model_id",
"model_name": "test_model",
"workflow_name": "predictions_batch",
"schedule_name": "test_schedule",
},
}
@mark.asyncio @mark.asyncio
@patch('laborious.workflows.predictions_batch.workflow', new_callable=AsyncMock) @patch('laborious.workflows.predictions_batch.workflow', new_callable=AsyncMock)
async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch): 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([ workflow_mock.execute_local_activity_method.assert_has_calls([
call( call(
Activities.prepare_activity, Activities.load_custom_query,
{ {
'schedule_name': input_data['schedule_name'], **metadata,
'model_name': input_data['model_name'], 'query': input_data['query'],
'model_id': input_data['model_id'],
'workflow_name': 'predictions_batch'
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=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 = { prediction_input = {
'metadata': metadata,
'data': {'data': 'test_data'}, 'data': {'data': 'test_data'},
'schema': input_data['schema'], 'schema': input_data['schema'],
'table_name': input_data['table_name'], 'table_name': input_data['table_name'],

10
values.yaml
View File

@@ -11,7 +11,7 @@ image:
# This sets the pull policy for images. # This sets the pull policy for images.
pullPolicy: Always pullPolicy: Always
# Overrides the image tag whose default is the chart appVersion. # 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/ # 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: imagePullSecrets:
@@ -123,7 +123,7 @@ env:
- name: GITHUB_REPO_URL - name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
- name: GITHUB_BRANCH - name: GITHUB_BRANCH
value: "SIENTIAPDE-1097-realizar-testes-basicos-no-cluster-suse-linux" value: "SIENTIAPDE-1110-criar-testes-e-2-e"
- name: PYTHON_APP - name: PYTHON_APP
value: "laborious.worker.worker" value: "laborious.worker.worker"
@@ -153,9 +153,9 @@ env:
value: "aignosi" value: "aignosi"
- name: OPC_ID - name: OPC_ID
value: "server-1" value: "1"
- name: OPC_URL - 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 - name: KAFKA_BOOTSTRAP_SERVERS
value: "kafka.kafka.svc.cluster.local:9092" 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 # 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 \ # kubectl create secret generic git-ssh-key-sientia-laborious-worker \
# --namespace sientia \ # --namespace sientia \