Skip to content
Open
Show file tree
Hide file tree
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
3 changes: 0 additions & 3 deletions noxfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -493,9 +493,6 @@ def functions_e2e(session: nox.Session) -> None:
),
)
try:
session.env[
"AzureFunctionsJobHost__extensions__durableTask__hubName"
] = _new_test_namespace("nox")
session.env["E2E_PAYLOAD_CONTAINER"] = _new_test_namespace("payloads")
session.install("-r", "requirements.txt")
_install_packages(session, editable=True)
Expand Down
19 changes: 16 additions & 3 deletions tests/azure-functions-durable/e2e/_harness.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

import json
import os
import re
import shutil
import signal
import socket
Expand Down Expand Up @@ -259,7 +260,8 @@ class FunctionApp:

Use as a context manager. On entry it starts the host and blocks until the
app's ``/api/ping`` route responds; on exit it terminates the whole process
group and surfaces the captured host log if startup failed.
group and surfaces the captured host log if startup failed. Each instance
uses a unique task hub so sample hosts cannot consume each other's work.
"""

# ``func start`` binds the HTTP port itself, some time after we pick a free
Expand All @@ -274,6 +276,7 @@ def __init__(self, app_name: str, port: Optional[int] = None):
raise FileNotFoundError(f"Sample app not found: {self.app_dir}")
self.port = port or find_free_port()
self.base_url = f"http://127.0.0.1:{self.port}"
self._task_hub_name = f"e2e{uuid.uuid4().hex}"
self._process: Optional[subprocess.Popen[str]] = None
self._log_path = self.app_dir / "_func_host.log"

Expand Down Expand Up @@ -331,6 +334,7 @@ def start(self) -> None:

def _build_env(self) -> dict[str, str]:
env = os.environ.copy()
env["AzureFunctionsJobHost__extensions__durableTask__hubName"] = self._task_hub_name
# Start the host exactly as a developer would after activating the
# app's OWN virtual environment (``<app>/.venv``). The environment MUST
# live inside the app directory: the Functions Python worker only
Expand Down Expand Up @@ -416,7 +420,7 @@ def read_host_log(self) -> str:
return self._read_log()

def wait_for_host_log(self, text: str, timeout: float = 30) -> str:
"""Wait until captured Functions host output contains ``text``."""
"""Wait for ``text`` and return its host record, not its physical line."""
deadline = time.time() + timeout
log = self._read_log()
while text not in log and time.time() < deadline:
Expand All @@ -426,7 +430,16 @@ def wait_for_host_log(self, text: str, timeout: float = 30) -> str:
raise TimeoutError(
f"Functions host log did not contain {text!r} within {timeout}s.\n"
f"{log}")
return next(line for line in log.splitlines() if text in line)
# Host stdout/stderr can concatenate timestamped records without a newline.
return next(
record
for line in log.splitlines()
for record in re.split(
r"(?=\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z\] )",
line,
)
if text in record
)

def stop(self) -> None:
proc = self._process
Expand Down
28 changes: 28 additions & 0 deletions tests/azure-functions-durable/e2e/test_host_isolation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

import os
import re

import pytest

from ._harness import FunctionApp


def test_sample_hosts_use_independent_task_hubs(monkeypatch: pytest.MonkeyPatch) -> None:
setting = "AzureFunctionsJobHost__extensions__durableTask__hubName"
monkeypatch.setenv(setting, "sharedhub")
apps = [FunctionApp(name) for name in (
"v1_style", "dtask_style", "tracing", "v1_style")]
environments = [
app._build_env() # pyright: ignore[reportPrivateUsage]
for app in apps
]
task_hubs = [env[setting] for env in environments]

assert len(set(task_hubs)) == len(apps)
assert all(re.fullmatch(r"[A-Za-z][A-Za-z0-9]{2,49}", hub) for hub in task_hubs)
assert "sharedhub" not in task_hubs
assert os.environ[setting] == "sharedhub"
for app, env in zip(apps, environments):
assert app._build_env()[setting] == env[setting] # pyright: ignore[reportPrivateUsage]
57 changes: 57 additions & 0 deletions tests/azure-functions-durable/e2e/test_host_log_records.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

from pathlib import Path

import pytest

from ._harness import FunctionApp


@pytest.mark.parametrize("message", [
"Starting new 'activity_chain' instance with ID = 'test-instance'.",
"test-instance: Orchestration activity_chain completed with status: COMPLETED",
])
@pytest.mark.parametrize(("before", "after"), [
("", ""),
("", "[2026-11-02T19:42:58.870Z] Request GET http://127.0.0.1:10002/"),
("[2026-11-02T19:42:58.869Z] Previous record", ""),
("[2026-11-02T19:42:58.869Z] Previous record",
"[2026-11-02T19:42:58.871Z] Next record"),
("[2026-11-02T19:42:58Z] Previous record",
"[2026-11-02T19:42:58.8701234Z] Next record"),
("[2026-11-02T19:42:58.869Z] Previous record\n",
"\n[2026-11-02T19:42:58.871Z] Next record"),
("[2026-11-02T19:42:58.869Z] Previous record\r\n",
"\r\n[2026-11-02T19:42:58.871Z] Next record"),
])
def test_wait_for_host_log_extracts_timestamped_record(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
message: str, before: str, after: str) -> None:
app = FunctionApp("dtask_style")
log_path = tmp_path / "host.log"
monkeypatch.setattr(app, "_log_path", log_path)
expected = f"[2026-11-02T19:42:58.870Z] {message}"
log_path.write_text(before + expected + after, encoding="utf-8")

record = app.wait_for_host_log(message, timeout=0)

assert record == expected


@pytest.mark.parametrize("output", [
"SDK message",
"[2026-11-02T19:42:58.870Z]\nSDK message",
"[2026-11-02T19:42:58.870Z] Unrelated record SDK message",
"[2026-11-02T19:42:58.870Z] SDK message unexpected suffix",
])
def test_wait_for_host_log_preserves_nonmatching_records(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, output: str) -> None:
app = FunctionApp("dtask_style")
log_path = tmp_path / "host.log"
monkeypatch.setattr(app, "_log_path", log_path)
log_path.write_text(output, encoding="utf-8")

record = app.wait_for_host_log("SDK message", timeout=0)

assert record == output.splitlines()[-1]
4 changes: 2 additions & 2 deletions tests/azure-functions-durable/e2e/test_v1_client_e2e.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,7 @@
Covers ``suspend``/``resume``, ``restart``, ``get_status_all``,
``get_status_by``, ``purge_instance_history_by``,
``create_http_management_payload`` / ``get_client_response_links``,
``wait_for_completion_or_create_check_status_response``, and the ``rewind``
NotImplementedError stub.
``wait_for_completion_or_create_check_status_response``, and ``rewind``.
"""

from datetime import datetime, timedelta, timezone
Expand Down Expand Up @@ -110,6 +109,7 @@ def test_rewind(v1_app):
instance_id = v1_app.start_orchestration("rewind_target")
failed = v1_app.wait_for_completion(instance_id)
assert failed["runtimeStatus"] == "Failed"
assert "failing on first attempt (rewind to retry)" in failed["output"]

result = http_request("POST", f"{v1_app.base_url}/api/rewind/{instance_id}")
assert result.status == 202
Expand Down
Loading