tests/test_runtime_consistency.py

Rama master · réplica de lectura del repositorio original.

"""事务、图片复验和 HTTP 入口的行为验证。"""
import sqlite3
import tempfile
import unittest
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

from peach.redirect import create_redirect_app
from peach.routes_media import _image_response
from peach.web_state import WebContract
from peach.config import PeachSettings
from peach.health import database_status, readiness
from support.ledger import fresh_ledger


class RuntimeConsistencyTests(unittest.TestCase):
    def test_readiness_distinguishes_missing_database_and_pending_schema(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory).resolve()
            path = root / "ledger.db"
            settings = PeachSettings(db_path=path, configured=True)
            self.assertFalse(readiness(settings)["ready"])
            self.assertFalse(path.exists())
            self.assertEqual(database_status(path), "missing")
            path.touch()
            self.assertEqual(database_status(path), "empty")
            fresh_ledger(root)
            self.assertEqual(database_status(path), "available")
            self.assertTrue(readiness(settings)["ready"])
            connection = sqlite3.connect(path)
            try:
                connection.execute("DELETE FROM schema_migration WHERE version=(SELECT max(version) FROM schema_migration)")
                connection.commit()
            finally:
                connection.close()
            result = readiness(settings)
            self.assertTrue(result["checks"]["database"])
            self.assertFalse(result["checks"]["schema"])

    def test_commit_invalidates_values_read_during_transaction(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory).resolve() / "test.db"
            with sqlite3.connect(path) as connection:
                connection.execute("CREATE TABLE sample(value INTEGER)")
                connection.execute("INSERT INTO sample VALUES(0)")
            connection.close()
            contract = WebContract(path)

            def read():
                with contract.read_connection() as connection:
                    return connection.execute("SELECT value FROM sample").fetchone()[0]

            with contract.write_transaction() as connection:
                contract.cache_bust()
                connection.execute("UPDATE sample SET value=1")
                self.assertEqual(contract.cached("sample", read), 0)
            self.assertEqual(contract.cached("sample", read), 1)
            generation = contract.cache_generation
            with self.assertRaises(ValueError):
                with contract.write_transaction() as connection:
                    connection.execute("UPDATE sample SET value=2")
                    raise ValueError("rollback")
            self.assertEqual(contract.cache_generation, generation)
            self.assertEqual(read(), 1)

    def test_aggregate_cache_is_bounded_and_keeps_recent_hits(self):
        contract = WebContract(Path("unused.db"))
        for key in range(192):
            contract.cached(str(key), lambda key=key: key)
        contract.cached("0", lambda: -1)
        contract.cached("new", lambda: 999)
        self.assertEqual(len(contract.cache), 192)
        self.assertIn("0", contract.cache)
        self.assertNotIn("1", contract.cache)

    def test_http_navigation_uses_fixed_origin_and_does_not_replay_writes(self):
        with TestClient(create_redirect_app("https://peach.test")) as client:
            health = client.get("/healthz")
            self.assertEqual(health.json()["service"], "peach-redirect")
            response = client.get("/items?q=a&q=b&t=secret", headers={"Host": "evil.test"},
                                  follow_redirects=False)
            self.assertEqual(response.headers["location"], "https://peach.test/items?q=a&q=b")
            self.assertEqual(client.post("/api/action", json={"value": 1}).status_code, 426)

    def test_mutable_image_revalidates_after_replacement(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory).resolve() / "cover.jpg"
            path.write_bytes(b"first")
            app = FastAPI()

            @app.get("/image")
            def image(request: Request):
                return _image_response(request, path)

            with TestClient(app) as client:
                response = client.get("/image")
                headers = {"If-None-Match": response.headers["etag"]}
                self.assertEqual(client.get("/image", headers=headers).status_code, 304)
                path.write_bytes(b"replacement")
                updated = client.get("/image", headers=headers)
                self.assertEqual(updated.status_code, 200)
                self.assertEqual(updated.content, b"replacement")
                self.assertEqual(updated.headers["cache-control"], "private, no-cache")