-
-
Notifications
You must be signed in to change notification settings - Fork 2k
/
test_autobahn.py
133 lines (116 loc) · 4.2 KB
/
test_autobahn.py
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
import json
import subprocess
import sys
from pathlib import Path
from typing import Any, Dict, Generator, List
import pytest
import python_on_whales
from pytest import TempPathFactory
@pytest.fixture(scope="session")
def report_dir(tmp_path_factory: TempPathFactory) -> Path:
return tmp_path_factory.mktemp("reports")
@pytest.fixture(scope="session", autouse=True)
def build_autobahn_testsuite() -> Generator[None, None, None]:
try:
python_on_whales.docker.build(
file="tests/autobahn/Dockerfile.autobahn",
tags=["autobahn-testsuite"],
context_path=".",
)
except python_on_whales.DockerException:
pytest.skip(msg="The docker daemon is not running.")
try:
yield
finally:
python_on_whales.docker.image.remove(x="autobahn-testsuite")
def get_failed_tests(report_path: str, name: str) -> List[Dict[str, Any]]:
path = Path(report_path)
result_summary = json.loads((path / "index.json").read_text())[name]
failed_messages = []
PASS = {"OK", "INFORMATIONAL"}
entry_fields = {"case", "description", "expectation", "expected", "received"}
for results in result_summary.values():
if results["behavior"] in PASS and results["behaviorClose"] in PASS:
continue
report = json.loads((path / results["reportfile"]).read_text())
failed_messages.append({field: report[field] for field in entry_fields})
return failed_messages
@pytest.mark.skipif(sys.platform == "darwin", reason="Don't run on macOS")
@pytest.mark.xfail
def test_client(report_dir: Path, request: Any) -> None:
try:
print("Starting autobahn-testsuite server")
autobahn_container = python_on_whales.docker.run(
detach=True,
image="autobahn-testsuite",
name="autobahn",
publish=[(9001, 9001)],
remove=True,
volumes=[
(f"{request.fspath.dirname}/client", "/config"),
(f"{report_dir}", "/reports"),
],
)
print("Running aiohttp test client")
client = subprocess.Popen(
["wait-for-it", "-s", "localhost:9001", "--"]
+ [sys.executable]
+ ["tests/autobahn/client/client.py"]
)
client.wait()
finally:
print("Stopping client and server")
client.terminate()
client.wait()
# https://github.com/gabrieldemarmiesse/python-on-whales/pull/580
autobahn_container.stop() # type: ignore[union-attr]
failed_messages = get_failed_tests(f"{report_dir}/clients", "aiohttp")
assert not failed_messages, "\n".join(
"\n\t".join(
f"{field}: {msg[field]}"
for field in ("case", "description", "expectation", "expected", "received")
)
for msg in failed_messages
)
@pytest.mark.skipif(sys.platform == "darwin", reason="Don't run on macOS")
@pytest.mark.xfail
def test_server(report_dir: Path, request: Any) -> None:
try:
print("Starting aiohttp test server")
server = subprocess.Popen(
[sys.executable] + ["tests/autobahn/server/server.py"]
)
print("Starting autobahn-testsuite client")
python_on_whales.docker.run(
image="autobahn-testsuite",
name="autobahn",
remove=True,
volumes=[
(f"{request.fspath.dirname}/server", "/config"),
(f"{report_dir}", "/reports"),
],
networks=["host"],
command=[
"wait-for-it",
"-s",
"localhost:9001",
"--",
"wstest",
"--mode",
"fuzzingclient",
"--spec",
"/config/fuzzingclient.json",
],
)
finally:
print("Stopping client and server")
server.terminate()
server.wait()
failed_messages = get_failed_tests(f"{report_dir}/servers", "AutobahnServer")
assert not failed_messages, "\n".join(
"\n\t".join(
f"{field}: {msg[field]}"
for field in ("case", "description", "expectation", "expected", "received")
)
for msg in failed_messages
)