azure-functions-durable-python/tests/conftest.py at e33c6cf50536d7fe24363da1e1720beddc96ab61 · Azure/azure-functions-durable-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
import pytest
import json

import azure.durable_functions as df
import azure.functions as func
from tests.test_utils.constants import RPC_BASE_URL
from azure.durable_functions.models.DurableOrchestrationBindings import \
    DurableOrchestrationBindings

TASK_HUB_NAME = "DurableFunctionsHub"
BASE_URL = "http://localhost:7071/runtime/webhooks/durabletask"
AUTH_CODE = "NOT-REAL-AUTH"


def get_binding_string():
    binding = {
        "taskHubName": TASK_HUB_NAME,
        "creationUrls": {
            "createNewInstancePostUri": f"{BASE_URL}/orchestrators/"
                                        "{functionName}[/{instanceId}]?code="
                                        f"{AUTH_CODE}",
            "createAndWaitOnNewInstancePostUri": f"{BASE_URL}/orchestrators/"
                                                 "{functionName}[/{instanceId}]?timeout="
                                                 "{timeoutInSeconds}&pollingInterval="
                                                 "{intervalInSeconds}&code="
                                                 f"{AUTH_CODE}"
        },
        "managementUrls": {
            "id": "INSTANCEID",
            "statusQueryGetUri": f"{BASE_URL}/instances/INSTANCEID?taskHub=DurableFunctionsHub&"
                                 f"connection=Storage&code={AUTH_CODE}",
            "sendEventPostUri": f"{BASE_URL}/instances/INSTANCEID/raiseEvent/"
                                "{eventName}?taskHub="
                                f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
            "terminatePostUri": f"{BASE_URL}/instances/INSTANCEID/terminate?reason="
                                "{text}&taskHub="
                                f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
            "rewindPostUri": f"{BASE_URL}/instances/INSTANCEID/rewind?reason="
                             "{text}&taskHub="
                             f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
            "purgeHistoryDeleteUri": f"{BASE_URL}/instances/INSTANCEID?taskHub="
                                     f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
            "suspendPostUri": f"{BASE_URL}/instances/INSTANCEID/suspend?reason="
                                "{text}&taskHub="
                                f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
            "resumePostUri": f"{BASE_URL}/instances/INSTANCEID/resume?reason="
                                "{text}&taskHub="
                                f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
            "restartPostUri": f"{BASE_URL}/instances/INSTANCEID/restart?taskHub="
                                f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
        },
        "rpcBaseUrl": RPC_BASE_URL
    }
    binding_string = json.dumps(binding)

    binding_string = replace_stand_in_bits(binding_string)
    return binding_string


@pytest.fixture()
def binding_string():
    return get_binding_string()


@pytest.fixture()
def binding_info():
    binding = DurableOrchestrationBindings.from_json(get_binding_string())
    return binding


def replace_stand_in_bits(binding_string):
    binding_string = binding_string.replace("TASK_HUB_NAME", TASK_HUB_NAME)
    binding_string = binding_string.replace("BASE_URL", BASE_URL)
    binding_string = binding_string.replace("AUTH_CODE", AUTH_CODE)
    return binding_string

@pytest.fixture()
def app():
    app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
    return app