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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
import asyncio
import nikola.plugins.command.auto as auto
from nikola.utils import get_logger
import pytest
import pathlib
import requests
import socket
import sys
from typing import Optional, Tuple, Any, Dict
from ..helper import FakeSite
SERVER_ADDRESS = "localhost"
TEST_MAX_DURATION = 10 # Watchdog: Give up the test if it did not succeed during this time span.
# Folder that has the fixture file we expect the server to serve:
OUTPUT_FOLDER = pathlib.Path(__file__).parent.parent / "data" / "dev_server_sample_output_folder"
LOGGER = get_logger("test_dev_server")
def find_unused_port() -> int:
"""Ask the OS for a currently unused port number.
(More precisely, a port that can be used for a TCP server servicing SERVER_ADDRESS.)
We use a method here rather than a fixture to minimize side effects of failing tests.
"""
s = socket.socket()
try:
ANY_PORT = 0
s.bind((SERVER_ADDRESS, ANY_PORT))
address, port = s.getsockname()
LOGGER.info("Trying to set up dev server on http://%s:%i/", address, port)
return port
finally:
s.close()
class MyFakeSite(FakeSite):
def __init__(self, config: Dict[str, Any], configuration_filename="conf.py"):
self.configured = True
self.debug = True
self.THEMES = []
self._plugin_places = []
self.registered_auto_watched_folders = set()
self.config = config
self.configuration_filename = configuration_filename
def test_serves_root_dir(
site_and_base_path: Tuple[MyFakeSite, str], expected_text: str
) -> None:
site, base_path = site_and_base_path
command_auto = auto.CommandAuto()
command_auto.set_site(site)
options = {
"browser": False,
"ipv6": False,
"address": SERVER_ADDRESS,
"port": find_unused_port(),
"db-file": "/dev/null",
"backend": "No backend",
"no-server": False
}
# We start an event loop, run the test in an executor,
# and wait for the event loop to terminate.
# These variables help to transport the test result to
# the main thread outside the event loop:
test_was_successful = False
test_problem_description = "Async test setup apparently broken"
test_inner_error: Optional[BaseException] = None
loop_for_this_test = None
async def grab_loop_and_run_test() -> None:
nonlocal test_problem_description, loop_for_this_test
loop_for_this_test = asyncio.get_running_loop()
watchdog_handle = loop_for_this_test.call_later(TEST_MAX_DURATION, lambda: loop_for_this_test.stop())
test_problem_description = f"Test did not complete within {TEST_MAX_DURATION} seconds."
def run_test() -> None:
nonlocal test_was_successful, test_problem_description, test_inner_error
try:
with requests.Session() as session:
server_root_uri = f"http://{options['address']}:{options['port']}"
# First subtest: Grab the document root index.html file:
server_base_uri = f"{server_root_uri}{base_path}"
LOGGER.info("Attempting to fetch HTML from %s", server_base_uri)
res = session.get(server_base_uri)
res.raise_for_status()
assert "text/html; charset=utf-8" == res.headers['content-type']
assert expected_text in res.text
# Second subtest: Does the dev server serve something for the livereload JS?
js_uri = f"{server_root_uri}/livereload.js?snipver=1"
LOGGER.info("Attempting to fetch JS from %s", js_uri)
res_js = session.get(js_uri)
res_js.raise_for_status()
content_type_js = res_js.headers['content-type']
assert "javascript" in content_type_js
test_was_successful = True
test_problem_description = "No problem. All is well."
except BaseException as be:
LOGGER.error("Could not receive HTTP as expected.", exc_info=True)
test_inner_error = be
test_was_successful = False
test_problem_description = "(see exception)"
finally:
if test_was_successful:
LOGGER.info("Test completed successfully.")
else:
LOGGER.error("Test failed: %s", test_problem_description)
loop_for_this_test.call_soon_threadsafe(lambda: watchdog_handle.cancel())
# We give the outer grab_loop_and_run_test a chance to complete
# before burning the bridge:
loop_for_this_test.call_soon_threadsafe(lambda: loop_for_this_test.call_later(0.05, lambda: loop_for_this_test.stop()))
await loop_for_this_test.run_in_executor(None, run_test)
# We defeat the nikola site building functionality, so this does not actually get called.
# But the code setting up site building wants a command list:
command_auto.nikola_cmd = ["echo"]
# Defeat the site building functionality, and instead insert the test:
command_auto.run_initial_rebuild = grab_loop_and_run_test
try:
# Start the development server
# which under the hood runs our test when trying to build the site:
command_auto.execute(options=options)
# Verify the test succeeded:
if test_inner_error is not None:
raise test_inner_error
assert test_was_successful, test_problem_description
finally:
# Nikola is written with the assumption that it can
# create the event loop at will without ever cleaning it up.
# As this tests runs several times in succession,
# that assumption becomes a problem.
LOGGER.info("Cleaning up loop.")
# Loop cleanup:
assert loop_for_this_test is not None
assert not loop_for_this_test.is_running()
loop_for_this_test.close()
asyncio.set_event_loop(None)
# We would like to leave it at that,
# but doing so causes the next test to fail.
#
# We did not find asyncio - API to reset the loop
# to "back to square one, as if just freshly started".
#
# The following code does not feel right, it's a kludge,
# but it apparently works for now:
if sys.platform == 'win32':
# For this case, the auto module has special code
# (at module load time! 😟) which we reluctantly reproduce here:
asyncio.set_event_loop(asyncio.ProactorEventLoop())
else:
asyncio.set_event_loop(asyncio.new_event_loop())
@pytest.fixture(scope="module",
params=["https://example.org",
"https://example.org:1234/blog",
"https://example.org:3456/blog/",
"http://example.org/deep/down/a/rabbit/hole"
])
def site_and_base_path(request) -> Tuple[MyFakeSite, str]:
"""Return a fake site and the base_path (root) the dev server should be serving."""
assert OUTPUT_FOLDER.is_dir(), \
f"Could not find dev server test fixture {OUTPUT_FOLDER.as_posix()}"
config = {
"post_pages": [],
"FILES_FOLDERS": [],
"GALLERY_FOLDERS": [],
"LISTINGS_FOLDERS": [],
"IMAGE_FOLDERS": [],
"SITE_URL": request.param,
"OUTPUT_FOLDER": OUTPUT_FOLDER.as_posix(),
}
return (MyFakeSite(config), auto.base_path_from_siteuri(request.param))
@pytest.fixture(scope="module")
def expected_text():
"""Read the index.html file from the fixture folder and return most of it.
For life reload, the server will fiddle with HTML <head>,
so this only returns everything after the opening <body> tag.
"""
with open(OUTPUT_FOLDER / "index.html", encoding="utf-8") as html_file:
all_html = html_file.read()
return all_html[all_html.find("<body>"):]
@pytest.mark.parametrize(("uri", "expected_basepath"), [
("http://localhost", ""),
("http://local.host", ""),
("http://localhost/", ""),
("http://local.host/", ""),
("http://localhost:123/", ""),
("http://local.host:456/", ""),
("https://localhost", ""),
("https://local.host", ""),
("https://localhost/", ""),
("https://local.host/", ""),
("https://localhost:123/", ""),
("https://local.host:456/", ""),
("http://example.org/blog", "/blog"),
("https://lorem.ipsum/dolet/", "/dolet"),
("http://example.org:124/blog", "/blog"),
("http://example.org:124/Deep/Rab_bit/hol.e/", "/Deep/Rab_bit/hol.e"),
# Would anybody in a sane mind actually do this?
("http://example.org:124/blog?lorem=ipsum&dol=et", "/blog"),
])
def test_basepath(uri: str, expected_basepath: Optional[str]) -> None:
assert expected_basepath == auto.base_path_from_siteuri(uri)
|
