product-fpol/tests/test_server.py

import contextlib
import http.client
import json
import pathlib
import re
import signal
import socket
import subprocess
import sys
import tempfile
import time
import unittest

SERVER = pathlib.Path(sys.argv.pop(1)).resolve()

@contextlib.contextmanager
def running_server(*arguments, stop_signal=signal.SIGTERM):
  with tempfile.TemporaryDirectory(prefix="fpol_test_") as directory:
    log_path = pathlib.Path(directory) / "server.log"
    with log_path.open("w") as log:
      process = subprocess.Popen([str(SERVER), "--port", "0", *map(str, arguments)], stdout=log, stderr=log)
      try:
        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
          output = log_path.read_text()
          match = re.search(r"Listening on http://127\.0\.0\.1:(\d+)", output)
          if match and process.poll() is None:
            yield int(match.group(1))
            break
          if process.poll() is not None:
            raise AssertionError(f"Server exited during startup:\n{output}")
          time.sleep(0.02)
        else:
          raise AssertionError(f"Server did not start:\n{log_path.read_text()}")
      finally:
        if process.poll() is None:
          process.send_signal(stop_signal)
        try:
          return_code = process.wait(timeout=5)
        except subprocess.TimeoutExpired:
          process.kill()
          process.wait()
          raise AssertionError("Server did not shut down within five seconds") from None
        if return_code != 0:
          raise AssertionError(f"Server exited with {return_code}:\n{log_path.read_text()}")

def request(port, path, method="GET"):
  connection = http.client.HTTPConnection("127.0.0.1", port, timeout=3)
  try:
    connection.request(method, path)
    response = connection.getresponse()
    return response.status, response.getheader("Content-Type"), response.read()
  finally:
    connection.close()

class server_tests(unittest.TestCase):
  def test_health_and_api_only_routes(self):
    with running_server() as port:
      status, content_type, body = request(port, "/api/health")
      self.assertEqual(status, 200)
      self.assertEqual(content_type, "application/json")
      self.assertEqual(json.loads(body), {"ok": True, "payload": {"status": "ok"}})
      for path in ("/", "/api/unknown", "/missing.js"):
        with self.subTest(path=path):
          status, _, body = request(port, path)
          self.assertEqual(status, 404)
          self.assertEqual(json.loads(body)["code"], "not_found")
      self.assertEqual(request(port, "/api/health", "POST")[0], 404)
  def test_static_assets_and_route_boundaries(self):
    with tempfile.TemporaryDirectory(prefix="fpol_web_") as directory:
      root = pathlib.Path(directory)
      (root / "assets").mkdir()
      (root / "api").mkdir()
      (root / "index.html").write_text("<!doctype html><title>Fantasy Politics</title>")
      (root / "assets" / "app.js").write_text("console.log('ready');")
      (root / "assets" / "app.css").write_text("body { color: green; }")
      (root / "assets" / "image.png").write_bytes(b"\x89PNG\r\n\x1a\n\x00\xff")
      (root / ".env").write_text("not a public asset")
      (root / ".hidden").mkdir()
      (root / ".hidden" / "data.txt").write_text("not a public asset")
      (root / "api" / "health").write_text("must not replace the health route")
      (root / "linked.html").symlink_to(root / "index.html")
      (root / "linked_assets").symlink_to(root / "assets", target_is_directory=True)
      with running_server("--web-root", root) as port:
        for route, filename, content_type in (
          ("/", "index.html", "text/html; charset=utf-8"),
          ("/index.html", "index.html", "text/html; charset=utf-8"),
          ("/assets/app.js", "assets/app.js", "text/javascript; charset=utf-8"),
          ("/assets/app.css", "assets/app.css", "text/css; charset=utf-8"),
          ("/assets/image.png", "assets/image.png", "image/png"),
        ):
          with self.subTest(route=route):
            self.assertEqual(request(port, route), (200, content_type, (root / filename).read_bytes()))
        for route in ("/linked.html", "/linked_assets/app.js", "/.env", "/.hidden/data.txt", "/../index.html", "/%2e%2e/index.html", "/missing.js", "/api/missing"):
          with self.subTest(route=route):
            self.assertEqual(request(port, route)[0], 404)
        self.assertEqual(json.loads(request(port, "/api/health")[2])["payload"], {"status": "ok"})
  def test_invalid_startup(self):
    with tempfile.TemporaryDirectory(prefix="fpol_invalid_") as directory:
      root = pathlib.Path(directory)
      file = root / "file.txt"
      file.write_text("not a web root")
      for arguments, diagnostic in (
        (["--unknown"], "Unknown option"),
        (["--port"], "Missing value"),
        (["--port", "abc"], "Port must be an integer"),
        (["--port", "65536"], "Port must be an integer"),
        (["--port", "-1"], "Port must be an integer"),
        (["--port", "80x"], "Port must be an integer"),
        (["--host", ""], "Missing value"),
        (["--web-root", root / "missing"], "Web root must be a directory"),
        (["--web-root", file], "Web root must be a directory"),
        (["--web-root", root], "Web root must contain index.html"),
      ):
        with self.subTest(arguments=arguments):
          result = subprocess.run([str(SERVER), *map(str, arguments)], capture_output=True, text=True, timeout=5)
          self.assertNotEqual(result.returncode, 0)
          self.assertIn("[error]", result.stderr)
          self.assertIn(diagnostic, result.stderr)
      (root / "index.html").symlink_to(file)
      result = subprocess.run([str(SERVER), "--web-root", str(root)], capture_output=True, text=True, timeout=5)
      self.assertNotEqual(result.returncode, 0)
      self.assertIn("Web root must contain index.html", result.stderr)
  def test_occupied_port(self):
    with socket.socket() as listener:
      listener.bind(("127.0.0.1", 0))
      listener.listen()
      result = subprocess.run([str(SERVER), "--port", str(listener.getsockname()[1])], capture_output=True, text=True, timeout=5)
      self.assertNotEqual(result.returncode, 0)
      self.assertIn("[error]", result.stderr)
  def test_help(self):
    result = subprocess.run([str(SERVER), "--help"], capture_output=True, text=True, timeout=5)
    self.assertEqual(result.returncode, 0)
    self.assertIn("--web-root", result.stdout)
  def test_interrupt_shutdown(self):
    with running_server(stop_signal=signal.SIGINT) as port:
      self.assertEqual(request(port, "/api/health")[0], 200)

if __name__ == "__main__":
  unittest.main()