787 lines
30 KiB
Python
787 lines
30 KiB
Python
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 PIL import Image
|
|
|
|
from roster_formatter import (
|
|
DEFAULT_LAYOUT,
|
|
PNG_HEIGHT,
|
|
PNG_WIDTH,
|
|
LayoutSettings,
|
|
RosterFormatError,
|
|
RosterPeriod,
|
|
format_roster,
|
|
)
|
|
from server import (
|
|
LAYOUT_DEFAULTS_VERSION,
|
|
OutputStore,
|
|
RequestError,
|
|
RosterHTTPServer,
|
|
RosterWebApp,
|
|
UploadedFile,
|
|
parse_multipart,
|
|
)
|
|
from tests.helpers import (
|
|
make_geometry_pdf,
|
|
make_main_roster_pdf,
|
|
make_nursery_only_roster_pdf,
|
|
)
|
|
|
|
|
|
def multipart_body(
|
|
fields: dict[str, str],
|
|
filename: str | None = None,
|
|
file_data: bytes | None = None,
|
|
*,
|
|
uploads: list[tuple[str, bytes]] | None = None,
|
|
) -> tuple[str, bytes]:
|
|
"""Build a request with zero, one, or repeated ``roster`` file fields."""
|
|
|
|
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",
|
|
]
|
|
)
|
|
selected_uploads = uploads
|
|
if selected_uploads is None:
|
|
selected_uploads = [] if filename is None or file_data is None else [(filename, file_data)]
|
|
for selected_filename, selected_data in selected_uploads:
|
|
chunks.extend(
|
|
[
|
|
f"--{boundary}\r\n".encode(),
|
|
(
|
|
'Content-Disposition: form-data; name="roster"; '
|
|
f'filename="{selected_filename}"\r\n'
|
|
).encode(),
|
|
b"Content-Type: application/pdf\r\n\r\n",
|
|
selected_data,
|
|
b"\r\n",
|
|
]
|
|
)
|
|
chunks.extend(
|
|
[
|
|
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,
|
|
main_x=73,
|
|
main_y=128,
|
|
main_scale=0.79,
|
|
nursery_x=65,
|
|
nursery_y=515,
|
|
nursery_scale=0.91,
|
|
)
|
|
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": LAYOUT_DEFAULTS_VERSION, "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_v1_layout_defaults_load_with_factory_combined_values_and_migrate(self) -> None:
|
|
legacy_layout = {
|
|
"content_shift": 117,
|
|
"bottom_margin": 63,
|
|
"logo_x": 54,
|
|
"logo_y": -6,
|
|
"logo_scale": 0.3,
|
|
"date_x": 469,
|
|
"date_y": 92,
|
|
"date_font_size": 51,
|
|
}
|
|
expected = LayoutSettings.from_mapping(
|
|
{name: str(value) for name, value in legacy_layout.items()},
|
|
defaults=DEFAULT_LAYOUT,
|
|
)
|
|
|
|
with TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
store = OutputStore(root, retention_months=3)
|
|
store.layout_defaults_path.write_text(
|
|
json.dumps({"version": 1, "layout": legacy_layout}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
loaded = store.load_layout_defaults()
|
|
|
|
self.assertEqual(expected, loaded)
|
|
self.assertEqual(DEFAULT_LAYOUT.main_x, loaded.main_x)
|
|
self.assertEqual(DEFAULT_LAYOUT.main_y, loaded.main_y)
|
|
self.assertEqual(DEFAULT_LAYOUT.main_scale, loaded.main_scale)
|
|
self.assertEqual(DEFAULT_LAYOUT.nursery_x, loaded.nursery_x)
|
|
self.assertEqual(DEFAULT_LAYOUT.nursery_y, loaded.nursery_y)
|
|
self.assertEqual(DEFAULT_LAYOUT.nursery_scale, loaded.nursery_scale)
|
|
|
|
store.save_layout_defaults(loaded)
|
|
migrated = json.loads(store.layout_defaults_path.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(LAYOUT_DEFAULTS_VERSION, migrated["version"])
|
|
self.assertEqual(set(DEFAULT_LAYOUT.as_form_values()), set(migrated["layout"]))
|
|
self.assertEqual(expected.as_json_values(), migrated["layout"])
|
|
|
|
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 MultipartParserTests(unittest.TestCase):
|
|
def test_one_roster_file_is_accepted(self) -> None:
|
|
content_type, body = multipart_body(
|
|
{"month": "2026-09"},
|
|
"main.pdf",
|
|
b"%PDF-one",
|
|
)
|
|
|
|
values, uploads = parse_multipart(content_type, body)
|
|
|
|
self.assertEqual({"month": "2026-09"}, values)
|
|
self.assertEqual(1, len(uploads))
|
|
self.assertEqual("main.pdf", uploads[0].filename)
|
|
self.assertEqual(b"%PDF-one", uploads[0].data)
|
|
|
|
def test_two_repeated_roster_files_are_accepted_in_order(self) -> None:
|
|
content_type, body = multipart_body(
|
|
{"month": "2026-09"},
|
|
uploads=[
|
|
("nursery.pdf", b"%PDF-nursery"),
|
|
("main.pdf", b"%PDF-main"),
|
|
],
|
|
)
|
|
|
|
values, uploads = parse_multipart(content_type, body)
|
|
|
|
self.assertEqual({"month": "2026-09"}, values)
|
|
self.assertEqual(["nursery.pdf", "main.pdf"], [upload.filename for upload in uploads])
|
|
self.assertEqual(
|
|
[b"%PDF-nursery", b"%PDF-main"],
|
|
[upload.data for upload in uploads],
|
|
)
|
|
|
|
def test_zero_roster_files_is_rejected(self) -> None:
|
|
content_type, body = multipart_body({"month": "2026-09"}, uploads=[])
|
|
|
|
with self.assertRaises(RequestError) as caught:
|
|
parse_multipart(content_type, body)
|
|
|
|
self.assertEqual("Upload one or two roster PDFs.", caught.exception.message)
|
|
|
|
def test_three_roster_files_is_rejected(self) -> None:
|
|
content_type, body = multipart_body(
|
|
{"month": "2026-09"},
|
|
uploads=[
|
|
("first.pdf", b"%PDF-first"),
|
|
("second.pdf", b"%PDF-second"),
|
|
("third.pdf", b"%PDF-third"),
|
|
],
|
|
)
|
|
|
|
with self.assertRaises(RequestError) as caught:
|
|
parse_multipart(content_type, body)
|
|
|
|
self.assertEqual("Upload one or two roster PDFs.", caught.exception.message)
|
|
|
|
|
|
class RosterWebAppUnitTests(unittest.TestCase):
|
|
@staticmethod
|
|
def _uploaded(path: Path) -> UploadedFile:
|
|
return UploadedFile(
|
|
filename=path.name,
|
|
content_type="application/pdf",
|
|
data=path.read_bytes(),
|
|
)
|
|
|
|
def test_combined_format_retains_only_the_final_png(self) -> None:
|
|
with TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
output_dir = root / "output"
|
|
main = make_main_roster_pdf(root / "main.pdf")
|
|
nursery = make_nursery_only_roster_pdf(root / "nursery.pdf")
|
|
app = RosterWebApp(output_dir, retention_months=0)
|
|
values = DEFAULT_LAYOUT.as_form_values()
|
|
values["month"] = "2026-09"
|
|
|
|
png, filename, saved, removed = app.format_upload(
|
|
[self._uploaded(main), self._uploaded(nursery)],
|
|
values,
|
|
)
|
|
|
|
self.assertEqual("SEP 2026 Roster.png", filename)
|
|
self.assertTrue(saved)
|
|
self.assertEqual([], removed)
|
|
self.assertEqual(png, (output_dir / filename).read_bytes())
|
|
self.assertEqual([filename], sorted(path.name for path in output_dir.iterdir()))
|
|
self.assertFalse(list(output_dir.glob("*.pdf")))
|
|
self.assertFalse(list(output_dir.glob("*.tmp")))
|
|
with Image.open(output_dir / filename) as image:
|
|
self.assertEqual((PNG_WIDTH, PNG_HEIGHT), image.size)
|
|
|
|
def test_each_pdf_is_independently_validated(self) -> None:
|
|
with TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
output_dir = root / "output"
|
|
main = make_main_roster_pdf(root / "main.pdf")
|
|
nursery = make_nursery_only_roster_pdf(root / "nursery.pdf")
|
|
valid_main = self._uploaded(main)
|
|
valid_nursery = self._uploaded(nursery)
|
|
invalid = UploadedFile("damaged.pdf", "application/pdf", b"not a pdf")
|
|
app = RosterWebApp(output_dir, retention_months=0)
|
|
|
|
for uploads, expected_index in (
|
|
([invalid, valid_nursery], 1),
|
|
([valid_main, invalid], 2),
|
|
):
|
|
with self.subTest(invalid_pdf=expected_index):
|
|
with self.assertRaisesRegex(
|
|
RosterFormatError,
|
|
rf"PDF {expected_index}: The uploaded file is not a PDF",
|
|
):
|
|
app.format_upload(uploads, {"month": "2026-09"})
|
|
|
|
self.assertEqual([], list(output_dir.iterdir()))
|
|
|
|
def test_filename_validation_is_independent_for_second_pdf(self) -> None:
|
|
with TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
main = make_main_roster_pdf(root / "main.pdf")
|
|
nursery = make_nursery_only_roster_pdf(root / "nursery.pdf")
|
|
app = RosterWebApp(root / "output", retention_months=0)
|
|
mislabeled = UploadedFile(
|
|
"nursery.txt",
|
|
"application/pdf",
|
|
nursery.read_bytes(),
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
RosterFormatError,
|
|
r"PDF 2: Choose a file whose name ends in \.pdf",
|
|
):
|
|
app.format_upload([self._uploaded(main), mislabeled], {"month": "2026-09"})
|
|
|
|
def test_per_file_limit_identifies_the_oversized_second_pdf(self) -> None:
|
|
with TemporaryDirectory() as temp_dir:
|
|
app = RosterWebApp(
|
|
Path(temp_dir) / "output",
|
|
retention_months=0,
|
|
max_upload_bytes=5,
|
|
)
|
|
first = UploadedFile("first.pdf", "application/pdf", b"%PDF")
|
|
second = UploadedFile("second.pdf", "application/pdf", b"123456")
|
|
|
|
with self.assertRaises(RequestError) as caught:
|
|
app.format_upload([first, second], {"month": "2026-09"})
|
|
|
|
self.assertIn("PDF 2:", caught.exception.message)
|
|
self.assertIn("Each PDF must be", caught.exception.message)
|
|
|
|
def test_request_limit_leaves_room_for_two_individually_valid_files(self) -> None:
|
|
with TemporaryDirectory() as temp_dir:
|
|
per_file_bytes = 20 * 1024 * 1024
|
|
app = RosterWebApp(
|
|
Path(temp_dir) / "output",
|
|
retention_months=0,
|
|
max_upload_bytes=per_file_bytes,
|
|
)
|
|
|
|
self.assertGreater(app.max_request_bytes, 2 * per_file_bytes)
|
|
|
|
|
|
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)
|
|
self.assertEqual(1, html.count('name="roster" type="file"'))
|
|
self.assertIn('name="roster" type="file" accept="application/pdf,.pdf" multiple', html)
|
|
self.assertIn('id="selected-files"', html)
|
|
self.assertIn("one main roster, optional nursery roster", html)
|
|
self.assertIn("<legend>Main roster</legend>", html)
|
|
self.assertIn("<legend>Nursery-only roster</legend>", html)
|
|
for field in (
|
|
"main_x",
|
|
"main_y",
|
|
"main_scale",
|
|
"nursery_x",
|
|
"nursery_y",
|
|
"nursery_scale",
|
|
):
|
|
self.assertIn(f'name="{field}"', html)
|
|
self.assertNotIn('name="main_roster"', html)
|
|
self.assertNotIn('name="nursery_roster"', html)
|
|
|
|
def test_static_script_supports_picker_and_drop_for_two_files(self) -> None:
|
|
with urlopen(f"{self.base_url}/static/app.js", timeout=5) as response:
|
|
script = response.read().decode()
|
|
|
|
self.assertIn("let selectedFiles = [];", script)
|
|
self.assertIn("if (files.length > 2)", script)
|
|
self.assertIn("chooseFiles(fileInput.files)", script)
|
|
self.assertIn("chooseFiles(files)", script)
|
|
self.assertIn(
|
|
'selectedFiles.forEach((file) => formData.append("roster", file, file.name))',
|
|
script,
|
|
)
|
|
self.assertIn("roster types are identified automatically", script)
|
|
self.assertIn("input.value = input.dataset.default", script)
|
|
self.assertIn("input.value = input.dataset.factory", script)
|
|
|
|
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",
|
|
"main_x": "72",
|
|
"main_y": "126",
|
|
"main_scale": "0.810",
|
|
"nursery_x": "68",
|
|
"nursery_y": "514",
|
|
"nursery_scale": "0.920",
|
|
}
|
|
)
|
|
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.assertIn(
|
|
'name="main_x" type="number" value="72.0" '
|
|
'data-default="72.0" data-factory="66"',
|
|
html,
|
|
)
|
|
self.assertIn(
|
|
'name="nursery_scale" type="number" value="0.92" '
|
|
'data-default="0.92" data-factory="0.85"',
|
|
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 lang="en" data-theme="system">', 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_two_pdf_upload_returns_combined_png_and_no_working_files(self) -> None:
|
|
main = make_main_roster_pdf(self.root / "main.pdf")
|
|
nursery = make_nursery_only_roster_pdf(self.root / "nursery.pdf")
|
|
fields = DEFAULT_LAYOUT.as_form_values()
|
|
fields["month"] = "2026-09"
|
|
content_type, body = multipart_body(
|
|
fields,
|
|
uploads=[
|
|
(nursery.name, nursery.read_bytes()),
|
|
(main.name, main.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:
|
|
png = response.read()
|
|
self.assertIn("SEP 2026 Roster.png", response.headers["Content-Disposition"])
|
|
|
|
output = self.output_dir / "SEP 2026 Roster.png"
|
|
self.assertEqual(png, output.read_bytes())
|
|
self.assertEqual([output.name], sorted(path.name for path in self.output_dir.iterdir()))
|
|
self.assertFalse(list(self.output_dir.glob("*.pdf")))
|
|
self.assertFalse(list(self.output_dir.glob("*.tmp")))
|
|
with Image.open(output) as image:
|
|
self.assertEqual((PNG_WIDTH, PNG_HEIGHT), image.size)
|
|
|
|
def test_two_main_pdfs_return_clear_ambiguity_error(self) -> None:
|
|
first = make_main_roster_pdf(self.root / "first.pdf")
|
|
second = make_main_roster_pdf(self.root / "second.pdf")
|
|
content_type, body = multipart_body(
|
|
{"month": "2026-09"},
|
|
uploads=[
|
|
(first.name, first.read_bytes()),
|
|
(second.name, second.read_bytes()),
|
|
],
|
|
)
|
|
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=10)
|
|
|
|
self.assertEqual(422, caught.exception.code)
|
|
error = json.loads(caught.exception.read())["error"]
|
|
self.assertIn("could not determine one full roster and one nursery-only roster", error)
|
|
self.assertIn("PDF 1: main", error)
|
|
self.assertIn("PDF 2: main", error)
|
|
self.assertEqual([], list(self.output_dir.iterdir()))
|
|
|
|
def test_request_level_limit_rejects_total_before_individual_file_limits(self) -> None:
|
|
main = make_main_roster_pdf(self.root / "main.pdf")
|
|
nursery = make_nursery_only_roster_pdf(self.root / "nursery.pdf")
|
|
content_type, body = multipart_body(
|
|
{"month": "2026-09"},
|
|
uploads=[
|
|
(main.name, main.read_bytes()),
|
|
(nursery.name, nursery.read_bytes()),
|
|
],
|
|
)
|
|
self.assertLess(len(main.read_bytes()), self.app.max_upload_bytes)
|
|
self.assertLess(len(nursery.read_bytes()), self.app.max_upload_bytes)
|
|
self.app.max_request_bytes = len(body) - 1
|
|
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(413, caught.exception.code)
|
|
self.assertEqual({"error": "The upload is too large."}, json.loads(caught.exception.read()))
|
|
self.assertEqual([], list(self.output_dir.iterdir()))
|
|
|
|
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()
|