durabletask-python/tests/durabletask/test_orchestration_versioning_e2e.py at ff73432f0a7008661b00e92fbd5e4642aec632f4 · andystaples/durabletask-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

import json

import pytest

from durabletask import client, task, worker
from durabletask.testing import create_test_backend

from _port_utils import find_free_port

PORT = find_free_port()
HOST = f"localhost:{PORT}"


@pytest.fixture(autouse=True)
def backend():
    """Create an in-memory backend for testing."""
    b = create_test_backend(port=PORT)
    yield b
    b.stop()
    b.reset()


def test_version_flows_through_to_orchestration_context():
    """Verifies the version set by the client is visible on ctx.version."""

    def return_version(ctx: task.OrchestrationContext, _: None):
        return ctx.version

    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
        w.add_orchestrator(return_version)
        w.use_versioning(worker.VersioningOptions(
            version="2.5.0",
            default_version="2.5.0",
            match_strategy=worker.VersionMatchStrategy.CURRENT_OR_OLDER,
            failure_strategy=worker.VersionFailureStrategy.FAIL
        ))
        w.start()

        with client.TaskHubGrpcClient(host_address=HOST) as task_hub_client:
            id = task_hub_client.schedule_new_orchestration(
                return_version, version="2.5.0")
            state = task_hub_client.wait_for_orchestration_completion(
                id, timeout=30)

    assert state is not None
    assert state.runtime_status == client.OrchestrationStatus.COMPLETED
    assert state.failure_details is None
    assert state.serialized_output == json.dumps("2.5.0")


def test_strict_version_mismatch_fails():
    """Worker with STRICT matching fails an orchestration whose version differs."""

    def simple(ctx: task.OrchestrationContext, _: None):
        return "done"

    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
        w.add_orchestrator(simple)
        w.use_versioning(worker.VersioningOptions(
            version="1.0.0",
            default_version="1.0.0",
            match_strategy=worker.VersionMatchStrategy.STRICT,
            failure_strategy=worker.VersionFailureStrategy.FAIL
        ))
        w.start()

        with client.TaskHubGrpcClient(host_address=HOST) as task_hub_client:
            id = task_hub_client.schedule_new_orchestration(
                simple, version="2.0.0")
            state = task_hub_client.wait_for_orchestration_completion(
                id, timeout=30)

    assert state is not None
    assert state.runtime_status == client.OrchestrationStatus.FAILED
    assert state.failure_details is not None
    assert "does not match the worker version" in state.failure_details.message


def test_newer_orchestration_version_fails_current_or_older():
    """Worker rejects orchestrations with a version newer than its own."""

    def simple(ctx: task.OrchestrationContext, _: None):
        return "done"

    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
        w.add_orchestrator(simple)
        w.use_versioning(worker.VersioningOptions(
            version="1.0.0",
            default_version="1.0.0",
            match_strategy=worker.VersionMatchStrategy.CURRENT_OR_OLDER,
            failure_strategy=worker.VersionFailureStrategy.FAIL
        ))
        w.start()

        with client.TaskHubGrpcClient(host_address=HOST) as task_hub_client:
            id = task_hub_client.schedule_new_orchestration(
                simple, version="1.1.0")
            state = task_hub_client.wait_for_orchestration_completion(
                id, timeout=30)

    assert state is not None
    assert state.runtime_status == client.OrchestrationStatus.FAILED
    assert state.failure_details is not None
    assert "is greater than the worker version" in state.failure_details.message