|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from typing import Any |
| 4 | + |
| 5 | +import pytest |
| 6 | + |
| 7 | +from durable_workflow.errors import NonDeterministicReplayError |
| 8 | +from durable_workflow.workflow import CompleteWorkflow, replay |
| 9 | +from tests.business_clock_workflows import PrefixedCoreTimerBusinessClockWorkflow |
| 10 | +from tests.test_replay_clock import redrive_fixture |
| 11 | + |
| 12 | +ORIGINAL = "2026-11-03T00:00:06.000000+00:00" |
| 13 | +INSERTED = "2026-11-03T00:05:00.000000+00:00" |
| 14 | +REUSE_FIELDS = ("reused_from_run_id", "reused_recorded_at") |
| 15 | + |
| 16 | + |
| 17 | +def fixture(prefixed: bool) -> dict[str, Any]: |
| 18 | + document = redrive_fixture("business-clock-core-redrive-acceptance-timer") |
| 19 | + if not prefixed: |
| 20 | + document["history"].pop(0) |
| 21 | + for index, event in enumerate(document["history"], start=1): |
| 22 | + event["sequence"] = index |
| 23 | + return document |
| 24 | + |
| 25 | + |
| 26 | +def result(document: dict[str, Any]) -> dict[str, Any]: |
| 27 | + task = document["workflow"] |
| 28 | + outcome = replay(PrefixedCoreTimerBusinessClockWorkflow, document["history"], task["input"], |
| 29 | + workflow_id=task["workflow_id"], run_id=task["run_id"], payload_codec="avro") |
| 30 | + try: |
| 31 | + assert len(outcome.commands) == 1 |
| 32 | + command = outcome.commands[0] |
| 33 | + assert isinstance(command, CompleteWorkflow), command |
| 34 | + return dict(command.result) |
| 35 | + finally: |
| 36 | + outcome.close() |
| 37 | + |
| 38 | + |
| 39 | +@pytest.mark.parametrize("prefixed", [False, True]) |
| 40 | +@pytest.mark.parametrize("missing", [REUSE_FIELDS, REUSE_FIELDS[:1], REUSE_FIELDS[1:]]) |
| 41 | +def test_prefix_before_resume_requires_both_reuse_fields_without_insertion_time_fallback( |
| 42 | + prefixed: bool, missing: tuple[str, ...], |
| 43 | +) -> None: |
| 44 | + document = fixture(prefixed) |
| 45 | + expected = result(document) |
| 46 | + assert expected["clock"][-1] == ORIGINAL |
| 47 | + for key in missing: |
| 48 | + document["history"][-1]["payload"].pop(key) |
| 49 | + expected["clock"][-1] = None |
| 50 | + assert result(document) == expected |
| 51 | + assert result(document) == expected # Cold reconstruction repeats the refusal. |
| 52 | + |
| 53 | + |
| 54 | +@pytest.mark.parametrize("prefixed", [False, True]) |
| 55 | +@pytest.mark.parametrize("missing", ["workflow_command_id", "continued_from_run_id", "workflow_link_id", |
| 56 | + "workflow_run_id", "resume_step_sequence", "recovery_kind", "replayed_started_at"]) |
| 57 | +def test_unbound_redrive_header_cannot_turn_an_unmarked_timer_into_an_ordinary_result( |
| 58 | + prefixed: bool, missing: str, |
| 59 | +) -> None: |
| 60 | + document = fixture(prefixed) |
| 61 | + started = next(event for event in document["history"] if event["event_type"] == "WorkflowStarted") |
| 62 | + started["payload"].pop(missing) |
| 63 | + for key in REUSE_FIELDS: |
| 64 | + document["history"][-1]["payload"].pop(key) |
| 65 | + assert result(document)["clock"] == [None] * 3 |
| 66 | + |
| 67 | + |
| 68 | +@pytest.mark.parametrize("prefixed", [False, True]) |
| 69 | +def test_bound_post_resume_timer_without_reuse_fields_uses_its_own_consumed_time(prefixed: bool) -> None: |
| 70 | + document = fixture(prefixed) |
| 71 | + started = next(event for event in document["history"] if event["event_type"] == "WorkflowStarted") |
| 72 | + started["payload"]["resume_step_sequence"] = 2 |
| 73 | + for key in REUSE_FIELDS: |
| 74 | + document["history"][-1]["payload"].pop(key) |
| 75 | + assert result(document)["clock"][-1] == INSERTED |
| 76 | + |
| 77 | + |
| 78 | +@pytest.mark.parametrize("prefix_fault", ["outcome", "identity"]) |
| 79 | +def test_unbound_acceptance_prefix_does_not_lend_timer_insertion_time(prefix_fault: str) -> None: |
| 80 | + document = fixture(True) |
| 81 | + for key in REUSE_FIELDS: |
| 82 | + document["history"][-1]["payload"].pop(key) |
| 83 | + if prefix_fault == "outcome": |
| 84 | + document["history"][0]["payload"]["outcome"] = "started" |
| 85 | + else: |
| 86 | + document["history"][0]["payload"]["workflow_run_id"] = "foreign" |
| 87 | + assert result(document)["clock"] == [None] * 3 |
| 88 | + |
| 89 | + |
| 90 | +def test_missing_start_retains_the_original_legacy_clock_error() -> None: |
| 91 | + document = fixture(True) |
| 92 | + document["history"].pop(1) |
| 93 | + with pytest.raises(NonDeterministicReplayError, match="missing WorkflowStarted persisted start time"): |
| 94 | + result(document) |
| 95 | + |
| 96 | + |
| 97 | +@pytest.mark.parametrize("prefixed", [False, True]) |
| 98 | +@pytest.mark.parametrize("alias", ["WorkflowTaskCompleted", "StartAccepted"]) |
| 99 | +def test_conflicting_start_kind_alias_cannot_hide_an_unbound_redrive_header(prefixed: bool, alias: str) -> None: |
| 100 | + document = fixture(prefixed) |
| 101 | + started = next(event for event in document["history"] if event["event_type"] == "WorkflowStarted") |
| 102 | + started["type"] = alias |
| 103 | + for key in REUSE_FIELDS: |
| 104 | + document["history"][-1]["payload"].pop(key) |
| 105 | + assert result(document) == {"clock": [None] * 3, "legacy": "2026-11-03T00:00:00.000000+00:00"} |
| 106 | + |
| 107 | + |
| 108 | +def test_pre_resume_activity_with_a_schedule_cannot_use_insertion_time_when_reuse_fields_are_missing() -> None: |
| 109 | + document = fixture(False) |
| 110 | + terminal = document["history"][1] |
| 111 | + for key in (*REUSE_FIELDS, "reused_activity_execution_id"): |
| 112 | + terminal["payload"].pop(key) |
| 113 | + document["history"].insert(1, { |
| 114 | + "event_type": "ActivityScheduled", "recorded_at": terminal["recorded_at"], |
| 115 | + "payload": {key: terminal["payload"][key] for key in ("sequence", "activity_type", "activity_execution_id")}, |
| 116 | + }) |
| 117 | + for index, event in enumerate(document["history"], start=1): |
| 118 | + event["sequence"] = index |
| 119 | + assert result(document)["clock"] == ["2026-11-03T00:00:00.000000+00:00", None, ORIGINAL] |
| 120 | + |
| 121 | + |
| 122 | +@pytest.mark.parametrize("continued", [False, True]) |
| 123 | +def test_ordinary_and_continued_histories_without_redrive_metadata_keep_their_own_times(continued: bool) -> None: |
| 124 | + from tests.test_business_replay_clock import history |
| 125 | + from tests.test_business_replay_clock import result as scalar_result |
| 126 | + |
| 127 | + events = history() |
| 128 | + if continued: |
| 129 | + events[0]["payload"]["continued_from_run_id"] = "previous-run" |
| 130 | + assert scalar_result(events)["clock"] == [ |
| 131 | + None, "2026-11-01T00:00:03.123456+00:00", "2026-11-01T00:00:05.654321+00:00", |
| 132 | + ] |
| 133 | + |
| 134 | + |
| 135 | +@pytest.mark.parametrize("marker", ["reused_from_run_id", "reused_recorded_at", "reused_activity_execution_id"]) |
| 136 | +def test_ordinary_success_with_a_stray_reuse_marker_has_no_proved_source_time(marker: str) -> None: |
| 137 | + from tests.test_business_replay_clock import history |
| 138 | + from tests.test_business_replay_clock import result as scalar_result |
| 139 | + |
| 140 | + events = history() |
| 141 | + events[2]["payload"][marker] = "unbound" |
| 142 | + assert scalar_result(events)["clock"][1] is None |
0 commit comments