https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> why does a test that runs a task cras...
# ask-marvin
a
@Marvin why does a test that runs a task crash when run in a github action?
m
✅ thought for 125.1 seconds
A few common reasons: Most likely causes in CI/GitHub Actions - Event loop issues — especially if the test uses async code, 
asyncio.run()
, or 
pytest-asyncio
- Process/subprocess differences — CI Linux runners can behave differently than local machines, especially with multiprocessing - Prefect API/server setup — tests may implicitly rely on an ephemeral API or local server that is not available/ready in CI - Environment/settings differences — 
PREFECT_API_URL
, profiles, temp settings, or secrets may differ in Actions - Timing/resource issues — CI is slower, so race conditions and short timeouts show up there Prefect-specific things to check 1. Are you running a task directly, or inside a flow? - Prefect tasks are safest to test either: - as plain Python by calling the underlying function logic, or - inside a flow run if you want orchestration behavior 2. If async: avoid 
asyncio.run()
inside pytest async tests - Prefer:
Copy code
@pytest.mark.asyncio
async def test_task():
    result = await my_async_fn()
3. If using task runners/processes: prefer threads in tests - Process-based execution is more likely to fail in CI than locally. 4. If the test touches the Prefect API: verify your CI env - Check whether 
PREFECT_API_URL
is set - If you expect ephemeral mode, make sure the test is actually configured for it 5. Increase timeouts in CI - A test that passes locally can fail in GitHub Actions just because the runner is slower Typical fixes - Use 
@pytest.mark.asyncio
for async tests - Don’t call 
asyncio.run()
from within an already-async test - Avoid process-based task runners in unit tests - Wrap Prefect settings with 
temporary_settings(...)
in the test - If using a server, wait until it’s healthy before running assertions - Make CI timeouts more generous Very common anti-pattern
Copy code
# fragile in CI
def test_task():
    result = my_task.submit().result(timeout=2)
    assert result == 1
Better:
Copy code
from prefect import flow, task

@task
def my_task():
    return 1

@flow
def test_flow():
    return my_task()

def test_task():
    assert test_flow() == 1
If you do want to test 
.submit()
behavior, it helps to share: - the test code - the full GitHub Actions error/traceback - whether the task is sync or async - whether it runs inside a flow - whether you set 
PREFECT_API_URL
If you paste the failing test and traceback, I can usually pinpoint the exact cause quickly. Relevant docs: - Prefect docs - Deployments and execution model
a
@Marvin heres the error message self = <prefect.client.orchestration.SyncPrefectClient object at 0x7f41a51f9710> def raise_for_api_version_mismatch(self) -> None: # Cloud is always compatible as a server if self.server_type == ServerType.CLOUD: return try:
api_version = self.api_version()
^^^^^^^^^^^^^^^^^^ /opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/site-packages/prefect/client/orchestration/__init__.py1476 _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ /opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/site-packages/prefect/client/orchestration/__init__.py1451 in api_version res = self._client.get("/admin/version") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ /opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/site-packages/httpx/_client.py1053 in get return self.request( /opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/site-packages/httpx/_client.py825 in request return self.send(request, auth=auth, follow_redirects=follow_redirects) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ /opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/site-packages/prefect/client/base.py677 in send response.raise_for_status() _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <Response [500 Internal Server Error]> def raise_for_status(self) -> Response: """ Raise an exception if the response contains an HTTPStatusError. The 
PrefectHTTPStatusError
contains useful additional information that is not contained in the 
HTTPStatusError
. """ try: return super().raise_for_status() except HTTPStatusError as exc:
raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause
E prefect.exceptions.PrefectHTTPStatusError: Server error '500 Internal Server Error' for url 'http://127.0.0.1:8949/api/admin/version' - Response: {'exception_message': 'Internal Server Error'} - For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500 /opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/site-packages/prefect/client/base.py164 PrefectHTTPStatusError The above exception was the direct cause of the following exception: mock_teams_webhook = TeamsWebhook(webhook_url=SecretStr('**********')) def test_alert_formats_message_correctly(mock_teams_webhook): """ Test: alert formats message correctly. Verifies that send_teams_alert formats the Teams MessageCard with all required fields and proper structure. """ mock_response = MagicMock() mock_response.status_code = 200 mock_response.raise_for_status = MagicMock() with patch("src.tasks.alerting.requests.post", return_value=mock_response) as mock_post:
result = send_teams_alert(
webhook=mock_teams_webhook, title="Pipeline Failed", message="COPY INTO failed", facts=[ {"name": "Domain", "value": "demo"}, {"name": "Environment", "value": "test"}, ], status="error", flow_run_url="https://app.prefect.cloud/runs/123" ) tests/test_alerting.py38
m
✅ thought for 23.1 seconds
Thanks — this traceback is actually very revealing. What’s happening Your test is not just formatting a Teams message. During 
send_teams_alert(...)
, something is causing Prefect to create a 
SyncPrefectClient
, and that client is trying to talk to a local API at: 
<http://127.0.0.1:8949/api/admin/version>
That request returns 500, and then Prefect raises the version-mismatch/client error. So the failure is not primarily GitHub Actions itself — it’s that in CI your test ends up hitting a Prefect API server, and that server is unhealthy or misconfigured. Why this happens Usually one of these is true: 1. Your task/function calls Prefect runtime/client APIs internally - e.g. 
get_run_logger()
- 
prefect.runtime.flow_run.*
- 
prefect.runtime.task_run.*
- 
get_client()
- block loading or artifact/event APIs 2. The code is decorated with 
@task
and you are invoking the task object in a way that triggers Prefect orchestration - instead of testing the pure Python function body 3. A local ephemeral API is being started automatically in the test environment - and in GitHub Actions it starts badly, incompletely, or with missing dependencies/state Most likely root cause here Because the URL is 
127.0.0.1:8949
, this looks like an ephemeral/local Prefect API rather than Prefect Cloud. That means your test likely depends on Prefect runtime context even though it looks like a simple unit test. What I’d check first In 
send_teams_alert
, look for anything like:
Copy code
from prefect import task
from prefect.logging import get_run_logger
from prefect.runtime import flow_run, task_run
from prefect.client.orchestration import get_client
from prefect.blocks.core import Block
Especially 
get_run_logger()
is a common trigger when code is called outside a proper flow/task run context. Best fixes Option 1: Make this a pure unit test If 
send_teams_alert
is supposed to just build a payload and call 
<http://requests.post|requests.post>
, separate the business logic from Prefect:
Copy code
def build_teams_payload(title, message, facts, status, flow_run_url=None):
    return {
        # payload here
    }

@task
def send_teams_alert(webhook, title, message, facts, status, flow_run_url=None):
    payload = build_teams_payload(title, message, facts, status, flow_run_url)
    response = <http://requests.post|requests.post>(webhook_url, json=payload)
    response.raise_for_status()
    return True
Then test 
build_teams_payload(...)
directly. Option 2: If testing the task, run it inside a flow If 
send_teams_alert
is a Prefect task, test it through a flow:
Copy code
from prefect import flow

@flow
def test_flow(webhook):
    return send_teams_alert(
        webhook=webhook,
        title="Pipeline Failed",
        message="COPY INTO failed",
        facts=[
            {"name": "Domain", "value": "demo"},
            {"name": "Environment", "value": "test"},
        ],
        status="error",
        flow_run_url="<https://app.prefect.cloud/runs/123>",
    )

def test_alert_formats_message_correctly(mock_teams_webhook):
    mock_response = MagicMock()
    mock_response.status_code = 200
    mock_response.raise_for_status = MagicMock()

    with patch("<http://src.tasks.alerting.requests.post|src.tasks.alerting.requests.post>", return_value=mock_response) as mock_post:
        result = test_flow(mock_teams_webhook)
        assert result is True
Option 3: Disable/avoid API dependency in tests If this should not need the API, set the test environment so Prefect does not try to use one. For example, in the test: ``` from prefect.settings import PREFECT_API_URL, temporary_settings def test_alert_formats_message_correctly(mock_teams_webhook): mock_response = MagicMock() mock_response.status_code = 200 mock_response.raise_for_status = MagicMock() with temporary_settings({PREFECT_API_URL: None}): with patch("src.tasks.alerting.requests.post", return_value=mock_response) as mock_post: result = send_teams_alert(
webhook=mock_teams_webhook, title="Pipeline Failed", message="COPY INTO failed", facts=[ {"name": "Domain", "value": "demo"}, {"name": "Environment", "value": "test"}, ], status="error", flow_run_url="https://app.prefect.cloud/runs/123", )
Copy code
But note: this only helps if your code can truly run *without* Prefect API features.

*Option 4: Mock the Prefect-specific calls*
If your function calls `get_run_logger()` or runtime APIs, mock those in the unit test.

Example:
with patch("src.tasks.alerting.get_run_logger") as mock_logger: mock_logger.return_value = MagicMock() ... ``` Why CI fails but local passes Locally you may have: - a running Prefect server - a valid profile - a working ephemeral SQLite state - environment vars already set In GitHub Actions, none of that is guaranteed, so the hidden API dependency breaks. Important clue from your traceback The failure occurs before your 
<http://requests.post|requests.post>
patch matters. That means something inside 
send_teams_alert
touches Prefect before or during execution, not just the webhook request. My strongest recommendation For this specific test, don’t unit test the Prefect task object directly. Test the payload-building logic as plain Python, and keep one smaller integration test for the Prefect task/flow wrapper. If you paste the actual 
send_teams_alert
implementation, I can tell you exactly which line is triggering the Prefect client and show the minimal fix.
a
@Marvin send_teams_alert is annotated with task
m
✅ thought for 14.5 seconds
That explains it. If 
send_teams_alert
is annotated with 
@task
, then calling it directly can invoke Prefect task machinery, not just your function body. In your test environment, that machinery is trying to use a Prefect API at 
127.0.0.1:8949
, and that’s what is failing. Why this breaks A Prefect task is not just a plain function anymore. Depending on how it’s called, Prefect may: - create task run state - access runtime context - initialize a client - check API/server compatibility So this:
Copy code
result = send_teams_alert(...)
is not equivalent to calling a normal Python function. Best practice For unit tests, test the underlying function logic as plain Python. For Prefect integration behavior, test the task inside a flow. Two good options Option 1: Split the logic from the task wrapper This is the cleanest approach.
Copy code
from prefect import task
import requests

def _send_teams_alert_impl(webhook, title, message, facts, status, flow_run_url=None):
    payload = {
        "title": title,
        "text": message,
    }
    response = <http://requests.post|requests.post>(webhook.get_secret_value(), json=payload)
    response.raise_for_status()
    return True

@task
def send_teams_alert(webhook, title, message, facts, status, flow_run_url=None):
    return _send_teams_alert_impl(
        webhook=webhook,
        title=title,
        message=message,
        facts=facts,
        status=status,
        flow_run_url=flow_run_url,
    )
Then your unit test becomes:
Copy code
def test_alert_formats_message_correctly(mock_teams_webhook):
    mock_response = MagicMock()
    mock_response.status_code = 200
    mock_response.raise_for_status = MagicMock()

    with patch("<http://src.tasks.alerting.requests.post|src.tasks.alerting.requests.post>", return_value=mock_response) as mock_post:
        result = _send_teams_alert_impl(
            webhook=mock_teams_webhook,
            title="Pipeline Failed",
            message="COPY INTO failed",
            facts=[
                {"name": "Domain", "value": "demo"},
                {"name": "Environment", "value": "test"},
            ],
            status="error",
            flow_run_url="<https://app.prefect.cloud/runs/123>",
        )

    assert result is True
Option 2: If you want to test the Prefect task, run it inside a flow That gives the task proper runtime context.
Copy code
from prefect import flow

@flow
def run_alert_task(webhook):
    return send_teams_alert(
        webhook=webhook,
        title="Pipeline Failed",
        message="COPY INTO failed",
        facts=[
            {"name": "Domain", "value": "demo"},
            {"name": "Environment", "value": "test"},
        ],
        status="error",
        flow_run_url="<https://app.prefect.cloud/runs/123>",
    )

def test_alert_formats_message_correctly(mock_teams_webhook):
    mock_response = MagicMock()
    mock_response.status_code = 200
    mock_response.raise_for_status = MagicMock()

    with patch("<http://src.tasks.alerting.requests.post|src.tasks.alerting.requests.post>", return_value=mock_response) as mock_post:
        result = run_alert_task(mock_teams_webhook)

    assert result is True
What I would not recommend - Relying on direct task invocation for a unit test - Depending on an ephemeral API server in GitHub Actions for a formatting test - Working around this by standing up more Prefect infrastructure unless you truly want an integration test Short version Because 
@task
wraps the function, your “unit test” is accidentally becoming a Prefect runtime/integration test. In CI, that causes a failing API call. Recommended fix: move the message formatting + 
<http://requests.post|requests.post>
into a plain Python helper and test that helper directly. Docs: - Prefect tasks - Prefect docs If you want, I can rewrite your test file into: 1. pure unit tests for payload formatting, and 2. one Prefect flow-based integration test.
Open in Slack
PreviousNext
Powered by