Merge pull request #4 from Aignosi/SIENTIAPDE-1097-realizar-testes-basicos-no-cluster-suse-linux
Sientiapde 1097 realizar testes basicos no cluster suse linux
This commit is contained in:
@@ -81,10 +81,10 @@ class Gates(BaseActivity):
|
||||
self.logger.error(f"Filter {fil} not found")
|
||||
continue
|
||||
try:
|
||||
if input_filter_functions[fil](data, config):
|
||||
if input_filter_functions[fil](data, config['config']):
|
||||
self.logger.debug(
|
||||
f"Data not passed the input filter {fil}:{config}")
|
||||
filter_output.append(config['POLICY'])
|
||||
filter_output.append(config['policy'])
|
||||
except Exception as e:
|
||||
trace = traceback.format_exc()
|
||||
self.notification_handler.build_and_send_notification(
|
||||
@@ -139,7 +139,7 @@ class Gates(BaseActivity):
|
||||
continue
|
||||
try:
|
||||
if mlflow_response_filter_functions[fil](data, config):
|
||||
filter_output.append(config['POLICY'])
|
||||
filter_output.append(config['policy'])
|
||||
comments.append(data['content']['message'])
|
||||
self.notification_handler.build_and_send_notification(
|
||||
notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}",
|
||||
@@ -201,7 +201,7 @@ class Gates(BaseActivity):
|
||||
continue
|
||||
try:
|
||||
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(
|
||||
notification_id=f"{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}",
|
||||
message=f"Data not passed the content filter {fil}:{config}",
|
||||
|
||||
@@ -13,7 +13,7 @@ def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool
|
||||
bool: True if the specific columns have null values, False otherwise.
|
||||
"""
|
||||
return not data[
|
||||
data['variable'].isin(config['VARIABLES']) & data['value'].isna()].empty
|
||||
data['variable'].isin(config['variables']) & data['value'].isna()].empty
|
||||
|
||||
|
||||
def filter_empty_data(data: DataFrame, _config: dict) -> bool:
|
||||
|
||||
@@ -32,10 +32,6 @@ async def main():
|
||||
servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'),
|
||||
logger=logger,
|
||||
project_name=os.getenv('PROJECT_NAME', 'laborious'),
|
||||
pipeline_name='-',
|
||||
trigger_name='-',
|
||||
model_name='-',
|
||||
model='-'
|
||||
)
|
||||
|
||||
logger.info('Starting Activities...')
|
||||
@@ -60,7 +56,7 @@ async def main():
|
||||
workers = [
|
||||
Worker(
|
||||
temporal_client,
|
||||
task_queue='predictions-queue',
|
||||
task_queue='predictions_batch-queue',
|
||||
workflows=[PredictionsBatch, PredictionProcess,
|
||||
FormatAndExportPrediction],
|
||||
activities=[
|
||||
|
||||
@@ -205,7 +205,8 @@ class PredictionProcess():
|
||||
{
|
||||
'schema': schema,
|
||||
'table_name': table_name,
|
||||
'model_id': model_id
|
||||
'model': model_id,
|
||||
'last_timestamp': last_timestamp
|
||||
},
|
||||
retry_policy=retry_policy,
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
|
||||
@@ -3,5 +3,5 @@ psycopg2-binary
|
||||
sqlalchemy
|
||||
asyncua
|
||||
redis
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14
|
||||
git+ssh://git@github.com/Aignosi/sientia-mlops-library.git
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.17
|
||||
git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1
|
||||
|
||||
@@ -42,7 +42,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac
|
||||
side_effect=Exception("Test error"))
|
||||
input_data = {
|
||||
'filters': {
|
||||
'EMPTY_DATA': {'POLICY': 'STOP'}
|
||||
'EMPTY_DATA': {'policy': 'STOP', 'config': {}}
|
||||
},
|
||||
'data': {'value': []},
|
||||
'path_priority': ['STOP', 'CONTINUE', 'REPEAT']
|
||||
@@ -55,7 +55,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac
|
||||
assert result == (None, 0, "")
|
||||
gates_activity.notification_handler.build_and_send_notification.assert_called_once_with(
|
||||
notification_id="INTPUT_GATE_ERROR__EMPTY_DATA",
|
||||
message="Error in filter EMPTY_DATA:{'POLICY': 'STOP'}: \n Test error",
|
||||
message="Error in filter EMPTY_DATA:{'policy': 'STOP', 'config': {}}: \n Test error",
|
||||
block="input_gate",
|
||||
level=NotificationLevel.ERROR,
|
||||
attachment_content=ANY
|
||||
@@ -84,7 +84,7 @@ async def test_input_gate_with_filter(gates_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
'filters': {
|
||||
'EMPTY_DATA': {'POLICY': 'STOP'}
|
||||
'EMPTY_DATA': {'policy': 'STOP', 'config': {}}
|
||||
},
|
||||
'data': {'value': []},
|
||||
'path_priority': ['STOP', 'CONTINUE', 'REPEAT']
|
||||
@@ -171,7 +171,7 @@ async def test_mlflow_response_gate_with_filter(gates_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
'filters': {
|
||||
'API_ERROR': {'POLICY': 'STOP'}
|
||||
'API_ERROR': {'policy': 'STOP'}
|
||||
},
|
||||
'data': {
|
||||
'success': False,
|
||||
@@ -273,7 +273,7 @@ async def test_mlflow_content_gate_with_filter(gates_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
'filters': {
|
||||
'NAN_VALUES': {'POLICY': 'STOP'}
|
||||
'NAN_VALUES': {'policy': 'STOP', 'config': {}}
|
||||
},
|
||||
'data': {'value': [None, None, None]},
|
||||
'type': 'test',
|
||||
|
||||
@@ -10,14 +10,14 @@ def test_filter_specific_variables_null_values():
|
||||
assert filter_specific_variables_null_values(
|
||||
DataFrame(
|
||||
{'variable': ['variable1', 'variable2'], 'value': [1, 2]}),
|
||||
config={'VARIABLES': ['variable2']}) is False
|
||||
config={'variables': ['variable2']}) is False
|
||||
|
||||
|
||||
def test_filter_specific_variables_null_values_with_null_values():
|
||||
assert filter_specific_variables_null_values(
|
||||
DataFrame(
|
||||
{'variable': ['variable1', 'variable2'], 'value': [1, None]}),
|
||||
config={'VARIABLES': ['variable2']}) is True
|
||||
config={'variables': ['variable2']}) is True
|
||||
|
||||
|
||||
def test_filter_empty_data():
|
||||
|
||||
@@ -426,7 +426,8 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
|
||||
{
|
||||
'schema': schema,
|
||||
'table_name': table_name,
|
||||
'model_id': model
|
||||
'model': model,
|
||||
'last_timestamp': last_timestamp,
|
||||
},
|
||||
retry_policy=ANY,
|
||||
start_to_close_timeout=ANY
|
||||
|
||||
@@ -11,7 +11,7 @@ image:
|
||||
# This sets the pull policy for images.
|
||||
pullPolicy: Always
|
||||
# Overrides the image tag whose default is the chart appVersion.
|
||||
tag: "0.0.2"
|
||||
tag: "0.1.1"
|
||||
|
||||
# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
|
||||
imagePullSecrets:
|
||||
@@ -123,7 +123,7 @@ env:
|
||||
- name: GITHUB_REPO_URL
|
||||
value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
|
||||
- name: GITHUB_BRANCH
|
||||
value: "SIENTIAPDE-994-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas"
|
||||
value: "SIENTIAPDE-1097-realizar-testes-basicos-no-cluster-suse-linux"
|
||||
- name: PYTHON_APP
|
||||
value: "laborious.worker.worker"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user