merge conflict

tests/conftest.py

@@ -66,6 +66,7 @@
     PREFECT_SERVER_LOGGING_LEVEL,
     PREFECT_UNIT_TEST_LOOP_DEBUG,
 )
+from prefect.types._datetime import DateTime, now
 from prefect.utilities.dispatch import get_registry_for_type
 
 # isort: split
@@ -143,7 +144,9 @@ def pytest_addoption(parser):
 ]
 
 
-def pytest_collection_modifyitems(session, config, items):
+def pytest_collection_modifyitems(
+    session: pytest.Session, config: pytest.Config, items: list[pytest.Item]
+):
     """
     Update tests to skip in accordance with service requests
     """
@@ -267,7 +270,7 @@ def tests_dir() -> pathlib.Path:
 
 
 @pytest.hookimpl(hookwrapper=True)
-def pytest_runtest_call(item):
+def pytest_runtest_call(item: pytest.Item):
     """
     This hook will be called within the test run. Allowing us to raise errors or add
     assertions to every test. On error, the test will be marked as failed. If we used
@@ -302,7 +305,7 @@ def assert_lifespan_is_not_left_open():
 TEST_PROFILE_CTX = None
 
 
-def pytest_sessionstart(session):
+def pytest_sessionstart(session: pytest.Session):
     """
     Creates a profile for the scope of the test session that modifies setting defaults.
 
@@ -377,7 +380,7 @@ def pytest_sessionstart(session):
 
 # def pytest_sessionfinish(session, exitstatus):
 @pytest.fixture(scope="session", autouse=True)
-def cleanup(drain_log_workers, drain_events_workers):
+def cleanup(drain_log_workers: None, drain_events_workers: None):
     # this fixture depends on other fixtures with important cleanup steps like
     # draining workers to ensure that the home directory is not deleted before
     # these steps are completed
@@ -518,7 +521,9 @@ def reset_registered_blocks():
 
 
 @pytest.fixture
-def caplog(caplog):
+def caplog(
+    caplog: pytest.LogCaptureFixture,
+) -> Generator[pytest.LogCaptureFixture, None, None]:
     """
     Overrides caplog to apply to all of our loggers that do not propagate and
     consequently would not be captured by caplog.
@@ -541,8 +546,8 @@ def disable_csrf_protection():
 
 
 @pytest.fixture
-def start_of_test() -> pendulum.DateTime:
-    return pendulum.now("UTC")
+def start_of_test() -> DateTime:
+    return now("UTC")
 
 
 @pytest.fixture(autouse=True)

tests/test_flows.py

@@ -13,12 +13,11 @@
 from itertools import combinations
 from pathlib import Path
 from textwrap import dedent
-from typing import List, Optional
+from typing import Any, List, Optional
 from unittest import mock
 from unittest.mock import ANY, MagicMock, call, create_autospec
 
 import anyio
-import pendulum
 import pydantic
 import pytest
 import regex as re
@@ -87,6 +86,7 @@
     get_most_recent_flow_run,
 )
 from prefect.transactions import get_transaction, transaction
+from prefect.types._datetime import DateTime, Timezone
 from prefect.types.entrypoint import EntrypointType
 from prefect.utilities.annotations import allow_failure, quote
 from prefect.utilities.callables import parameter_schema
@@ -103,7 +103,7 @@ def mock_sigterm_handler():
         pytest.skip("Can't test signal handlers from a thread")
     mock = MagicMock()
 
-    def handler(*args, **kwargs):
+    def handler(*args: Any, **kwargs: Any):
         mock(*args, **kwargs)
 
     prev_handler = signal.signal(signal.SIGTERM, handler)
@@ -5555,14 +5555,14 @@ def flow_function(name: str) -> str:
     def test_annotations_and_defaults_rely_on_imports(self, tmp_path: Path):
         source_code = dedent(
             """
-        import pendulum
         import datetime
         from prefect import flow
+        from prefect.types import DateTime
 
         @flow
         def f(
             x: datetime.datetime,
-            y: pendulum.DateTime = pendulum.datetime(2025, 1, 1),
+            y: DateTime = DateTime(2025, 1, 1),
             z: datetime.timedelta = datetime.timedelta(seconds=5),
         ):
             return x, y, z
@@ -5573,7 +5573,7 @@ def f(
         assert result is not None
         assert result(datetime.datetime(2025, 1, 1)) == (
             datetime.datetime(2025, 1, 1),
-            pendulum.datetime(2025, 1, 1),
+            DateTime(2025, 1, 1, tzinfo=Timezone("UTC")),
             datetime.timedelta(seconds=5),
         )
 

tests/test_logging.py

@@ -10,7 +10,6 @@
 from unittest import mock
 from unittest.mock import ANY, MagicMock
 
-import pendulum
 import pytest
 from rich.color import Color, ColorType
 from rich.console import Console
@@ -68,6 +67,7 @@
 )
 from prefect.testing.cli import temporary_console_width
 from prefect.testing.utilities import AsyncMock
+from prefect.types._datetime import from_timestamp, now
 from prefect.utilities.names import obfuscate
 from prefect.workers.base import BaseJobConfiguration, BaseWorker
 
@@ -464,8 +464,7 @@ def test_sets_timestamp_from_record_created_time(
         log_dict = mock_log_worker.instance().send.call_args[0][0]
 
         assert (
-            log_dict["timestamp"]
-            == pendulum.from_timestamp(record.created).to_iso8601_string()
+            log_dict["timestamp"] == from_timestamp(record.created).to_iso8601_string()
         )
 
     def test_sets_timestamp_from_time_if_missing_from_recrod(
@@ -487,7 +486,7 @@ def drop_created_and_emit(emit, record):
 
         log_dict = mock_log_worker.instance().send.call_args[0][0]
 
-        assert log_dict["timestamp"] == pendulum.from_timestamp(now).to_iso8601_string()
+        assert log_dict["timestamp"] == from_timestamp(now).to_iso8601_string()
 
     def test_does_not_send_logs_that_opt_out(self, logger, mock_log_worker, task_run):
         with TaskRunContext.model_construct(task_run=task_run):
@@ -821,7 +820,7 @@ def log_dict(self):
             task_run_id=uuid.uuid4(),
             name="test.logger",
             level=10,
-            timestamp=pendulum.now("utc"),
+            timestamp=now("utc"),
             message="hello",
         ).model_dump(mode="json")