38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
from unittest.mock import MagicMock
|
|
from pytest import fixture, mark
|
|
from sientia_do.notifications.models import Notification
|
|
from scouter.activities.base import BaseActivity
|
|
|
|
|
|
@fixture
|
|
def base_activity():
|
|
return BaseActivity(
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_prepare_activity(base_activity):
|
|
base_activity.notification_handler.base_notification = Notification(
|
|
project="project",
|
|
pipeline="pipeline",
|
|
trigger="-",
|
|
model_name="-",
|
|
model_id="-",
|
|
)
|
|
|
|
await base_activity.prepare_activity(
|
|
{
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'model_name': 'test_model',
|
|
'model_id': 'test_model_id',
|
|
}
|
|
)
|
|
|
|
assert base_activity.notification_handler.base_notification.schedule_name == "test_schedule"
|
|
assert base_activity.notification_handler.base_notification.model_name == "test_model"
|
|
assert base_activity.notification_handler.base_notification.model_id == "test_model_id"
|
|
assert base_activity.notification_handler.base_notification.pipeline_name == "test_workflow"
|