93 lines
4.3 KiB
Python
93 lines
4.3 KiB
Python
"""Run from the assignment directory: python -m unittest discover -s tests -p test_starter.py."""
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from click.testing import CliRunner
|
|
|
|
STARTER = Path(__file__).resolve().parents[1] / "solution" / "server.py"
|
|
spec = importlib.util.spec_from_file_location("hw3_starter", STARTER)
|
|
starter = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = starter
|
|
spec.loader.exec_module(starter)
|
|
ENV_NAMES = ("SERVER_HOST", "SERVER_PORT", "SERVER_DOMAIN", "SERVER_WORKING_DIRECTORY")
|
|
|
|
|
|
class StarterTests(unittest.TestCase):
|
|
def run_config(self, args, env):
|
|
clean_env = {k: v for k, v in os.environ.items() if k not in ENV_NAMES}
|
|
clean_env.update(env)
|
|
with patch.dict(os.environ, clean_env, clear=True), patch.object(starter.socket, "socket") as factory, patch.object(starter, "HTTPServer") as server:
|
|
factory.return_value.accept.side_effect = OSError("stop after bind")
|
|
result = CliRunner().invoke(starter.main, args)
|
|
return result, factory, server
|
|
|
|
def test_defaults(self):
|
|
result, factory, server = self.run_config(["--working-directory", str(STARTER.parent)], {})
|
|
self.assertEqual(result.exit_code, 0, result.output)
|
|
factory.return_value.bind.assert_called_once_with(("0.0.0.0", 8080))
|
|
self.assertEqual(server.call_args.args[2:], ("localhost", STARTER.parent))
|
|
|
|
def test_environment(self):
|
|
env = dict(zip(ENV_NAMES, ("127.0.0.1", "9090", "example.com", str(STARTER.parent))))
|
|
result, factory, server = self.run_config([], env)
|
|
self.assertEqual(result.exit_code, 0, result.output)
|
|
factory.return_value.bind.assert_called_once_with(("127.0.0.1", 9090))
|
|
self.assertEqual(server.call_args.args[2:], ("example.com", STARTER.parent))
|
|
|
|
def test_cli_overrides_environment(self):
|
|
env = dict(zip(ENV_NAMES, ("8.8.8.8", "80", "wrong.example", "wrong-directory")))
|
|
result, factory, server = self.run_config(
|
|
["--host", "127.0.0.1", "--port", "9091", "--server-domain", "example.com",
|
|
"--working-directory", str(STARTER.parent)], env)
|
|
self.assertEqual(result.exit_code, 0, result.output)
|
|
factory.return_value.bind.assert_called_once_with(("127.0.0.1", 9091))
|
|
self.assertEqual(server.call_args.args[2:], ("example.com", STARTER.parent))
|
|
|
|
def test_missing_directory(self):
|
|
for args, env in [([], {}), ([], {"SERVER_WORKING_DIRECTORY": ""}),
|
|
(["--working-directory", ""], {"SERVER_WORKING_DIRECTORY": str(STARTER.parent)})]:
|
|
with self.subTest(args=args, env=env):
|
|
result, factory, _ = self.run_config(args, env)
|
|
self.assertEqual(result.exit_code, 1, result.output)
|
|
factory.assert_not_called()
|
|
|
|
def test_tcp_startup(self):
|
|
with socket.socket() as probe:
|
|
probe.bind(("127.0.0.1", 0))
|
|
port = probe.getsockname()[1]
|
|
with tempfile.TemporaryDirectory() as workdir:
|
|
process = subprocess.Popen(
|
|
[sys.executable, str(STARTER), "--host", "127.0.0.1", "--port", str(port),
|
|
"--working-directory", workdir],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
deadline = time.monotonic() + 5
|
|
while True:
|
|
self.assertIsNone(process.poll(), "starter exited before accepting a connection")
|
|
try:
|
|
client = socket.create_connection(("127.0.0.1", port), timeout=0.5)
|
|
break
|
|
except OSError:
|
|
if time.monotonic() >= deadline:
|
|
self.fail("starter did not bind within 5 seconds")
|
|
time.sleep(0.05)
|
|
with client:
|
|
client.sendall(b"GET / HTTP/1.1\r\n")
|
|
# The starter has no HTTP handler yet, but its TCP lifecycle works.
|
|
self.assertEqual(client.recv(1), b"")
|
|
finally:
|
|
process.terminate()
|
|
process.wait(timeout=5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|