diff --git a/docs/_build/doctrees/auth.doctree b/docs/_build/doctrees/auth.doctree index b550b078..d0f636f1 100644 Binary files a/docs/_build/doctrees/auth.doctree and b/docs/_build/doctrees/auth.doctree differ diff --git a/docs/_build/doctrees/file.doctree b/docs/_build/doctrees/file.doctree index bd794497..a61a720c 100644 Binary files a/docs/_build/doctrees/file.doctree and b/docs/_build/doctrees/file.doctree differ diff --git a/docs/_build/doctrees/index.doctree b/docs/_build/doctrees/index.doctree index 4d795d0b..1a4ad96e 100644 Binary files a/docs/_build/doctrees/index.doctree and b/docs/_build/doctrees/index.doctree differ diff --git a/docs/_build/doctrees/indexing.doctree b/docs/_build/doctrees/indexing.doctree index 3fb622e2..a235ef7d 100644 Binary files a/docs/_build/doctrees/indexing.doctree and b/docs/_build/doctrees/indexing.doctree differ diff --git a/docs/_build/doctrees/jobs.doctree b/docs/_build/doctrees/jobs.doctree index 7e2ece18..dcc8a87d 100644 Binary files a/docs/_build/doctrees/jobs.doctree and b/docs/_build/doctrees/jobs.doctree differ diff --git a/docs/_build/doctrees/metadata.doctree b/docs/_build/doctrees/metadata.doctree index 53c986e5..e9615124 100644 Binary files a/docs/_build/doctrees/metadata.doctree and b/docs/_build/doctrees/metadata.doctree differ diff --git a/docs/_build/doctrees/object.doctree b/docs/_build/doctrees/object.doctree index c791cd27..baa38775 100644 Binary files a/docs/_build/doctrees/object.doctree and b/docs/_build/doctrees/object.doctree differ diff --git a/docs/_build/doctrees/query.doctree b/docs/_build/doctrees/query.doctree index b5695fff..a016b390 100644 Binary files a/docs/_build/doctrees/query.doctree and b/docs/_build/doctrees/query.doctree differ diff --git a/docs/_build/doctrees/submission.doctree b/docs/_build/doctrees/submission.doctree index 70bc7562..4360ba5a 100644 Binary files a/docs/_build/doctrees/submission.doctree and b/docs/_build/doctrees/submission.doctree differ diff --git a/docs/_build/doctrees/tools.doctree b/docs/_build/doctrees/tools.doctree index 81b49913..5e050a6f 100644 Binary files a/docs/_build/doctrees/tools.doctree and b/docs/_build/doctrees/tools.doctree differ diff --git a/docs/_build/doctrees/tools/drs_pull.doctree b/docs/_build/doctrees/tools/drs_pull.doctree index c0dea75b..c889324a 100644 Binary files a/docs/_build/doctrees/tools/drs_pull.doctree and b/docs/_build/doctrees/tools/drs_pull.doctree differ diff --git a/docs/_build/doctrees/tools/indexing.doctree b/docs/_build/doctrees/tools/indexing.doctree index 226661ff..00edc01a 100644 Binary files a/docs/_build/doctrees/tools/indexing.doctree and b/docs/_build/doctrees/tools/indexing.doctree differ diff --git a/docs/_build/doctrees/tools/metadata.doctree b/docs/_build/doctrees/tools/metadata.doctree index 2e360db3..19f7300d 100644 Binary files a/docs/_build/doctrees/tools/metadata.doctree and b/docs/_build/doctrees/tools/metadata.doctree differ diff --git a/docs/_build/doctrees/wss.doctree b/docs/_build/doctrees/wss.doctree index 95fd0aac..29e15ff1 100644 Binary files a/docs/_build/doctrees/wss.doctree and b/docs/_build/doctrees/wss.doctree differ diff --git a/gen3/workflow.py b/gen3/workflow.py new file mode 100644 index 00000000..1420daeb --- /dev/null +++ b/gen3/workflow.py @@ -0,0 +1,443 @@ +""" +Contains class for interacting with Gen3's Workflow Service. +""" + +import asyncio +from dataclasses import dataclass +import time + +import aiohttp +import backoff +import requests +from cdislogging import get_logger + +from gen3.auth import Gen3Auth +from gen3.utils import DEFAULT_BACKOFF_SETTINGS, raise_for_status_and_print_error + +logging = get_logger(__name__) + + +class Gen3Workflow: + """ + A class for interacting with the Gen3 Workflow service. + + Examples: + This generates the Gen3Workflow class pointed at the sandbox commons while + using the credentials.json downloaded from the commons profile page. + + >>> auth = Gen3Auth(refresh_file="credentials.json") + ... workflow = Gen3Workflow(auth) + + Attributes: + endpoint (str): public endpoint for reading/querying workflow - only necessary if auth_provider not provided + auth_provider (Gen3Auth): auth manager + """ + + def __init__( + self, + endpoint=None, + auth_provider=None, + service_location="workflows", + ): + """ + Initialization for instance of the class to setup basic endpoint info. + + Args: + endpoint (str): URL for a Data Commons that has workflow service deployed + auth_provider (Gen3Auth, optional): Gen3Auth class to handle passing your + token, required for admin endpoints + service_location (str, optional): deployment location relative to the + endpoint provided + """ + # legacy interface required endpoint as 1st arg + if endpoint and isinstance(endpoint, Gen3Auth): + auth_provider = endpoint + endpoint = None + if auth_provider and isinstance(auth_provider, Gen3Auth): + endpoint = auth_provider.endpoint + endpoint = endpoint.strip("/") + # if running locally, gen3-workflow is deployed by itself without a location relative + # to the commons + if "http://localhost" in endpoint: + service_location = "" + + if not endpoint.endswith(service_location): + endpoint += "/" + service_location + + self.endpoint = endpoint.rstrip("/") + # TODO: need TES URL and S3 URL + self.ui_url = self.endpoint + "/ui" + # TODO: include "/task" at the end and replace settings in methods + # change to tes_task_url + self.tes_url = self.endpoint + "/ga4gh/tes/v1" + self._auth_provider = auth_provider + + def is_healthy(self): + """ + Return if is healthy or not + + Returns: + bool: True if healthy + """ + try: + response = requests.get( + self.endpoint + "/_status", auth=self._auth_provider + ) + response.raise_for_status() + except Exception as exc: + logging.error(exc) + return False + + return response.json().get("status") == "OK" + + @backoff.on_exception(backoff.expo, Exception, **DEFAULT_BACKOFF_SETTINGS) + def get_version(self): + """ + Return the version + + Returns: + str: the version + """ + response = requests.get(self.endpoint + "/_version", auth=self._auth_provider) + raise_for_status_and_print_error(response) + return response.text + + # Create task + # https://github.com/uc-cdis/gen3-workflow/blob/3cc65d4bb175a17a3e1e391d3082577cabeb3a6c/scripts/performance_test/run_tes_task.py#L19-L28 + def create_task(self, body: dict) -> str: + """ + Create a TES task. + + Args: + body (dict): The json of the TES workflow. + + Returns: + str: the task_id of the created task + """ + + tes_task_url = f"{self.tes_url}/tasks" + response = requests.post( + tes_task_url, + json=body, + # Do like the get_page + auth=self._auth_provider, + timeout=60, + ) + expected_status = 200 + if response.status_code != expected_status: + raise_for_status_and_print_error(response) + data = response.json() + return data.get("id") + + @backoff.on_exception(backoff.expo, Exception, **DEFAULT_BACKOFF_SETTINGS) + async def async_get_tes_task( + self, + task_id: str, + user: str, + expected_status: int = 200, + _ssl=None, + ): + """ + Asynchronous function to get a TES task. + + Args: + task_id (str): ID of the TES task + user (str): same truthy flag as the sync version; if falsy, no auth + header is sent + expected_status (int): Expected HTTP status_code, default=200 + _ssl (None, optional): passed to aiohttp's `ssl=` argument + + Returns: + dict: TES task + """ + url = f"{self.tes_url}/tasks/{task_id}?view=FULL" + + # aiohttp can't use requests-style auth objects, so set the JWT header + # manually (same approach as the indexd async methods) + headers = ( + {"Authorization": self._auth_provider._get_auth_value()} if user else {} + ) + + async with aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=60) + ) as session: + async with session.get(url, headers=headers, ssl=_ssl) as response: + if response.status != expected_status: + body = await response.text() + logging.error(f"Unexpected status {response.status}: {body}") + response.raise_for_status() + return await response.json() + + def get_tes_task(self, task_id: str, user: str, expected_status=200): + """ + Get task for task_id. + + Args: + task_id (str): ID of the TES task to poll. + user (str): User_id that matches credentials. + expected_status (int): Expected HTTP status_code, default=200. + + Returns: + Task object + """ + + tes_task_url = f"{self.tes_url}/tasks/{task_id}?view=FULL" + + logging.info(f"Getting task for {task_id}") + + response = requests.get( + url=tes_task_url, + **({"auth": self._auth_provider} if user else {"headers": {}}), + timeout=60, + ) + if response.status_code != expected_status: + raise_for_status_and_print_error(response) + + return response.json() + + async def async_poll_task_until_expected_state( + self, + task_id: str, + user: str, + expected_final_state: str = "COMPLETE", + max_retries: int = 60, + poll_interval: int = 5, + _ssl=None, + ) -> dict: + """ + Asynchronously poll the TES task until it reaches the specified expected + state or exceeds max retries. Use with asyncio.gather to wait on many + tasks concurrently. + + Args: + task_id (str): ID of the TES task to poll + user (str): same truthy flag as the sync version + expected_final_state (str): Final state this task is expected to reach + max_retries (int): Maximum number of polling attempts + poll_interval (int): Seconds between polling attempts + _ssl (None, optional): passed to aiohttp's `ssl=` argument + + Returns: + dict: the task, once it reaches expected_final_state + """ + transient_states = {"QUEUED", "INITIALIZING", "RUNNING"} + final_states = { + "COMPLETE", + "FAILED", + "EXECUTOR_ERROR", + "CANCELED", + "SYSTEM_ERROR", + } + logging.info(f"Polling task {task_id} until {expected_final_state}") + + state = None + task_info = None + + # NOTE: no @backoff on this method on purpose. It already loops, and + # retrying the whole poll would restart the clock and mask real + # task failures. Transient HTTP errors are retried inside + # async_get_tes_task. + for attempt in range(1, max_retries + 1): + task_info = await self.async_get_tes_task( + task_id=task_id, + user=user, + expected_status=200, + _ssl=_ssl, + ) + state = task_info.get("state") + + if state == expected_final_state: + logging.info(f"TES task reached final state '{state}'") + return task_info + + if state in ("SYSTEM_ERROR", "EXECUTOR_ERROR"): + error_message = f"Task failed with error state '{state}'" + logging.error(error_message) + raise Exception(error_message) + if state in final_states: + error_message = ( + f"TES task reached a final state that is not " + f"'{expected_final_state}'. Final state: {state}, " + f"Response: {task_info}" + ) + logging.warning(error_message) + raise Exception(error_message) + if state not in transient_states: + logging.warning( + f"Unexpected TES task state '{state}' encountered. " + f"Response: {task_info}" + ) + + if attempt < max_retries: + logging.info( + f"Attempt {attempt} of {max_retries}: Task state is " + f"'{state}', retrying after {poll_interval} seconds..." + ) + await asyncio.sleep(poll_interval) + + raise Exception( + f"TES task did not reach a final state in time. " + f"Last known state: {state}, Response: {task_info}" + ) + + def poll_task_until_expected_state( + self, + task_id: str, + user: str, + expected_final_state: str = "COMPLETE", + max_retries: int = 60, + poll_interval: int = 5, + ) -> dict: + """ + Poll the TES task status until it reaches the specified expected state or exceeds max retries. + + Args: + task_id (str): ID of the TES task to poll. + user (str): User_id that matches credentials. + expected_final_state (str): Final state, which this task is expected to return + (e.g., one of {"COMPLETE", "FAILED"}). + max_retries (int): Maximum number of polling attempts before giving up. + poll_interval (int): Time in seconds between polling attempts. + + Returns: + dict: Final task information once it reaches expected_final_state. + """ + + # TODO: maybe move these higher up for accessing from tests, etc. + transient_states = {"QUEUED", "INITIALIZING", "RUNNING"} + final_states = { + "COMPLETE", + "FAILED", + "EXECUTOR_ERROR", + "CANCELED", + "SYSTEM_ERROR", + } + state = None + task_info = None + + logging.info(f"Polling task {task_id} until {expected_final_state}") + for attempt in range(1, max_retries + 1): + # TODO: might wrap in try-except here. + task_info = self.get_tes_task( + task_id=task_id, + user=user, + expected_status=200, + ) + state = task_info.get("state") + + if state == expected_final_state: + logging.info(f"TES task reached final state '{state}'") + return task_info + + # TODO: can't use raise_for_status - these are all using task dicts. + if state in ["SYSTEM_ERROR", "EXECUTOR_ERROR"]: + error_message = f"Task failed with error state '{state}'" + logging.error(error_message) + raise Exception(error_message) + if state in final_states: + error_message = ( + f"TES task reached a final state, that is not '{expected_final_state}'." + f" Final state: {state}, Response: {task_info}" + ) + logging.warning(error_message) + raise Exception(error_message) + if state not in transient_states: + message = f"Unexpected TES task state '{state}' encountered. Response: {task_info}" + logging.warning(message) + + if attempt < max_retries: + logging.info( + f"Attempt {attempt} of {max_retries}: Task state is '{state}'," + f" retrying after {poll_interval} seconds..." + ) + time.sleep(poll_interval) + + raise Exception( + f"TES task did not reach a final state in time. Last known state: {state}, Response: {task_info}" + ) + + def get_tasks_for_page(self, user: str, page=None) -> dict: + """ + Get a list of tasks for page for user + + Args: + user (str): User_id that matches credentials. + page (str): Page token, default=None. + + Returns a page of tasks for user + { + "next_page_token": "", + "tasks": [] + } + """ + + tes_task_url = f"{self.tes_url}/tasks/?view=MINIMAL" + if page: + tes_task_url += f"&page_token={page}" + + response = requests.get( + url=tes_task_url, + **({"auth": self._auth_provider} if user else {"headers": {}}), + timeout=60, + ) + expected_status = 200 + if response.status_code != expected_status: + raise_for_status_and_print_error(response) + + return response.json() + + def get_all_tasks_for_user(self, user: str) -> dict: + """ + Get all tasks for user. + + Args: + user (str): User_id that matches credentials. + + Returns JSON with {"tasks": []} + """ + + all_tasks = [] + next_page_token = None + + while True: + logging.info( + f"Calling get_tasks_for_page with page_token = {next_page_token}" + ) + page_data = self.get_tasks_for_page(user=user, page=next_page_token) + all_tasks.extend(page_data.get("tasks", [])) + + next_page_token = page_data.get("next_page_token") + if not next_page_token: + break + + return {"tasks": all_tasks} + + def cancel_tes_task(self, task_id: str, user: str, expected_status=200): + """ + Cancel task with task_id. + + Args: + task_id (str): ID of the TES task to poll. + user (str): User_id that matches credentials. + expected_status (int): Expected HTTP status_code, default=200. + + Returns: + Task object + """ + + tes_task_url = f"{self.tes_url}/tasks/{task_id}:cancel" + + logging.info(f"Cancelling task for {task_id}") + + response = requests.post( + url=tes_task_url, + **({"auth": self._auth_provider} if user else {"headers": {}}), + timeout=60, + ) + if response.status_code != expected_status: + raise_for_status_and_print_error(response) + + return response.json() + + # Perform S3 action + # https://github.com/uc-cdis/gen3-code-vigil/blob/20bb83b/gen3-integration-tests/services/gen3workflow.py#L152 diff --git a/tests/test_workflow.py b/tests/test_workflow.py new file mode 100644 index 00000000..3a7ea10b --- /dev/null +++ b/tests/test_workflow.py @@ -0,0 +1,338 @@ +import asyncio +import logging +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest +import requests +from requests.exceptions import HTTPError + +from gen3.workflow import Gen3Workflow +from gen3.tools.metadata import async_ingest_metadata_manifest +from gen3.utils import get_or_create_event_loop_for_thread + +CURRENT_DIR = os.path.dirname(os.path.realpath(__file__)) + +TEST_ID = "some-test-id" +TEST_BODY = { + "name": "Hello-World", + "executors": [{"image": "quay.io/nextflow/bash", "command": ["echo hello world!"]}], +} +TEST_TASK = { + "creation_time": "2026-08-21T21:11:06.198654625Z", + "description": "", + "executors": [ + { + "command": ["echo hello world!"], + } + ], + "id": TEST_ID, + "inputs": [], + "logs": [], + "name": "Hello-World", + "outputs": [], + "state": "COMPLETE", + "tags": {"_AUTHZ": f"/services/workflow/gen3-workflow/tasks/123/{TEST_ID}"}, + "volumes": [], +} + + +@patch("gen3.workflow.requests.get") +def test_is_healthy(requests_mock): + """ + Test is healthy response + """ + workflow = Gen3Workflow("https://example.com") + + def _mock_request(url, **kwargs): + assert url.endswith("/_status") + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 200 + mocked_response.json.return_value = { + "status": "OK", + "timestamp": "2020-03-13T15:23:53.765568+00:00", + } + mocked_response.raise_for_status.side_effect = lambda *args: None + + return mocked_response + + requests_mock.side_effect = _mock_request + + response = workflow.is_healthy() + + assert response + + +@patch("gen3.workflow.requests.get") +def test_is_not_healthy(requests_mock): + """ + Test is not healthy response + """ + workflow = Gen3Workflow("https://example.com") + + def _mock_request(url, **kwargs): + assert url.endswith("/_status") + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 500 + mocked_response.text = "Not Healthy" + mocked_response.json.return_value = {} + mocked_response.raise_for_status.side_effect = HTTPError("uh oh") + + return mocked_response + + requests_mock.side_effect = _mock_request + + response = workflow.is_healthy() + + assert not response + + +@patch("gen3.workflow.requests.get") +def test_get_version(requests_mock): + """ + Test getting version + """ + workflow = Gen3Workflow("https://example.com") + + def _mock_request(url, **kwargs): + assert url.endswith("/_version") + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 200 + mocked_response.text = "1.2.0" + mocked_response.json.return_value = {} + mocked_response.raise_for_status.side_effect = lambda *args: None + + return mocked_response + + requests_mock.side_effect = _mock_request + + response = workflow.get_version() + + assert response + + +@patch("gen3.workflow.requests.post") +def test_create_task(requests_mock): + """ + Test creating a task + """ + workflow = Gen3Workflow("https://example.com") + + test_id = TEST_ID + + def _mock_request(url, **kwargs): + assert url.endswith("/tasks") + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 200 + mocked_response.json.return_value = {"id": test_id} + mocked_response.raise_for_status.side_effect = lambda *args: None + + return mocked_response + + requests_mock.side_effect = _mock_request + + response = workflow.create_task(TEST_BODY) + + assert response + assert response == test_id + assert requests_mock.call_args.kwargs["json"] == TEST_BODY + + +@pytest.mark.parametrize( + "test_user", + ["main@example.org", None], +) +@patch("gen3.workflow.requests.get") +def test_get_task(requests_mock, test_user): + """ + Test getting a task + """ + workflow = Gen3Workflow("https://example.com") + + test_id = TEST_TASK.get("id") + + def _mock_request(url, **kwargs): + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 200 + mocked_response.json.return_value = TEST_TASK + mocked_response.raise_for_status.side_effect = lambda *args: None + + return mocked_response + + requests_mock.side_effect = _mock_request + + response = workflow.get_tes_task(task_id=test_id, user=test_user) + + assert response + assert response == TEST_TASK + print(f"CALL ARGS {requests_mock.call_args}") + assert requests_mock.call_args.kwargs["url"].endswith(f"/tasks/{test_id}?view=FULL") + if test_user is not None: + # We have a user so we should have an "auth" arg but no "headers" arg + assert requests_mock.call_args.kwargs["auth"] is None + assert "headers" not in requests_mock.call_args.kwargs + else: + assert "auth" not in requests_mock.call_args.kwargs + assert requests_mock.call_args.kwargs["headers"] == {} + + +# TODO: parametrize final state (not error states) +@patch("gen3.workflow.requests.get") +def test_poll_task_until_expected_state(requests_mock): + """ + Test getting a task + """ + workflow = Gen3Workflow("https://example.com") + + test_id = TEST_TASK.get("id") + test_user = "main@example.org" + expected_state = "COMPLETE" + + def create_mock_response(state): + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 200 + mocked_response.json.return_value = {**TEST_TASK, "state": state} + mocked_response.raise_for_status.side_effect = lambda *args: None + + return mocked_response + + mock_responses = [ + create_mock_response(state="RUNNING"), + create_mock_response(state=expected_state), + ] + requests_mock.side_effect = mock_responses + + response = workflow.poll_task_until_expected_state( + task_id=test_id, + user=test_user, + expected_final_state=expected_state, + poll_interval=1, + ) + + assert response + assert response == TEST_TASK + assert response.get("state") == expected_state + assert requests_mock.call_count >= len(mock_responses) + assert requests_mock.call_args.kwargs["url"].endswith(f"/tasks/{test_id}?view=FULL") + # TODO: check for "auth" and "headers" + + +@patch("gen3.workflow.requests.get") +def test_poll_task_never_reaches_expected_state(requests_mock): + """ + Test polling a task that does not get to final expected state after max_retries + """ + workflow = Gen3Workflow("https://example.com") + + test_id = TEST_TASK.get("id") + test_user = "main@example.org" + expected_state = "COMPLETE" + + def create_mock_response(state): + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 200 + mocked_response.json.return_value = {**TEST_TASK, "state": state} + mocked_response.raise_for_status.side_effect = lambda *args: None + + return mocked_response + + # test with responses that do not include expected_state + mock_responses = [ + create_mock_response(state="QUEUED"), + create_mock_response(state="INITIALIZING"), + create_mock_response(state="RUNNING"), + create_mock_response(state="RUNNING"), + ] + requests_mock.side_effect = mock_responses + + max_retries = len(mock_responses) - 1 + last_response = mock_responses[-1] + expected_message = ( + "TES task did not reach a final state in time." + f' Last known state: {last_response.json().get("state")},' + f" Response: {last_response.json()}" + ) + + with pytest.raises(Exception) as err: + workflow.poll_task_until_expected_state( + task_id=test_id, + user=test_user, + expected_final_state=expected_state, + max_retries=max_retries, + poll_interval=1, + ) + assert expected_message in str(err.value) + assert last_response.json().get("state") != expected_state + + +# Test failed state or incorrect final state +@pytest.mark.parametrize( + "test_final_state, expected_state, message", + [ + ("SYSTEM_ERROR", "COMPLETE", "Task failed with error state 'SYSTEM_ERROR'"), + ("EXECUTOR_ERROR", "COMPLETE", "Task failed with error state 'EXECUTOR_ERROR'"), + ( + "FAILED", + "COMPLETE", + "TES task reached a final state, that is not 'COMPLETE'.", + ), + ( + "CANCELED", + "COMPLETE", + "TES task reached a final state, that is not 'COMPLETE'.", + ), + ("CANCELED", "FAILED", "TES task reached a final state, that is not 'FAILED'."), + ], +) +@patch("gen3.workflow.requests.get") +def test_poll_task_with_error_states( + requests_mock, test_final_state, expected_state, message +): + """ + Test polling a task that does not get to final expected state after max_retries + """ + workflow = Gen3Workflow("https://example.com") + + test_id = TEST_TASK.get("id") + test_user = "main@example.org" + + def create_mock_response(state): + + mocked_response = MagicMock(requests.Response) + mocked_response.status_code = 200 + mocked_response.json.return_value = {**TEST_TASK, "state": state} + mocked_response.raise_for_status.side_effect = lambda *args: None + + return mocked_response + + # test with responses that do not include expected_state + mock_responses = [ + create_mock_response(state="QUEUED"), + create_mock_response(state="INITIALIZING"), + create_mock_response(state=test_final_state), + ] + requests_mock.side_effect = mock_responses + + last_response = mock_responses[-1] + + with pytest.raises(Exception) as err: + # Use default max_retries (60) + workflow.poll_task_until_expected_state( + task_id=test_id, + user=test_user, + expected_final_state=expected_state, + poll_interval=1, + ) + assert message in str(err.value) + assert last_response.json().get("state") != expected_state + + +# DO a test for unknown state