Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
234 changes: 121 additions & 113 deletions test/test_linked_tasks.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
from pathlib import Path
import unittest

from defusedxml.ElementTree import fromstring
import pytest
Expand All @@ -15,115 +14,124 @@
RUN_LINKED_TASK_NOW = asset_dir / "linked_tasks_run_now.xml"


class TestLinkedTasks(unittest.TestCase):
def setUp(self) -> None:
self.server = TSC.Server("http://test", False)
self.server.version = "3.15"

# Fake signin
self.server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
self.server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"

self.baseurl = self.server.linked_tasks.baseurl

def test_parse_linked_task_flow_run(self):
xml = fromstring(GET_LINKED_TASKS.read_bytes())
task_runs = LinkedTaskFlowRunItem._parse_element(xml, self.server.namespace)
assert 1 == len(task_runs)
task = task_runs[0]
assert task.flow_run_id == "e3d1fc25-5644-4e32-af35-58dcbd1dbd73"
assert task.flow_run_priority == 1
assert task.flow_run_consecutive_failed_count == 3
assert task.flow_run_task_type == "runFlow"
assert task.flow_id == "ab1231eb-b8ca-461e-a131-83f3c2b6a673"
assert task.flow_name == "flow-name"

def test_parse_linked_task_step(self):
xml = fromstring(GET_LINKED_TASKS.read_bytes())
steps = LinkedTaskStepItem.from_task_xml(xml, self.server.namespace)
assert 1 == len(steps)
step = steps[0]
assert step.id == "f554a4df-bb6f-4294-94ee-9a709ef9bda0"
assert step.stop_downstream_on_failure
assert step.step_number == 1
assert 1 == len(step.task_details)
task = step.task_details[0]
assert task.flow_run_id == "e3d1fc25-5644-4e32-af35-58dcbd1dbd73"
assert task.flow_run_priority == 1
assert task.flow_run_consecutive_failed_count == 3
assert task.flow_run_task_type == "runFlow"
assert task.flow_id == "ab1231eb-b8ca-461e-a131-83f3c2b6a673"
assert task.flow_name == "flow-name"

def test_parse_linked_task(self):
tasks = LinkedTaskItem.from_response(GET_LINKED_TASKS.read_bytes(), self.server.namespace)
assert 1 == len(tasks)
task = tasks[0]
assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"

def test_get_linked_tasks(self):
with requests_mock.mock() as m:
m.get(self.baseurl, text=GET_LINKED_TASKS.read_text())
tasks, pagination_item = self.server.linked_tasks.get()

assert 1 == len(tasks)
task = tasks[0]
assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"

def test_get_by_id_str_linked_task(self):
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"

with requests_mock.mock() as m:
m.get(f"{self.baseurl}/{id_}", text=GET_LINKED_TASKS.read_text())
task = self.server.linked_tasks.get_by_id(id_)

assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"

def test_get_by_id_obj_linked_task(self):
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
in_task = LinkedTaskItem()
in_task.id = id_

with requests_mock.mock() as m:
m.get(f"{self.baseurl}/{id_}", text=GET_LINKED_TASKS.read_text())
task = self.server.linked_tasks.get_by_id(in_task)

assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"

def test_run_now_str_linked_task(self):
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"

with requests_mock.mock() as m:
m.post(f"{self.baseurl}/{id_}/runNow", text=RUN_LINKED_TASK_NOW.read_text())
job = self.server.linked_tasks.run_now(id_)

assert job.id == "269a1e5a-1220-4a13-ac01-704982693dd8"
assert job.status == "InProgress"
assert job.created_at == parse_datetime("2022-02-15T00:22:22Z")
assert job.linked_task_id == id_

def test_run_now_obj_linked_task(self):
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
in_task = LinkedTaskItem()
in_task.id = id_

with requests_mock.mock() as m:
m.post(f"{self.baseurl}/{id_}/runNow", text=RUN_LINKED_TASK_NOW.read_text())
job = self.server.linked_tasks.run_now(in_task)

assert job.id == "269a1e5a-1220-4a13-ac01-704982693dd8"
assert job.status == "InProgress"
assert job.created_at == parse_datetime("2022-02-15T00:22:22Z")
assert job.linked_task_id == id_
@pytest.fixture(scope="function")
def server():
"""Fixture to create a TSC.Server instance for testing."""
server = TSC.Server("http://test", False)

# Fake signin
server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"
server.version = "3.15"

return server


def test_parse_linked_task_flow_run(server: TSC.Server) -> None:
xml = fromstring(GET_LINKED_TASKS.read_bytes())
task_runs = LinkedTaskFlowRunItem._parse_element(xml, server.namespace)
assert 1 == len(task_runs)
task = task_runs[0]
assert task.flow_run_id == "e3d1fc25-5644-4e32-af35-58dcbd1dbd73"
assert task.flow_run_priority == 1
assert task.flow_run_consecutive_failed_count == 3
assert task.flow_run_task_type == "runFlow"
assert task.flow_id == "ab1231eb-b8ca-461e-a131-83f3c2b6a673"
assert task.flow_name == "flow-name"


def test_parse_linked_task_step(server: TSC.Server) -> None:
xml = fromstring(GET_LINKED_TASKS.read_bytes())
steps = LinkedTaskStepItem.from_task_xml(xml, server.namespace)
assert 1 == len(steps)
step = steps[0]
assert step.id == "f554a4df-bb6f-4294-94ee-9a709ef9bda0"
assert step.stop_downstream_on_failure
assert step.step_number == 1
assert 1 == len(step.task_details)
task = step.task_details[0]
assert task.flow_run_id == "e3d1fc25-5644-4e32-af35-58dcbd1dbd73"
assert task.flow_run_priority == 1
assert task.flow_run_consecutive_failed_count == 3
assert task.flow_run_task_type == "runFlow"
assert task.flow_id == "ab1231eb-b8ca-461e-a131-83f3c2b6a673"
assert task.flow_name == "flow-name"


def test_parse_linked_task(server: TSC.Server) -> None:
tasks = LinkedTaskItem.from_response(GET_LINKED_TASKS.read_bytes(), server.namespace)
assert 1 == len(tasks)
task = tasks[0]
assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"


def test_get_linked_tasks(server: TSC.Server) -> None:
with requests_mock.mock() as m:
m.get(server.linked_tasks.baseurl, text=GET_LINKED_TASKS.read_text())
tasks, pagination_item = server.linked_tasks.get()

assert 1 == len(tasks)
task = tasks[0]
assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"


def test_get_by_id_str_linked_task(server: TSC.Server) -> None:
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"

with requests_mock.mock() as m:
m.get(f"{server.linked_tasks.baseurl}/{id_}", text=GET_LINKED_TASKS.read_text())
task = server.linked_tasks.get_by_id(id_)

assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"


def test_get_by_id_obj_linked_task(server: TSC.Server) -> None:
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
in_task = LinkedTaskItem()
in_task.id = id_

with requests_mock.mock() as m:
m.get(f"{server.linked_tasks.baseurl}/{id_}", text=GET_LINKED_TASKS.read_text())
task = server.linked_tasks.get_by_id(in_task)

assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
assert task.num_steps == 1
assert task.schedule is not None
assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"


def test_run_now_str_linked_task(server: TSC.Server) -> None:
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"

with requests_mock.mock() as m:
m.post(f"{server.linked_tasks.baseurl}/{id_}/runNow", text=RUN_LINKED_TASK_NOW.read_text())
job = server.linked_tasks.run_now(id_)

assert job.id == "269a1e5a-1220-4a13-ac01-704982693dd8"
assert job.status == "InProgress"
assert job.created_at == parse_datetime("2022-02-15T00:22:22Z")
assert job.linked_task_id == id_


def test_run_now_obj_linked_task(server: TSC.Server) -> None:
id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
in_task = LinkedTaskItem()
in_task.id = id_

with requests_mock.mock() as m:
m.post(f"{server.linked_tasks.baseurl}/{id_}/runNow", text=RUN_LINKED_TASK_NOW.read_text())
job = server.linked_tasks.run_now(in_task)

assert job.id == "269a1e5a-1220-4a13-ac01-704982693dd8"
assert job.status == "InProgress"
assert job.created_at == parse_datetime("2022-02-15T00:22:22Z")
assert job.linked_task_id == id_
Loading