feat: scaffold Flask + React app with schema and JSON error handling
Claude-Session: https://claude.ai/code/session_01M9MLit5Ko3X4s7rzC5Kv7X
This commit is contained in:
@@ -0,0 +1,26 @@
|
||||
"""Point the app at a dedicated test database before any test module imports it."""
|
||||
import os
|
||||
|
||||
import psycopg2
|
||||
from psycopg2 import sql
|
||||
|
||||
ADMIN_URL = os.environ.get("TEST_ADMIN_DATABASE_URL", "postgresql://postgres:[email protected]:5432/postgres")
|
||||
TEST_DATABASE = "books_test"
|
||||
|
||||
|
||||
def _ensure_test_database() -> None:
|
||||
conn = psycopg2.connect(ADMIN_URL)
|
||||
conn.autocommit = True # CREATE DATABASE cannot run inside a transaction
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (TEST_DATABASE,))
|
||||
if cur.fetchone() is None:
|
||||
cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(TEST_DATABASE)))
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
_ensure_test_database()
|
||||
# Assigned, never defaulted: tests TRUNCATE tables, so they must not inherit a real DATABASE_URL.
|
||||
os.environ["DATABASE_URL"] = ADMIN_URL.rsplit("/", 1)[0] + "/" + TEST_DATABASE
|
||||
os.environ["PASSWORD_PEPPER"] = "test-pepper-0123456789abcdef0123456789abcdef"
|
||||
@@ -0,0 +1,58 @@
|
||||
import os
|
||||
import unittest
|
||||
|
||||
import psycopg2
|
||||
|
||||
from app import app
|
||||
|
||||
HTTPS = "https://localhost" # the session cookie is Secure
|
||||
|
||||
|
||||
def db_execute(sql: str, params: tuple = ()) -> list[tuple]:
|
||||
conn = psycopg2.connect(os.environ["DATABASE_URL"])
|
||||
try:
|
||||
with conn, conn.cursor() as cur:
|
||||
cur.execute(sql, params)
|
||||
return cur.fetchall() if cur.description else []
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
class ApiTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
if not os.environ["DATABASE_URL"].endswith("/books_test"):
|
||||
raise RuntimeError("Refusing to truncate a database that is not books_test")
|
||||
db_execute("TRUNCATE users, sessions, books, notes RESTART IDENTITY CASCADE")
|
||||
self.client = app.test_client()
|
||||
|
||||
def call(self, method, path, json=None, client=None):
|
||||
kwargs = {"method": method, "base_url": HTTPS}
|
||||
if json is not None:
|
||||
kwargs["json"] = json
|
||||
elif method != "GET":
|
||||
kwargs["content_type"] = "application/json"
|
||||
return (client or self.client).open(path, **kwargs)
|
||||
|
||||
def register(self, username="alice", password="correct horse battery", client=None):
|
||||
response = self.call("POST", "/api/auth/register", {"username": username, "password": password}, client)
|
||||
self.assertEqual(response.status_code, 201, response.get_json())
|
||||
return response
|
||||
|
||||
def other_user(self, username="bob"):
|
||||
client = app.test_client()
|
||||
self.register(username, client=client)
|
||||
return client
|
||||
|
||||
def genre_id(self, name: str) -> int:
|
||||
return db_execute("SELECT id FROM genres WHERE name = %s", (name,))[0][0]
|
||||
|
||||
def create_book(self, client=None, **overrides) -> dict:
|
||||
payload = {
|
||||
"title": "Dune",
|
||||
"author": "Frank Herbert",
|
||||
"genre_id": self.genre_id("Science Fiction"),
|
||||
"total_pages": 412,
|
||||
} | overrides
|
||||
response = self.call("POST", "/api/books", payload, client)
|
||||
self.assertEqual(response.status_code, 201, response.get_json())
|
||||
return response.get_json()
|
||||
@@ -0,0 +1,32 @@
|
||||
from pathlib import Path
|
||||
|
||||
from tests.support import HTTPS, ApiTestCase, db_execute
|
||||
|
||||
SCHEMA = Path(__file__).parent.parent / "schema.sql"
|
||||
|
||||
|
||||
class AppTests(ApiTestCase):
|
||||
def test_health(self):
|
||||
response = self.call("GET", "/api/health")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.get_json(), {"status": "ok"})
|
||||
|
||||
def test_genres_seeded_and_reseeding_is_idempotent(self):
|
||||
before = db_execute("SELECT last_value FROM genres_id_seq")[0][0]
|
||||
db_execute(SCHEMA.read_text())
|
||||
db_execute(SCHEMA.read_text())
|
||||
self.assertEqual(db_execute("SELECT count(*) FROM genres")[0][0], 16)
|
||||
self.assertEqual(db_execute("SELECT last_value FROM genres_id_seq")[0][0], before)
|
||||
|
||||
def test_unknown_route_returns_json_404(self):
|
||||
response = self.call("GET", "/api/nope")
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertIn("error", response.get_json())
|
||||
|
||||
def test_mutation_without_json_content_type_is_rejected(self):
|
||||
response = self.client.post("/api/health", data="x", content_type="text/plain", base_url=HTTPS)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertEqual(
|
||||
response.get_json(),
|
||||
{"error": "Request body must be JSON with Content-Type: application/json"},
|
||||
)
|
||||
Reference in New Issue
Block a user