"""Real application HTTP checks for installation, cookies, and backup controls. All persistence and artwork paths are isolated in temporary directories; workers, logging file handlers, and the metrics listener are disabled for these tests. """ import io from pathlib import Path import tempfile import unittest from unittest.mock import patch from fastapi.testclient import TestClient from backend.app import db, main from backend.app.config import settings from backend.app.services import backups OPERATOR_TOKEN = "installation-http-operator-token-test-123456789" OWNER_PASSWORD = "installation-http-owner-password-123456789" BACKUP_PASSPHRASE = "installation-http-backup-passphrase-123456789" class InstallationHttpTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) for key, value in { "sqlite_path": str(self.root / "magent.db"), "sqlite_journal_mode": "DELETE", "jwt_secret": "installation-http-test-jwt-secret-1234567890", "settings_encryption_key": None, "admin_username": "unused-environment-admin", "admin_password": "", "setup_token": OPERATOR_TOKEN, "auth_cookie_secure": True, "auth_cookie_domain": None, "auth_cookie_samesite": "strict", }.items(): context = patch.object(settings, key, value) context.start() self.addCleanup(context.stop) for context in ( patch.object(main, "configure_logging"), patch.object(main, "start_metrics"), patch.object(main, "_background_tasks", []), patch.object(main, "_background_started", False), patch.object(backups, "_assets_root", return_value=self.root / "assets"), patch.dict("os.environ", {"BACKGROUND_TASKS_ENABLED": "false"}), ): context.start() self.addCleanup(context.stop) self.origin = str(settings.cors_allow_origin).rstrip("/") self.client = self.enterContext(TestClient(main.app, base_url="https://magent.test")) self.client.headers["Origin"] = self.origin def create_owner(self): response = self.client.post("/setup/bootstrap", json={ "setup_token": OPERATOR_TOKEN, "username": "owner", "password": OWNER_PASSWORD, }) self.assertEqual(response.status_code, 201, response.text) return response def sign_in(self): response = self.client.post("/auth/login", data={"username": "owner", "password": OWNER_PASSWORD}) self.assertEqual(response.status_code, 200, response.text) self.assertIn(settings.auth_cookie_name, self.client.cookies) auth_cookie = next(value for value in response.headers.get_list("set-cookie") if value.startswith(settings.auth_cookie_name + "=")) self.assertIn("HttpOnly", auth_cookie) self.assertIn("Secure", auth_cookie) self.assertIn("SameSite=strict", auth_cookie) self.assertNotIn("Authorization", self.client.headers) def test_fresh_setup_cookie_settings_completion_and_backup_round_trip(self): status = self.client.get("/setup/status") self.assertEqual(status.json(), {"setup_required": True, "needs_admin": True}) self.assertEqual(status.headers["cache-control"], "no-store") self.assertIn("default-src 'none'", status.headers["content-security-policy"]) self.assertEqual(self.client.get("/setup/state").status_code, 401) self.assertEqual(self.client.get("/admin/backups").status_code, 401) self.create_owner() self.sign_in() self.assertEqual(self.client.get("/setup/state").json()["step"], "apps") response = self.client.put("/admin/settings", json={ "jellyfin_base_url": "http://jellyfin.test:8096", "jellyfin_api_key": "test-integration-key-for-setup", "site_login_message": "Welcome to this installation", }) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["updated"], 3) values = {row["key"]: row for row in self.client.get("/admin/settings").json()["settings"]} self.assertEqual(values["jellyfin_base_url"]["value"], "http://jellyfin.test:8096") self.assertIsNone(values["jellyfin_api_key"]["value"]) self.assertTrue(values["jellyfin_api_key"]["isSet"]) response = self.client.put("/setup/state", json={"step": "review"}) self.assertEqual(response.status_code, 200, response.text) completed = self.client.post("/setup/complete") self.assertEqual(completed.status_code, 200, completed.text) self.assertTrue(completed.json()["completed"]) self.assertEqual(self.client.get("/setup/status").json(), {"setup_required": False, "needs_admin": False}) self.assertEqual(main._background_tasks, []) exported = self.client.post("/admin/backups/export", json={ "passphrase": BACKUP_PASSPHRASE, "include_cache": False, }) self.assertEqual(exported.status_code, 200, exported.text[:100]) self.assertTrue(exported.content.startswith(backups.MAGIC)) self.assertEqual(exported.headers["cache-control"], "no-store") self.assertNotIn(b"test-integration-key-for-setup", exported.content) restored = self.client.post("/admin/backups/restore", files={ "file": ("restore.magent-backup", io.BytesIO(exported.content), "application/octet-stream"), }, data={"passphrase": BACKUP_PASSPHRASE, "confirmation": "RESTORE"}) self.assertEqual(restored.status_code, 202, restored.text) self.assertTrue(restored.json()["restart_required"]) self.assertEqual(db.get_setting("site_login_message"), "Welcome to this installation") self.assertIsNotNone(self.client.get("/admin/backups").json()["pending_restore"]) cancelled = self.client.delete("/admin/backups/restore") self.assertEqual(cancelled.status_code, 200, cancelled.text) self.assertIsNone(self.client.get("/admin/backups").json()["pending_restore"]) def test_cross_origin_bootstrap_and_authenticated_changes_are_rejected(self): response = self.client.post("/setup/bootstrap", headers={"Origin": "https://unrelated.invalid"}, json={ "setup_token": OPERATOR_TOKEN, "username": "owner", "password": OWNER_PASSWORD, }) self.assertEqual(response.status_code, 403) self.assertFalse(db.has_admin_user()) self.create_owner() self.sign_in() response = self.client.put("/setup/state", headers={"Origin": "https://unrelated.invalid"}, json={"step": "review"}) self.assertEqual(response.status_code, 403) response = self.client.post("/admin/backups/export", headers={"Origin": "https://unrelated.invalid"}, json={"passphrase": BACKUP_PASSPHRASE}) self.assertEqual(response.status_code, 403) self.assertEqual(self.client.get("/setup/state").json()["step"], "apps") def test_setup_validation_errors_do_not_echo_password_or_token(self): secret_password = "private-password-marker-" + "p" * 1024 secret_token = "private-token-marker-" + "t" * 1024 for payload, secret in ( ({"setup_token": OPERATOR_TOKEN, "username": "owner", "password": secret_password}, secret_password), ({"setup_token": secret_token, "username": "owner", "password": OWNER_PASSWORD}, secret_token), ({"setup_token": OPERATOR_TOKEN, "password": OWNER_PASSWORD}, OWNER_PASSWORD), ): with self.subTest(secret=secret[:22]): response = self.client.post("/setup/bootstrap", json=payload) self.assertEqual(response.status_code, 422, response.text) self.assertNotIn(secret, response.text) self.assertNotIn(OPERATOR_TOKEN, response.text) for error in response.json()["detail"]: self.assertNotIn("input", error) def test_backup_validation_errors_do_not_echo_passphrases(self): self.create_owner() self.sign_in() passphrase = "private-backup-passphrase-marker-" + "p" * 1024 response = self.client.post("/admin/backups/export", json={"passphrase": passphrase}) self.assertEqual(response.status_code, 422) self.assertNotIn(passphrase, response.text) response = self.client.post("/admin/backups/restore", files={"file": ("archive", b"data")}, data={ "passphrase": passphrase, "confirmation": "RESTORE", }) self.assertEqual(response.status_code, 422) self.assertNotIn(passphrase, response.text) self.assertIsNone(self.client.get("/admin/backups").json()["pending_restore"]) def test_real_middleware_rejects_oversized_bootstrap_before_creation(self): response = self.client.post("/setup/bootstrap", content=b"x" * (17 * 1024), headers={"Content-Type": "application/json"}) self.assertEqual(response.status_code, 413, response.text) self.assertFalse(db.has_admin_user()) if __name__ == "__main__": unittest.main()