Merge branch 'main' into remove_redundant_task_transition_rule

src/prefect/testing/fixtures.py

@@ -37,27 +37,17 @@ def is_port_in_use(port: int) -> bool:
 
 
 @pytest.fixture(scope="session")
-async def hosted_orion_api():
+async def hosted_api_server(unused_tcp_port_factory):
     """
-    Runs an instance of the Prefect API at a dedicated URL instead of the ephemeral
-    application. Requires a port from 2222-2227 to be available.
+    Runs an instance of the Prefect API server in a subprocess instead of the using the
+    ephemeral application.
 
     Uses the same database as the rest of the tests.
 
     Yields:
-        The connection string
+        The API URL
     """
-
-    ports = [2222 + i for i in range(5)]
-
-    while True:
-        try:
-            port = ports.pop()
-        except IndexError as exc:
-            raise RuntimeError("No ports available to run test API.") from exc
-
-        if not is_port_in_use(port):
-            break
+    port = unused_tcp_port_factory()
 
     # Will connect to the same database as normal test clients
     async with open_process(
@@ -109,12 +99,12 @@ async def hosted_orion_api():
 
 
 @pytest.fixture
-def use_hosted_orion(hosted_orion_api):
+def use_hosted_api_server(hosted_api_server):
     """
     Sets `PREFECT_API_URL` to the test session's hosted API endpoint.
     """
-    with temporary_settings({PREFECT_API_URL: hosted_orion_api}):
-        yield hosted_orion_api
+    with temporary_settings({PREFECT_API_URL: hosted_api_server}):
+        yield hosted_api_server
 
 
 @pytest.fixture

tests/cli/test_version.py

@@ -18,7 +18,7 @@ def test_version_ephemeral_server_type():
     )
 
 
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 def test_version_server_server_type():
     invoke_and_assert(
         ["version"], expected_output_contains="Server type:         server"
@@ -96,7 +96,7 @@ def test_correct_output_ephemeral_postgres(monkeypatch):
     )
 
 
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 def test_correct_output_non_ephemeral_server_type():
     version_info = prefect.__version_info__
     built = pendulum.parse(prefect.__version_info__["date"])

tests/client/test_orion_client.py

@@ -547,14 +547,14 @@ async def test_client_runs_migrations_for_ephemeral_app(enabled, monkeypatch):
 
 
 async def test_client_does_not_run_migrations_for_hosted_app(
-    hosted_orion_api, monkeypatch
+    hosted_api_server, monkeypatch
 ):
     with temporary_settings(updates={PREFECT_API_DATABASE_MIGRATE_ON_START: True}):
         mock = AsyncMock()
         monkeypatch.setattr(
             "prefect.server.database.interface.PrefectDBInterface.create_db", mock
         )
-        async with PrefectClient(hosted_orion_api):
+        async with PrefectClient(hosted_api_server):
             pass
 
     mock.assert_not_awaited()
@@ -1495,8 +1495,8 @@ def test_server_type_ephemeral(orion_client):
     assert orion_client.server_type == ServerType.EPHEMERAL
 
 
-async def test_server_type_server(hosted_orion_api):
-    async with PrefectClient(hosted_orion_api) as orion_client:
+async def test_server_type_server(hosted_api_server):
+    async with PrefectClient(hosted_api_server) as orion_client:
         assert orion_client.server_type == ServerType.SERVER
 
 

tests/engine/reliability/test_deadlocks.py

@@ -53,7 +53,7 @@ async def run():
 
 
 @pytest.mark.skip(reason="Causes a deadlock.")
-async def test_sync_task_after_async_in_async_flow(use_hosted_orion):
+async def test_sync_task_after_async_in_async_flow(use_hosted_api_server):
     @flow
     async def run():
         await async_multiply_by_two(42)

tests/infrastructure/test_base.py

@@ -58,7 +58,7 @@ class Config:
 
 
 @pytest.mark.skip(reason="Unclear failure.")
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 @pytest.mark.parametrize(
     "infrastructure_type",
     [
@@ -249,7 +249,7 @@ async def test_submission_does_not_override_existing_name(
 
 @pytest.mark.skip("Flaky test that needs investigation")
 @pytest.mark.service("docker")
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 @pytest.mark.skipif(
     (Version(MIN_COMPAT_PREFECT_VERSION) > Version(prefect.__version__.split("+")[0])),
     reason=f"Expected breaking change in next version: {MIN_COMPAT_PREFECT_VERSION}",

tests/infrastructure/test_docker_container.py

@@ -456,9 +456,9 @@ def test_network_mode_defaults_to_none_if_api_url_cannot_be_parsed(
     assert network_mode is None
 
 
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 def test_replaces_localhost_api_with_dockerhost_when_not_using_host_network(
-    mock_docker_client, hosted_orion_api
+    mock_docker_client, hosted_api_server
 ):
     DockerContainer(
         command=["echo", "hello"],
@@ -467,15 +467,15 @@ def test_replaces_localhost_api_with_dockerhost_when_not_using_host_network(
     mock_docker_client.containers.create.assert_called_once()
     call_env = mock_docker_client.containers.create.call_args[1].get("environment")
     assert "PREFECT_API_URL" in call_env
-    assert call_env["PREFECT_API_URL"] == hosted_orion_api.replace(
+    assert call_env["PREFECT_API_URL"] == hosted_api_server.replace(
         "localhost", "host.docker.internal"
     )
 
 
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 def test_does_not_replace_localhost_api_when_using_host_network(
     mock_docker_client,
-    hosted_orion_api,
+    hosted_api_server,
     monkeypatch,
 ):
     # We will warn if setting 'host' network mode on non-linux platforms
@@ -488,10 +488,10 @@ def test_does_not_replace_localhost_api_when_using_host_network(
     mock_docker_client.containers.create.assert_called_once()
     call_env = mock_docker_client.containers.create.call_args[1].get("environment")
     assert "PREFECT_API_URL" in call_env
-    assert call_env["PREFECT_API_URL"] == hosted_orion_api
+    assert call_env["PREFECT_API_URL"] == hosted_api_server
 
 
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 def test_warns_at_runtime_when_using_host_network_mode_on_non_linux_platform(
     mock_docker_client,
     monkeypatch,
@@ -660,7 +660,7 @@ def test_does_not_add_docker_host_gateway_on_other_platforms(
         "http://host.docker.internal:10/foo/api",
     ],
 )
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 def test_warns_if_docker_version_does_not_support_host_gateway_on_linux(
     mock_docker_client,
     explicit_api_url,
@@ -725,7 +725,7 @@ def test_task_status_receives_result_identifier(mock_docker_client):
     fake_status.started.assert_called_once_with(result.identifier)
 
 
-@pytest.mark.usefixtures("use_hosted_orion")
+@pytest.mark.usefixtures("use_hosted_api_server")
 @pytest.mark.parametrize("platform", ["win32", "darwin"])
 def test_does_not_warn_about_gateway_if_not_using_linux(
     mock_docker_client,

tests/infrastructure/test_kubernetes_job.py

@@ -692,7 +692,7 @@ def test_no_raise_on_submission_with_hosted_api(
     mock_cluster_config,
     mock_k8s_batch_client,
     mock_k8s_client,
-    use_hosted_orion,
+    use_hosted_api_server,
 ):
     KubernetesJob(command=["echo", "hello"]).run(MagicMock())
 

tests/test_deployments.py

@@ -649,7 +649,7 @@ class TestRunDeployment:
     def test_running_a_deployment_blocks_until_termination(
         self,
         test_deployment,
-        use_hosted_orion,
+        use_hosted_api_server,
         terminal_state,
     ):
         d, deployment_id = test_deployment
@@ -696,7 +696,7 @@ def test_running_a_deployment_blocks_until_termination(
     async def test_running_a_deployment_blocks_until_termination_async(
         self,
         test_deployment,
-        use_hosted_orion,
+        use_hosted_api_server,
         terminal_state,
     ):
         d, deployment_id = test_deployment
@@ -791,7 +791,7 @@ async def test_run_deployment_with_deployment_id_uuid(
     def test_returns_flow_run_on_timeout(
         self,
         test_deployment,
-        use_hosted_orion,
+        use_hosted_api_server,
     ):
         d, deployment_id = test_deployment
 
@@ -822,7 +822,7 @@ def test_returns_flow_run_on_timeout(
     def test_returns_flow_run_immediately_when_timeout_is_zero(
         self,
         test_deployment,
-        use_hosted_orion,
+        use_hosted_api_server,
     ):
         d, deployment_id = test_deployment
 
@@ -855,7 +855,7 @@ def test_returns_flow_run_immediately_when_timeout_is_zero(
     def test_polls_indefinitely(
         self,
         test_deployment,
-        use_hosted_orion,
+        use_hosted_api_server,
     ):
         d, deployment_id = test_deployment
 
@@ -889,7 +889,9 @@ def test_polls_indefinitely(
             run_deployment(f"{d.flow_name}/{d.name}", timeout=None, poll_interval=0)
             assert len(flow_polls.calls) == 100
 
-    def test_schedules_immediately_by_default(self, test_deployment, use_hosted_orion):
+    def test_schedules_immediately_by_default(
+        self, test_deployment, use_hosted_api_server
+    ):
         d, deployment_id = test_deployment
 
         scheduled_time = pendulum.now()
@@ -901,7 +903,9 @@ def test_schedules_immediately_by_default(self, test_deployment, use_hosted_orio
 
         assert (flow_run.expected_start_time - scheduled_time).total_seconds() < 1
 
-    def test_accepts_custom_scheduled_time(self, test_deployment, use_hosted_orion):
+    def test_accepts_custom_scheduled_time(
+        self, test_deployment, use_hosted_api_server
+    ):
         d, deployment_id = test_deployment
 
         scheduled_time = pendulum.now() + pendulum.Duration(minutes=5)
@@ -914,7 +918,7 @@ def test_accepts_custom_scheduled_time(self, test_deployment, use_hosted_orion):
 
         assert (flow_run.expected_start_time - scheduled_time).total_seconds() < 1
 
-    def test_custom_flow_run_names(self, test_deployment, use_hosted_orion):
+    def test_custom_flow_run_names(self, test_deployment, use_hosted_api_server):
         d, deployment_id = test_deployment
 
         flow_run = run_deployment(
@@ -958,7 +962,7 @@ def test_accepts_idempotency_key(self, test_deployment):
         assert flow_run_a.id == flow_run_b.id
 
     async def test_links_to_parent_flow_run_when_used_in_flow(
-        self, test_deployment, use_hosted_orion, orion_client: PrefectClient
+        self, test_deployment, use_hosted_api_server, orion_client: PrefectClient
     ):
         d, deployment_id = test_deployment
 
@@ -978,7 +982,7 @@ def foo():
         assert slugify(f"{d.flow_name}/{d.name}") in task_run.task_key
 
     async def test_tracks_dependencies_when_used_in_flow(
-        self, test_deployment, use_hosted_orion, orion_client
+        self, test_deployment, use_hosted_api_server, orion_client
     ):
         d, deployment_id = test_deployment