forked from temporalio/sdk-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_service.py
More file actions
149 lines (131 loc) · 4.91 KB
/
Copy pathtest_service.py
File metadata and controls
149 lines (131 loc) · 4.91 KB
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import inspect
import os
import re
from typing import Any, Callable, Dict, Mapping, Tuple, Type
import google.protobuf.empty_pb2
import google.protobuf.message
import grpc
import pytest
import temporalio
import temporalio.api.errordetails.v1
import temporalio.api.operatorservice.v1
import temporalio.api.testservice.v1
import temporalio.api.workflowservice.v1
import temporalio.service
from temporalio.client import Client
from temporalio.testing import WorkflowEnvironment
def test_all_grpc_calls_present(client: Client):
def assert_all_calls_present(
service: Any,
package: Any,
new_stub: Callable[[grpc.Channel], Any],
custom_req_resp: Mapping[
str,
Tuple[
Type[google.protobuf.message.Message],
Type[google.protobuf.message.Message],
],
] = {},
) -> None:
# Collect service calls
service_calls: Dict[str, Tuple[Type, Type]] = {}
for _, call in inspect.getmembers(service):
if isinstance(call, temporalio.service.ServiceCall):
service_calls[call.name] = (call.req_type, call.resp_type)
# Collect gRPC service calls with a fake channel
channel = CallCollectingChannel(package, custom_req_resp)
new_stub(channel)
# Confirm they are the same
assert channel.calls == service_calls
assert_all_calls_present(
client.workflow_service,
temporalio.api.workflowservice.v1,
temporalio.api.workflowservice.v1.WorkflowServiceStub,
)
assert_all_calls_present(
client.operator_service,
temporalio.api.operatorservice.v1,
temporalio.api.operatorservice.v1.OperatorServiceStub,
)
assert_all_calls_present(
client.test_service,
temporalio.api.testservice.v1,
temporalio.api.testservice.v1.TestServiceStub,
{
# Abnormal req/resp
"GetCurrentTime": (
google.protobuf.empty_pb2.Empty,
temporalio.api.testservice.v1.GetCurrentTimeResponse,
),
"SleepUntil": (
temporalio.api.testservice.v1.SleepUntilRequest,
temporalio.api.testservice.v1.SleepResponse,
),
"UnlockTimeSkippingWithSleep": (
temporalio.api.testservice.v1.SleepRequest,
temporalio.api.testservice.v1.SleepResponse,
),
},
)
class CallCollectingChannel(grpc.Channel):
def __init__(
self,
package: Any,
custom_req_resp: Mapping[
str,
Tuple[
Type[google.protobuf.message.Message],
Type[google.protobuf.message.Message],
],
],
) -> None:
super().__init__()
self.package = package
self.custom_req_resp = custom_req_resp
self.calls: Dict[str, Tuple[Type, Type]] = {}
def unary_unary(self, method, request_serializer, response_deserializer):
# Last part after slash
name = method.rsplit("/", 1)[-1]
req_resp = self.custom_req_resp.get(name, None) or (
getattr(self.package, name + "Request"),
getattr(self.package, name + "Response"),
)
# Camel to snake case
name = re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()
self.calls[name] = req_resp
CallCollectingChannel.__abstractmethods__ = set()
def test_version():
# Extract version from pyproject.toml
with open(
os.path.join(os.path.dirname(__file__), "..", "pyproject.toml"), "r"
) as f:
pyproject = f.read()
version = pyproject[pyproject.find('version = "') + 11 :]
version = version[: version.find('"')]
assert temporalio.service.__version__ == version
assert temporalio.__version__ == version
async def test_check_health(client: Client):
assert await client.service_client.check_health()
# Unknown service
with pytest.raises(temporalio.service.RPCError) as err:
assert await client.service_client.check_health(service="whatever")
assert err.value.status == temporalio.service.RPCStatusCode.NOT_FOUND
async def test_grpc_status(client: Client, env: WorkflowEnvironment):
if env.supports_time_skipping:
pytest.skip(
"Java test server: https://github.com/temporalio/sdk-java/issues/1557"
)
# Try to make a simple client call on a non-existent namespace
with pytest.raises(temporalio.service.RPCError) as err:
await client.workflow_service.describe_namespace(
temporalio.api.workflowservice.v1.DescribeNamespaceRequest(
namespace="does not exist",
)
)
# Confirm right failure type
assert not err.value.grpc_status.details[0].Is(
temporalio.api.errordetails.v1.QueryFailedFailure.DESCRIPTOR
)
assert err.value.grpc_status.details[0].Is(
temporalio.api.errordetails.v1.NamespaceNotFoundFailure.DESCRIPTOR
)