from datetime import date, datetime import json import os from pathlib import Path import socket from tempfile import TemporaryDirectory from threading import Thread import unittest from urllib.error import HTTPError from urllib.request import Request, urlopen from roster_formatter import DEFAULT_LAYOUT, LayoutSettings, RosterPeriod, format_roster from server import OutputStore, RosterHTTPServer, RosterWebApp from tests.helpers import make_geometry_pdf def multipart_body(fields: dict[str, str], filename: str, file_data: bytes) -> tuple[str, bytes]: boundary = "----RosterRegressionBoundary" chunks: list[bytes] = [] for name, value in fields.items(): chunks.extend( [ f"--{boundary}\r\n".encode(), f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(), value.encode(), b"\r\n", ] ) chunks.extend( [ f"--{boundary}\r\n".encode(), ( 'Content-Disposition: form-data; name="roster"; ' f'filename="{filename}"\r\n' ).encode(), b"Content-Type: application/pdf\r\n\r\n", file_data, b"\r\n", f"--{boundary}--\r\n".encode(), ] ) return f"multipart/form-data; boundary={boundary}", b"".join(chunks) class OutputStoreTests(unittest.TestCase): def test_layout_defaults_persist_across_store_restart_and_prune(self) -> None: with TemporaryDirectory() as temp_dir: root = Path(temp_dir) settings = LayoutSettings( content_shift=112, bottom_margin=64, logo_x=57, logo_y=-8, logo_scale=0.31, date_x=472, date_y=91, date_font_size=52, ) store = OutputStore(root, retention_months=3) store.save_layout_defaults(settings) defaults_path = root / ".roster-layout-defaults.json" self.assertTrue(defaults_path.is_file()) self.assertEqual( {"version": 1, "layout": settings.as_json_values()}, json.loads(defaults_path.read_text(encoding="utf-8")), ) restarted_store = OutputStore(root, retention_months=3) self.assertEqual(settings, restarted_store.load_layout_defaults()) os.utime(defaults_path, (datetime(2020, 1, 1).timestamp(),) * 2) self.assertEqual([], restarted_store.prune(date(2026, 7, 25))) self.assertTrue(defaults_path.is_file()) self.assertEqual(settings, restarted_store.load_layout_defaults()) def test_invalid_layout_defaults_fall_back_to_factory_values(self) -> None: with TemporaryDirectory() as temp_dir: root = Path(temp_dir) store = OutputStore(root, retention_months=3) store.layout_defaults_path.write_text("{not valid json", encoding="utf-8") with self.assertLogs("roster.web", level="WARNING"): loaded = store.load_layout_defaults() self.assertEqual(DEFAULT_LAYOUT, loaded) def test_retention_removes_only_matching_rosters_older_than_window(self) -> None: with TemporaryDirectory() as temp_dir: root = Path(temp_dir) for filename in ( "MARCH 2026 Roster.png", "APRIL 2026 Roster.png", "JULY 2026 Roster.png", "JAN 2099 Roster.png", "notes.png", "shifted.pdf", ): (root / filename).write_bytes(b"test") os.utime( root / "JAN 2099 Roster.png", (datetime(2026, 3, 31, 12).timestamp(),) * 2, ) store = OutputStore(root, retention_months=3) removed = store.prune(date(2026, 7, 25)) self.assertEqual( ["JAN 2099 Roster.png", "MARCH 2026 Roster.png"], removed, ) self.assertFalse((root / "MARCH 2026 Roster.png").exists()) self.assertTrue((root / "APRIL 2026 Roster.png").exists()) self.assertTrue((root / "notes.png").exists()) self.assertTrue((root / "shifted.pdf").exists()) def test_symlink_is_never_listed_read_or_pruned(self) -> None: with TemporaryDirectory() as temp_dir: root = Path(temp_dir) secret = root / "private.txt" secret.write_bytes(b"private data") link = root / "JAN 2020 Roster.png" link.symlink_to(secret.name) store = OutputStore(root, retention_months=3) self.assertIsNone(store.read(link.name)) self.assertEqual([], store.list_items()) self.assertEqual([], store.prune(date(2026, 7, 25))) self.assertTrue(link.is_symlink()) def test_zero_retention_disables_cleanup(self) -> None: with TemporaryDirectory() as temp_dir: root = Path(temp_dir) old = root / "JAN 2020 Roster.png" old.write_bytes(b"test") self.assertEqual([], OutputStore(root, retention_months=0).prune(date(2026, 7, 25))) self.assertTrue(old.exists()) class WebServerTests(unittest.TestCase): def setUp(self) -> None: self.temporary = TemporaryDirectory() self.root = Path(self.temporary.name) self.output_dir = self.root / "output" self.source = make_geometry_pdf(self.root / "source.pdf") self.app = RosterWebApp( self.output_dir, retention_months=0, allowed_hosts={"roster.example.test"}, ) self.server = RosterHTTPServer(("127.0.0.1", 0), self.app) self.thread = Thread(target=self.server.serve_forever, daemon=True) self.thread.start() host, port = self.server.server_address self.base_url = f"http://{host}:{port}" def tearDown(self) -> None: self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) self.temporary.cleanup() def test_health_and_index(self) -> None: with urlopen(f"{self.base_url}/health", timeout=5) as response: self.assertEqual({"status": "ok"}, json.loads(response.read())) with urlopen(f"{self.base_url}/", timeout=5) as response: html = response.read().decode() self.assertIn("Roster Formatter", html) self.assertNotIn("__DEFAULT_MONTH__", html) self.assertNotIn("__LOGO_SCALE__", html) self.assertIn('data-max-upload-mb="20"', html) def test_layout_defaults_endpoint_persists_canonical_values_in_index(self) -> None: submitted = DEFAULT_LAYOUT.as_form_values() submitted.update( { "content_shift": "123", "bottom_margin": "65.5", "logo_x": "58", "logo_y": "-7", "logo_scale": "0.310", "date_x": "475", "date_y": "92", "date_font_size": "53", } ) expected = LayoutSettings.from_mapping(submitted).as_form_values() request = Request( f"{self.base_url}/api/layout-defaults", data=json.dumps(submitted).encode("utf-8"), method="POST", headers={"Content-Type": "application/json"}, ) with urlopen(request, timeout=5) as response: self.assertEqual( {"status": "saved", "defaults": expected}, json.loads(response.read()), ) self.assertTrue((self.output_dir / ".roster-layout-defaults.json").is_file()) restarted_store = OutputStore(self.output_dir, retention_months=0) self.assertEqual(LayoutSettings.from_mapping(submitted), restarted_store.load_layout_defaults()) with urlopen(f"{self.base_url}/", timeout=5) as response: html = response.read().decode() self.assertIn( 'name="content_shift" type="number" value="123.0" ' 'data-default="123.0" data-factory="100"', html, ) self.assertIn( 'name="logo_scale" type="number" value="0.31" ' 'data-default="0.31" data-factory="0.27"', html, ) self.assertIn( 'name="date_font_size" type="number" value="53" ' 'data-default="53" data-factory="48"', html, ) self.assertNotIn("__FACTORY_", html) def test_layout_defaults_endpoint_requires_every_layout_field(self) -> None: submitted = DEFAULT_LAYOUT.as_form_values() submitted.pop("logo_scale") request = Request( f"{self.base_url}/api/layout-defaults", data=json.dumps(submitted).encode("utf-8"), method="POST", headers={"Content-Type": "application/json"}, ) with self.assertRaises(HTTPError) as caught: urlopen(request, timeout=5) self.assertEqual(422, caught.exception.code) self.assertEqual( {"error": "Submit every layout setting before saving defaults."}, json.loads(caught.exception.read()), ) self.assertFalse((self.output_dir / ".roster-layout-defaults.json").exists()) def test_theme_defaults_to_system_and_static_assets_expose_all_choices(self) -> None: with urlopen(f"{self.base_url}/", timeout=5) as response: html = response.read().decode() self.assertIn('', html) self.assertLess(html.index('/static/theme.js'), html.index('/static/styles.css')) for choice in ("system", "light", "dark"): self.assertIn(f'data-theme-choice="{choice}"', html) with urlopen(f"{self.base_url}/static/theme.js", timeout=5) as response: self.assertTrue(response.headers["Content-Type"].startswith("text/javascript")) theme_script = response.read().decode() self.assertIn('new Set(["system", "light", "dark"])', theme_script) self.assertIn('localStorage.getItem("roster-theme")', theme_script) self.assertIn("document.documentElement.dataset.theme = theme", theme_script) with urlopen(f"{self.base_url}/static/styles.css", timeout=5) as response: styles = response.read().decode() self.assertIn(':root[data-theme="dark"]', styles) self.assertIn("prefers-color-scheme: dark", styles) def test_untrusted_host_is_rejected(self) -> None: request = Request( f"{self.base_url}/api/history", headers={"Host": "attacker.example"}, ) with self.assertRaises(HTTPError) as caught: urlopen(request, timeout=5) self.assertEqual(403, caught.exception.code) self.assertEqual( {"error": "Host is not allowed."}, json.loads(caught.exception.read()), ) def test_configured_caddy_host_is_allowed(self) -> None: request = Request( f"{self.base_url}/health", headers={"Host": "roster.example.test"}, ) with urlopen(request, timeout=5) as response: self.assertEqual({"status": "ok"}, json.loads(response.read())) def test_web_output_is_byte_identical_to_shared_renderer(self) -> None: direct = self.root / "direct.png" format_roster(self.source, direct, RosterPeriod(2026, 7)) fields = DEFAULT_LAYOUT.as_form_values() fields["month"] = "2026-07" content_type, body = multipart_body( fields, "current-roster.pdf", self.source.read_bytes(), ) request = Request( f"{self.base_url}/format", data=body, method="POST", headers={"Content-Type": content_type}, ) with urlopen(request, timeout=15) as response: web_png = response.read() self.assertEqual("true", response.headers["X-Roster-Saved"]) self.assertIn("JULY 2026 Roster.png", response.headers["Content-Disposition"]) self.assertEqual(direct.read_bytes(), web_png) self.assertEqual(web_png, (self.output_dir / "JULY 2026 Roster.png").read_bytes()) def test_format_missing_layout_fields_uses_saved_defaults(self) -> None: settings = LayoutSettings( content_shift=116, bottom_margin=67, logo_x=54, logo_y=-6, logo_scale=0.3, date_x=468, date_y=94, date_font_size=51, ) self.app.store.save_layout_defaults(settings) direct = self.root / "direct-saved-defaults.png" format_roster(self.source, direct, RosterPeriod(2026, 8), settings) content_type, body = multipart_body( {"month": "2026-08"}, "current-roster.pdf", self.source.read_bytes(), ) request = Request( f"{self.base_url}/format", data=body, method="POST", headers={"Content-Type": content_type}, ) with urlopen(request, timeout=15) as response: web_png = response.read() self.assertEqual(direct.read_bytes(), web_png) self.assertEqual(web_png, (self.output_dir / "AUG 2026 Roster.png").read_bytes()) def test_rejected_post_cannot_smuggle_a_second_request(self) -> None: inner_content_type, inner_body = multipart_body( {"month": "2026-07"}, "smuggled.pdf", self.source.read_bytes(), ) host, port = self.server.server_address inner_request = ( f"POST /format HTTP/1.1\r\n" f"Host: {host}:{port}\r\n" f"Content-Type: {inner_content_type}\r\n" f"Content-Length: {len(inner_body)}\r\n" "Connection: close\r\n" "\r\n" ).encode() + inner_body outer_request = ( f"POST /format HTTP/1.1\r\n" f"Host: {host}:{port}\r\n" "Origin: http://attacker.example\r\n" "Content-Type: text/plain\r\n" f"Content-Length: {len(inner_request)}\r\n" "\r\n" ).encode() + inner_request response_parts: list[bytes] = [] with socket.create_connection((host, port), timeout=5) as connection: connection.sendall(outer_request) connection.shutdown(socket.SHUT_WR) while True: try: chunk = connection.recv(65536) except ConnectionResetError: break if not chunk: break response_parts.append(chunk) response = b"".join(response_parts) self.assertEqual(1, response.count(b"HTTP/1.1")) self.assertIn(b"403 Forbidden", response) self.assertIn(b"Connection: close", response) self.assertEqual([], list(self.output_dir.glob("*.png"))) def test_invalid_upload_gets_safe_error(self) -> None: content_type, body = multipart_body( {"month": "2026-07"}, "not-a-roster.pdf", b"this is not a pdf", ) request = Request( f"{self.base_url}/format", data=body, method="POST", headers={"Content-Type": content_type}, ) with self.assertRaises(HTTPError) as caught: urlopen(request, timeout=5) self.assertEqual(422, caught.exception.code) self.assertEqual("close", caught.exception.headers["Connection"]) self.assertEqual( {"error": "The uploaded file is not a PDF."}, json.loads(caught.exception.read()), ) if __name__ == "__main__": unittest.main()