diff options
Diffstat (limited to 'tests/integration/test_dev_server.py')
| -rw-r--r-- | tests/integration/test_dev_server.py | 223 |
1 files changed, 223 insertions, 0 deletions
diff --git a/tests/integration/test_dev_server.py b/tests/integration/test_dev_server.py new file mode 100644 index 0000000..769678f --- /dev/null +++ b/tests/integration/test_dev_server.py @@ -0,0 +1,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) |