Skip to content

Commit

Permalink
adding tests to workflows
Browse files Browse the repository at this point in the history 
adding tests to workflows

fixing tests

fixing tests

fixing tests

fixing tests
  • Loading branch information
@nishantmunjal7
nishantmunjal7 committed May 6, 2024
1 parent f67ad7f commit 6bc9559
Show file tree
Hide file tree
Showing 3 changed files with 59 additions and 36 deletions.
6 changes: 4 additions & 2 deletions pyatlan/client/constants.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -417,12 +417,14 @@
WORKFLOW_INDEX_RUN_SEARCH = API(
WORKFLOW_INDEX_RUN_API, HTTPMethod.POST, HTTPStatus.OK, endpoint=EndPoint.HERACLES
)
WORKFLOW_RERUN_API = "workflows/submit" #triggers a workflow using the current user's credentials
# triggers a workflow using the current user's credentials
WORKFLOW_RERUN_API = "workflows/submit"
WORKFLOW_RERUN = API(
WORKFLOW_RERUN_API, HTTPMethod.POST, HTTPStatus.OK, endpoint=EndPoint.HERACLES
)

WORKFLOW_OWNER_RERUN_API = "workflows/triggerAsOwner" #triggers a workflow using the workflow owner's credentials
# triggers a workflow using the workflow owner's credentials
WORKFLOW_OWNER_RERUN_API = "workflows/triggerAsOwner"
WORKFLOW_OWNER_RERUN = API(
WORKFLOW_OWNER_RERUN_API, HTTPMethod.POST, HTTPStatus.OK, endpoint=EndPoint.HERACLES
)
Expand Down
4 changes: 3 additions & 1 deletion pyatlan/client/workflow.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@
WORKFLOW_UPDATE,
SCHEDULE_QUERY_WORKFLOWS_SEARCH,
SCHEDULE_QUERY_WORKFLOWS_MISSED,
WORKFLOW_CHANGE_OWNER, WORKFLOW_OWNER_RERUN,
WORKFLOW_CHANGE_OWNER,
WORKFLOW_OWNER_RERUN,
)
from pyatlan.errors import ErrorCode
from pyatlan.model.enums import AtlanWorkflowPhase, WorkflowPackage
Expand Down Expand Up @@ -159,6 +160,7 @@ def find_missed_schedule_query_crons_between_duration(
"startDate": request.start_date,
"endDate": request.end_date,
}

raw_json = self._client._call_api(
SCHEDULE_QUERY_WORKFLOWS_MISSED, query_params=query_params
)
Expand Down
85 changes: 52 additions & 33 deletions tests/unit/test_workflow_client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,9 @@
WorkflowSearchResult,
WorkflowSearchResultDetail,
WorkflowSearchResultStatus,
WorkflowSpec, WorkflowRunResponseList, ScheduleQueriesSearchRequest,
WorkflowSpec,
WorkflowRunResponseList,
ScheduleQueriesSearchRequest,
)


Expand Down Expand Up @@ -355,7 +357,10 @@ def test_update_when_given_wrong_parameter_raises_validation_error(
assert error_msg in str(err.value)


def test_update_workflow_owner(client: WorkflowClient, mock_api_caller,run_response: WorkflowResponse,
def test_update_workflow_owner(
client: WorkflowClient,
mock_api_caller,
run_response: WorkflowResponse,
):
mock_api_caller._call_api.return_value = run_response.dict()
response = client.update_workflow_owner(
Expand All @@ -366,14 +371,16 @@ def test_update_workflow_owner(client: WorkflowClient, mock_api_caller,run_respo

@pytest.fixture()
def workflows_list(search_result: WorkflowSearchResult) -> WorkflowRunResponseList:
return WorkflowRunResponseList(
items=[]
) # type: ignore[call-arg]
def test_find_missed_schedule_query_crons_between_duration(client: WorkflowClient, mock_api_caller, workflows_list: WorkflowRunResponseList):
return WorkflowRunResponseList(items=[]) # type: ignore[call-arg]


def test_find_missed_schedule_query_crons_between_duration(
client: WorkflowClient, mock_api_caller, workflows_list: WorkflowRunResponseList
):
response = client.find_missed_schedule_query_crons_between_duration(
ScheduleQueriesSearchRequest(
start_date="2024-05-03T16:30:00.000+05:30",
end_date="2024-05-05T00:59:00.000+05:30",
startDate="2024-05-03T16:30:00.000+05:30",
endDate="2024-05-05T00:59:00.000+05:30",
)
)
assert response == workflows_list
Expand All @@ -384,38 +391,39 @@ def saved_query_workflows() -> List[WorkflowSearchResult]:
return [
WorkflowSearchResult(
_index="test_index",
_seq_no="test_seq_no",
_primary_term="test_primary_term",
_type="test_type",
_id="test_id",
sort=["test_sort"],
_source={
"apiVersion": "v1",
"kind": "Workflow",
"metadata": {
"name": "asq-123456",
"annotations": {
"package.argoproj.io/name": "@atlan/schedule-query"
}
},
"spec": {}
}
)
_source=WorkflowSearchResultDetail(
api_version="test_api_version",
kind="test_kind",
metadata=WorkflowMetadata(
name="asq-123456", namespace="test_namespace"
),
spec=WorkflowSpec(),
),
),
]

def test_find_schedule_query_cron_by_saved_query_id(client: WorkflowClient, mock_api_caller, saved_query_workflows: List[WorkflowSearchResult]):

def test_find_schedule_query_cron_by_saved_query_id(
client: WorkflowClient,
mock_api_caller,
saved_query_workflows: List[WorkflowSearchResult],
):
mock_response = {
"hits": {
"total": {
"value": 1, # Adding total hits count
"relation": "eq"
},
"hits": saved_query_workflows
"total": {"value": 1, "relation": "eq"}, # Adding total hits count
"hits": saved_query_workflows,
},
"_shards": { # Required field _shards
"total": 5,
"successful": 5,
"skipped": 0,
"failed": 0
}
"failed": 0,
},
}
mock_api_caller._call_api.return_value = mock_response

Expand All @@ -424,8 +432,12 @@ def test_find_schedule_query_cron_by_saved_query_id(client: WorkflowClient, mock

# Assertions to check if the method behaves as expected
assert len(results) == 1 # Check if one result is returned
assert isinstance(results[0], WorkflowSearchResult) # Validate the type of the result
assert results[0].source.metadata.name == "asq-123456" # Validate the content of the result
assert isinstance(
results[0], WorkflowSearchResult
) # Validate the type of the result
assert (
results[0].source.metadata.name == "asq-123456"
) # Validate the content of the result


@pytest.fixture()
Expand All @@ -435,10 +447,18 @@ def re_trigger_schedule_query_workflow_response() -> WorkflowRunResponse:
metadata=WorkflowMetadata(name="scheduled-query-123", namespace="default"),
spec=WorkflowSpec(),
)
def test_re_trigger_schedule_query_workflow(client, mock_api_caller, re_trigger_schedule_query_workflow_response: WorkflowRunResponse):


def test_re_trigger_schedule_query_workflow(
client,
mock_api_caller,
re_trigger_schedule_query_workflow_response: WorkflowRunResponse,
):
"""Test the re_trigger_schedule_query_workflow method."""

mock_api_caller._call_api.return_value = re_trigger_schedule_query_workflow_response.dict()
mock_api_caller._call_api.return_value = (
re_trigger_schedule_query_workflow_response.dict()
)

schedule_query_id = "scheduled-query-123"
namespace = "default"
Expand All @@ -450,4 +470,3 @@ def test_re_trigger_schedule_query_workflow(client, mock_api_caller, re_trigger_
assert isinstance(response, WorkflowRunResponse)
assert response.metadata.name == schedule_query_id
assert response.metadata.namespace == namespace

0 comments on commit 6bc9559

Please sign in to comment.