feat: add private image upload serviceHEADmain

1 files changed, 118 insertions, 0 deletions

diff --git a/tests/test_server.py b/tests/test_server.py
new file mode 100644
index 0000000..aaff61e
--- /dev/null
+++ b/tests/test_server.py
@@ -0,0 +1,118 @@
+import http.client
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+from server import MAX_FILE_BYTES, detect_image_type, make_server
+
+
+PNG = b"\x89PNG\r\n\x1a\n" + b"test-image"
+
+
+def multipart(filename: str, payload: bytes, boundary: str = "test-boundary") -> tuple[bytes, str]:
+    body = (
+        f"--{boundary}\r\n"
+        f'Content-Disposition: form-data; name="image"; filename="{filename}"\r\n'
+        "Content-Type: application/octet-stream\r\n\r\n"
+    ).encode() + payload + f"\r\n--{boundary}--\r\n".encode()
+    return body, f"multipart/form-data; boundary={boundary}"
+
+
+class ServerTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.upload_dir = Path(self.temp.name) / "private"
+        self.server = make_server("127.0.0.1", 0, self.upload_dir)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.port = self.server.server_address[1]
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join()
+        self.temp.cleanup()
+
+    def request(self, method, path, body=None, headers=None):
+        connection = http.client.HTTPConnection("127.0.0.1", self.port, timeout=3)
+        connection.request(method, path, body=body, headers=headers or {})
+        response = connection.getresponse()
+        data = response.read()
+        connection.close()
+        return response, data
+
+    def test_accepts_png_and_stores_it_under_target(self):
+        body, content_type = multipart("photo.png", PNG)
+        response, data = self.request(
+            "POST", "/logo-contest", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
+        )
+        self.assertEqual(response.status, 201, data)
+        target_dir = self.upload_dir / "logo-contest"
+        files = list(target_dir.iterdir())
+        self.assertEqual(len(files), 1)
+        self.assertEqual(files[0].read_bytes(), PNG)
+        self.assertNotIn("photo", files[0].name)
+        self.assertEqual(files[0].stat().st_mode & 0o777, 0o600)
+        self.assertEqual(target_dir.stat().st_mode & 0o777, 0o700)
+
+        hidden_response, _ = self.request("GET", f"/logo-contest/{files[0].name}")
+        self.assertEqual(hidden_response.status, 404)
+
+    def test_root_is_rejected_but_target_page_is_available(self):
+        root_response, _ = self.request("GET", "/")
+        target_response, page = self.request("GET", "/911u19jasdl")
+        self.assertEqual(root_response.status, 418)
+        self.assertEqual(target_response.status, 200)
+        self.assertIn(b"Private Drop", page)
+
+    def test_upload_to_root_is_rejected_as_teapot(self):
+        body, content_type = multipart("photo.png", PNG)
+        response, _ = self.request(
+            "POST", "/", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
+        )
+        self.assertEqual(response.status, 418)
+
+    def test_targets_are_stored_in_separate_directories(self):
+        for target in ("logo-contest", "911u19jasdl"):
+            body, content_type = multipart("photo.png", PNG)
+            response, _ = self.request(
+                "POST", f"/{target}", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
+            )
+            self.assertEqual(response.status, 201)
+            self.assertEqual(len(list((self.upload_dir / target).iterdir())), 1)
+
+    def test_invalid_or_nested_targets_are_rejected(self):
+        for path in ("/bad.target", "/nested/target", "/%2e%2e", "/healthz"):
+            body, content_type = multipart("photo.png", PNG)
+            response, _ = self.request(
+                "POST", path, body, {"Content-Type": content_type, "Content-Length": str(len(body))}
+            )
+            self.assertEqual(response.status, 404)
+
+    def test_rejects_non_image_content(self):
+        body, content_type = multipart("fake.jpg", b"not really an image")
+        response, _ = self.request(
+            "POST", "/logo-contest", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
+        )
+        self.assertEqual(response.status, 400)
+        self.assertFalse((self.upload_dir / "logo-contest").exists())
+
+    def test_rejects_oversized_request_before_reading_body(self):
+        response, _ = self.request(
+            "POST", "/logo-contest", b"", {"Content-Type": "multipart/form-data; boundary=x", "Content-Length": str(MAX_FILE_BYTES + 1024 * 1024 + 1)}
+        )
+        self.assertEqual(response.status, 413)
+
+    def test_unknown_paths_are_not_served(self):
+        response, _ = self.request("GET", "/server.py")
+        self.assertEqual(response.status, 404)
+
+    def test_detects_supported_signatures(self):
+        self.assertEqual(detect_image_type(b"\xff\xd8\xffrest"), (".jpg", "image/jpeg"))
+        self.assertEqual(detect_image_type(PNG), (".png", "image/png"))
+        self.assertEqual(detect_image_type(b"\x00\x00\x00\x18ftypheic\x00\x00\x00\x00heic"), (".heic", "image/heic"))
+
+
+if __name__ == "__main__":
+    unittest.main()