-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathtest_server.py
More file actions
127 lines (106 loc) · 3.91 KB
/
test_server.py
File metadata and controls
127 lines (106 loc) · 3.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
"""Test the ThingServer.
While the server is covered by many of the other tests, it would
be helpful to have some more bottom-up unit testing in this file.
"""
import pytest
import labthings_fastapi as lt
from fastapi.testclient import TestClient
from starlette.routing import Route
from labthings_fastapi.example_things import MyThing
def test_server_from_config_non_thing_error():
"""Test a typeerror is raised if something that's not a Thing is added."""
with pytest.raises(TypeError, match="not a Thing"):
lt.ThingServer.from_config(
lt.ThingServerConfig(
things={"thingone": lt.ThingConfig(cls="builtins:object")}
)
)
def test_server_thing_descriptions():
"""Check the server ThingDescriptions.
Check the expected Action and Properties and their URLS are in the server
thing_descriptions.
"""
conf = {
"things": {
"thing1": "labthings_fastapi.example_things:MyThing",
"thing2": {
"class": "labthings_fastapi.example_things:MyThing",
"kwargs": {},
},
}
}
thing_names = ["thing1", "thing2"]
props = ["counter", "foo"]
actions = [
"action_with_only_kwargs",
"action_without_arguments",
"anaction",
"increment_counter",
"make_a_dict",
"slowly_increase_counter",
]
server = lt.ThingServer.from_config(conf)
with TestClient(server.app) as client:
response = client.get("/thing_descriptions/")
response.raise_for_status()
thing_descriptions = response.json()
for thing_name in thing_names:
thing_description = thing_descriptions[thing_name]
for action_name in actions:
action = thing_description["actions"][action_name]
expected_href = thing_name + "/" + action_name
assert action["forms"][0]["href"] == expected_href
for prop_name in props:
prop = thing_description["properties"][prop_name]
expected_href = thing_name + "/" + prop_name
assert prop["forms"][0]["href"] == expected_href
def test_api_prefix():
"""Check we can add a prefix to the URLs on a server."""
class Example(lt.Thing):
"""An example Thing"""
server = lt.ThingServer(things={"example": Example}, api_prefix="/api/v3")
paths = [route.path for route in server.app.routes if isinstance(route, Route)]
for expected_path in [
"/api/v3/action_invocations",
"/api/v3/action_invocations/{id}",
"/api/v3/action_invocations/{id}/output",
"/api/v3/action_invocations/{id}",
"/api/v3/blob/{blob_id}",
"/api/v3/thing_descriptions/",
"/api/v3/things/",
"/api/v3/example/",
]:
assert expected_path in paths
unprefixed_paths = {p for p in paths if not p.startswith("/api/v3/")}
assert unprefixed_paths == {
"/openapi.json",
"/docs",
"/docs/oauth2-redirect",
"/redoc",
}
def test_things_endpoints():
"""Test that the two endpoints for listing Things work."""
server = lt.ThingServer(
{
"thing_a": MyThing,
"thing_b": MyThing,
}
)
with TestClient(server.app) as client:
# Check the thing_descriptions endpoint
response = client.get("/thing_descriptions/")
response.raise_for_status()
tds = response.json()
assert "thing_a" in tds
assert "thing_b" in tds
# Check the things endpoint. This should map names to URLs
response = client.get("/things/")
response.raise_for_status()
things = response.json()
assert "thing_a" in things
assert "thing_b" in things
# Fetch a thing description from the URL in `things`
response = client.get(things["thing_a"])
response.raise_for_status()
td = response.json()
assert td["title"] == "MyThing"