chore: pytestify flowtask

tableauserverclient/server/endpoint/flow_task_endpoint.py

@@ -18,7 +18,7 @@ def baseurl(self) -> str:
         return f"{self.parent_srv.baseurl}/sites/{self.parent_srv.site_id}/tasks/flows"
 
     @api(version="3.22")
-    def create(self, flow_item: TaskItem) -> TaskItem:
+    def create(self, flow_item: TaskItem) -> bytes:
         if not flow_item:
             error = "No flow provided"
             raise ValueError(error)

test/test_datasource.py

@@ -895,7 +895,8 @@ def test_publish_description(server: TSC.Server) -> None:
     ds_elem = body.find(".//datasource")
     assert ds_elem is not None
     assert ds_elem.attrib["description"] == "Sample description"
-
+
+
 def test_get_datasource_no_owner(server: TSC.Server) -> None:
     with requests_mock.mock() as m:
         m.get(server.datasources.baseurl, text=GET_NO_OWNER.read_text())

test/test_flowtask.py

@@ -1,47 +1,46 @@
-import os
-import unittest
 from datetime import time
 from pathlib import Path
 
+import pytest
 import requests_mock
 
 import tableauserverclient as TSC
-from tableauserverclient.datetime_helpers import parse_datetime
 from tableauserverclient.models.task_item import TaskItem
 
 TEST_ASSET_DIR = Path(__file__).parent / "assets"
-GET_XML_CREATE_FLOW_TASK_RESPONSE = os.path.join(TEST_ASSET_DIR, "tasks_create_flow_task.xml")
+GET_XML_CREATE_FLOW_TASK_RESPONSE = TEST_ASSET_DIR / "tasks_create_flow_task.xml"
 
 
-class TaskTests(unittest.TestCase):
-    def setUp(self):
-        self.server = TSC.Server("http://test", False)
-        self.server.version = "3.22"
+@pytest.fixture(scope="function")
+def server():
+    """Fixture to create a TSC.Server instance for testing."""
+    server = TSC.Server("http://test", False)
 
-        # Fake Signin
-        self.server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
-        self.server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"
+    # Fake signin
+    server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
+    server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"
+    server.version = "3.22"
 
-        self.baseurl = self.server.flow_tasks.baseurl
+    return server
 
-    def test_create_flow_task(self):
-        monthly_interval = TSC.MonthlyInterval(start_time=time(23, 30), interval_value=15)
-        monthly_schedule = TSC.ScheduleItem(
-            "Monthly Schedule",
-            50,
-            TSC.ScheduleItem.Type.Flow,
-            TSC.ScheduleItem.ExecutionOrder.Parallel,
-            monthly_interval,
-        )
-        target_item = TSC.Target("flow_id", "flow")
 
-        task = TaskItem(None, "RunFlow", None, schedule_item=monthly_schedule, target=target_item)
+def test_create_flow_task(server: TSC.Server) -> None:
+    monthly_interval = TSC.MonthlyInterval(start_time=time(23, 30), interval_value=15)
+    monthly_schedule = TSC.ScheduleItem(
+        "Monthly Schedule",
+        50,
+        TSC.ScheduleItem.Type.Flow,
+        TSC.ScheduleItem.ExecutionOrder.Parallel,
+        monthly_interval,
+    )
+    target_item = TSC.Target("flow_id", "flow")
 
-        with open(GET_XML_CREATE_FLOW_TASK_RESPONSE, "rb") as f:
-            response_xml = f.read().decode("utf-8")
-        with requests_mock.mock() as m:
-            m.post(f"{self.baseurl}", text=response_xml)
-            create_response_content = self.server.flow_tasks.create(task).decode("utf-8")
+    task = TaskItem("", "RunFlow", 0, schedule_item=monthly_schedule, target=target_item)
 
-        self.assertTrue("schedule_id" in create_response_content)
-        self.assertTrue("flow_id" in create_response_content)
+    response_xml = GET_XML_CREATE_FLOW_TASK_RESPONSE.read_text()
+    with requests_mock.mock() as m:
+        m.post(f"{server.flow_tasks.baseurl}", text=response_xml)
+        create_response_content = server.flow_tasks.create(task).decode("utf-8")
+
+    assert "schedule_id" in create_response_content
+    assert "flow_id" in create_response_content