From a15e4bdf51c250852e976c10787e943c52cd3cb9 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:14:14 -0500 Subject: [PATCH] fix(codegen): emit a real health timestamp and fail-closed scaffolding (#1257) The generated FastAPI project looked deployable and passed a naive smoke test while being non-functional. `/api/health` returned the literal `"2024-01-01T00:00:00Z"`, so the probe could not distinguish a live process from a wedged one or a served cache. Auth, database and message routes returned convincing 200-shaped payloads with no implementation behind them. Rule applied to the template: implement everything the generator can genuinely implement; make everything it cannot fail loudly with 501. A stub that answers successfully teaches operators to trust a lie. - `/api/health` evaluates `datetime.now(timezone.utc)` per request - `/` reports its own gaps via `UNIMPLEMENTED_ENDPOINTS` - `/api/messages` backed by a real (documented non-persistent) store; POST now returns 201 - `POST /auth/login` and `GET /api/data` return 501 with instructions - `create_access_token` / `decode_access_token` are real implementations Also in the same template: - `SECRET_KEY` no longer defaults to `secrets.token_urlsafe(32)`. A per-process random secret invalidates every token on restart and rejects tokens minted by sibling workers, surfacing as intermittent logouts rather than as the misconfiguration it is. It now reads env and fails closed at signing time. - `passlib[bcrypt]` pinned in generated requirements; the template imports `passlib.context`, so auth projects failed at import. - `HTTPException` import made conditional, unused `secrets` dropped, `JWTError` given a real use. Tests execute the generated app with `TestClient` rather than grepping its source, since the defect was behavioural: two health calls must return different, timezone-aware, near-now timestamps. Closes #1257 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/code_generator.py | 177 +++++++++++++++--- tests/unit/test_code_generator.py | 166 ++++++++++++++++ 2 files changed, 318 insertions(+), 25 deletions(-) diff --git a/src/youtube_extension/backend/code_generator.py b/src/youtube_extension/backend/code_generator.py index 501cfb763..614fe8490 100644 --- a/src/youtube_extension/backend/code_generator.py +++ b/src/youtube_extension/backend/code_generator.py @@ -463,7 +463,10 @@ async def _generate_python_api(self, project_path: Path, video_analysis: dict, f if "database" in features: requirements.append("sqlalchemy==2.0.23") if "authentication" in features: + # The generated main.py imports passlib.context; omitting it here + # produced a project that fails at import time. requirements.append("python-jose[cryptography]==3.3.0") + requirements.append("passlib[bcrypt]==1.7.4") # Generate main.py main_py = self._generate_fastapi_main(title, features) @@ -1096,42 +1099,153 @@ def _generate_vanilla_styles_css( # ─── FastAPI generator ────────────────────────────────────────── def _generate_fastapi_main(self, title: str, features: list[str]) -> str: - """Generate main.py for FastAPI projects""" + """Generate main.py for FastAPI projects. + + Design rule for this template: emit only behaviour the generator can + genuinely implement. Anything that depends on knowledge the generator + does not have -- a user store, a database schema -- is emitted as an + endpoint that fails with HTTP 501 rather than one that returns a + convincing 200. A stub that answers successfully makes a non-functional + service pass a smoke test, which is worse than no endpoint at all. + """ auth_imports = "" auth_code = "" + scaffolding_endpoints: list[str] = [] if "authentication" in features: + scaffolding_endpoints.append("POST /auth/login") auth_imports = ''' -import os -import secrets -from datetime import datetime, timedelta from jose import JWTError, jwt from passlib.context import CryptContext''' auth_code = ''' -# Authentication setup -SECRET_KEY = os.getenv("SECRET_KEY", secrets.token_urlsafe(32)) +# ─── Authentication ───────────────────────────────────────────────────────── +# Token minting and verification below are real implementations. The login +# route is not: only you know how your users are stored and verified, so it +# fails with 501 instead of returning a token-shaped response that would let a +# caller believe authentication works. + +SECRET_KEY = os.getenv("SECRET_KEY") ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") + +def _require_secret_key() -> str: + """Return the signing key, or fail closed if it was never configured. + + Deliberately not defaulted to a generated value: a per-process random + secret silently invalidates every token on restart and rejects tokens + minted by sibling workers, which presents as intermittent logouts rather + than as the misconfiguration it is. + """ + if not SECRET_KEY: + raise RuntimeError( + "SECRET_KEY is not set. Refusing to sign or verify tokens without " + "a configured signing key." + ) + return SECRET_KEY + + +def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: + """Mint a signed JWT carrying `data` plus an expiry claim.""" + key = _require_secret_key() + payload = data.copy() + payload["exp"] = datetime.now(timezone.utc) + ( + expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) + ) + return jwt.encode(payload, key, algorithm=ALGORITHM) + + +def decode_access_token(token: str) -> dict: + """Verify a JWT minted by `create_access_token` and return its claims.""" + key = _require_secret_key() + try: + return jwt.decode(token, key, algorithms=[ALGORITHM]) + except JWTError as exc: + raise HTTPException(status_code=401, detail="Invalid or expired token") from exc + + @app.post("/auth/login") async def login(): - """Placeholder login endpoint""" - return {"message": "Login endpoint - implement authentication logic"}''' + """Scaffolding only -- not implemented. + + To implement: look the user up, verify the password with + `pwd_context.verify`, then return `create_access_token({"sub": user_id})`. + """ + raise HTTPException( + status_code=501, + detail=( + "Not implemented. POST /auth/login is generated scaffolding: verify " + "credentials against your user store, then return " + "create_access_token(...)." + ), + )''' database_code = "" if "database" in features: + scaffolding_endpoints.append("GET /api/data") database_code = ''' +# ─── Database ─────────────────────────────────────────────────────────────── +# Not implemented: the generator has no connection string, schema, or model for +# your data. Replace the 501 below once you have wired up a real query. + @app.get("/api/data") async def get_data(): - """Placeholder data endpoint""" - return {"data": "Connect to your database here"}''' + """Scaffolding only -- not implemented.""" + raise HTTPException( + status_code=501, + detail=( + "Not implemented. GET /api/data is generated scaffolding: connect " + "to your database and return real rows." + ), + )''' + + # HTTPException is only referenced by the scaffolding routes; importing + # it unconditionally would leave generated projects with a dead import. + fastapi_import = ( + "from fastapi import FastAPI, HTTPException" + if scaffolding_endpoints + else "from fastapi import FastAPI" + ) + if "authentication" in features: + stdlib_imports = ( + "import os\n" + "from datetime import datetime, timedelta, timezone\n" + "from typing import List, Optional" + ) + else: + stdlib_imports = ( + "from datetime import datetime, timezone\n" + "from typing import List, Optional" + ) + scaffolding_literal = ( + "[\n " + + ",\n ".join(f'"{route}"' for route in scaffolding_endpoints) + + ",\n]" + if scaffolding_endpoints + else "[]" + ) + + return f'''"""{title} - return f'''from fastapi import FastAPI, HTTPException +Generated by UVAI from a YouTube tutorial. + +Routes named in ``UNIMPLEMENTED_ENDPOINTS`` are scaffolding: they respond with +HTTP 501 until you implement them. Everything else in this file is a working +implementation. A clean startup is not evidence that the scaffolded routes +work -- by design, they cannot pass a smoke test while they remain stubs. +""" + +{stdlib_imports} + +{fastapi_import} from fastapi.middleware.cors import CORSMiddleware -from pydantic import BaseModel -from typing import List, Optional{auth_imports} +from pydantic import BaseModel{auth_imports} + +#: Routes that exist but are not implemented. Kept as data so the service can +#: report its own gaps rather than leaving callers to discover them at runtime. +UNIMPLEMENTED_ENDPOINTS: List[str] = {scaffolding_literal} app = FastAPI( title="{title}", @@ -1157,32 +1271,45 @@ class MessageResponse(BaseModel): message: str status: str +#: Process-local message store. This is a real implementation, not a stub, but +#: it is deliberately not persistent: it empties on restart and is not shared +#: between workers. Swap it for your datastore before relying on it. +_MESSAGES: List[MessageResponse] = [] + @app.get("/") async def root(): - """Root endpoint""" + """Service index, including the routes that are not implemented yet.""" return {{ "message": "Welcome to {title}", "description": "This API was generated by UVAI from a YouTube tutorial", - "endpoints": ["/docs", "/api/messages", "/api/health"] + "endpoints": ["/docs", "/api/health", "/api/messages"], + "unimplemented_endpoints": UNIMPLEMENTED_ENDPOINTS }} @app.get("/api/health") async def health_check(): - """Health check endpoint""" - return {{"status": "healthy", "timestamp": "2024-01-01T00:00:00Z"}} + """Liveness probe. + + The timestamp is evaluated per request. A constant here would make the + probe indistinguishable from a served cache or a wedged process, which + defeats the only thing a health route exists to support. + """ + return {{ + "status": "healthy", + "timestamp": datetime.now(timezone.utc).isoformat() + }} @app.get("/api/messages", response_model=List[MessageResponse]) async def get_messages(): - """Get all messages""" - return [ - {{"message": "Hello from your UVAI generated API!", "status": "active"}}, - {{"message": "This API was created from a YouTube tutorial", "status": "active"}} - ] + """Return every message currently held in the in-memory store.""" + return _MESSAGES -@app.post("/api/messages", response_model=MessageResponse) +@app.post("/api/messages", response_model=MessageResponse, status_code=201) async def create_message(message: Message): - """Create a new message""" - return {{"message": f"Received: {{message.text}}", "status": "created"}} + """Append a message to the in-memory store and echo the stored record.""" + record = MessageResponse(message=message.text, status="created") + _MESSAGES.append(record) + return record {auth_code} {database_code} diff --git a/tests/unit/test_code_generator.py b/tests/unit/test_code_generator.py index d5e2e9ee8..aaf4456ab 100644 --- a/tests/unit/test_code_generator.py +++ b/tests/unit/test_code_generator.py @@ -3,9 +3,13 @@ from __future__ import annotations import asyncio +import importlib.util import json +import sys import tempfile import threading +import time +from datetime import datetime, timezone from pathlib import Path import pytest @@ -1217,3 +1221,165 @@ async def test_successful_generation_keeps_its_directory( assert Path(result["project_path"]).is_dir() assert len(list(tmp_path.glob("uvai_project_*"))) == 1 + + +def _load_generated_app(source: str, tmp_path: Path, module_name: str): + """Import a generated ``main.py`` and hand back its module. + + The generated file is executed rather than merely inspected: the defect in + #1257 was behavioural (a constant health timestamp, endpoints that answered + 200 without doing anything), and only running the app can prove it is gone. + """ + main_py = tmp_path / "main.py" + main_py.write_text(source, encoding="utf-8") + + spec = importlib.util.spec_from_file_location(module_name, main_py) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + try: + spec.loader.exec_module(module) + except Exception: + sys.modules.pop(module_name, None) + raise + return module + + +@pytest.fixture +def generated_app(tmp_path, request): + """Build, import and serve a generated FastAPI project.""" + created: list[str] = [] + + def _factory(features: list[str]): + generator = ProjectCodeGenerator(use_ai_generation=False) + source = generator._generate_fastapi_main("Generated API", features) + module_name = ( + f"uvai_generated_main_{abs(hash((request.node.nodeid, tuple(features))))}" + ) + target = tmp_path / module_name + target.mkdir() + module = _load_generated_app(source, target, module_name) + created.append(module_name) + return source, module + + yield _factory + + for name in created: + sys.modules.pop(name, None) + + +class TestGeneratedFastAPIBehaviour: + """Execute the generated FastAPI app instead of trusting its shape. + + Regression cover for #1257: the template shipped a hardcoded health + timestamp and placeholder endpoints that returned 200, so a generated + project passed a naive smoke test while being non-functional. + """ + + def test_health_timestamp_is_evaluated_per_request(self, generated_app): + from fastapi.testclient import TestClient + + source, module = generated_app([]) + assert "2024-01-01T00:00:00Z" not in source + + with TestClient(module.app) as client: + first = client.get("/api/health") + time.sleep(0.01) + second = client.get("/api/health") + + assert first.status_code == 200 + assert second.status_code == 200 + assert first.json()["status"] == "healthy" + + first_ts = first.json()["timestamp"] + second_ts = second.json()["timestamp"] + assert first_ts != second_ts, "health timestamp is constant across calls" + + parsed = datetime.fromisoformat(first_ts) + assert parsed.tzinfo is not None, "timestamp must be timezone-aware" + assert abs((datetime.now(timezone.utc) - parsed).total_seconds()) < 60 + + def test_unimplemented_endpoints_fail_loudly(self, generated_app): + from fastapi.testclient import TestClient + + _, module = generated_app(["authentication", "database"]) + + with TestClient(module.app) as client: + assert client.post("/auth/login").status_code == 501 + assert client.get("/api/data").status_code == 501 + index = client.get("/").json() + + assert index["unimplemented_endpoints"] == ["POST /auth/login", "GET /api/data"] + + def test_index_reports_no_gaps_without_optional_features(self, generated_app): + from fastapi.testclient import TestClient + + _, module = generated_app([]) + + with TestClient(module.app) as client: + index = client.get("/").json() + + assert index["unimplemented_endpoints"] == [] + assert "/api/health" in index["endpoints"] + + def test_messages_round_trip_through_a_real_store(self, generated_app): + from fastapi.testclient import TestClient + + _, module = generated_app([]) + + with TestClient(module.app) as client: + assert client.get("/api/messages").json() == [] + + created = client.post("/api/messages", json={"text": "hello"}) + assert created.status_code == 201 + assert created.json() == {"message": "hello", "status": "created"} + + assert client.get("/api/messages").json() == [ + {"message": "hello", "status": "created"} + ] + + def test_token_helpers_are_real_implementations(self, generated_app, monkeypatch): + _, module = generated_app(["authentication"]) + monkeypatch.setattr(module, "SECRET_KEY", "unit-test-secret") + + token = module.create_access_token({"sub": "user-1"}) + assert module.decode_access_token(token)["sub"] == "user-1" + + with pytest.raises(Exception) as excinfo: + module.decode_access_token("not-a-jwt") + assert getattr(excinfo.value, "status_code", None) == 401 + + def test_signing_fails_closed_without_a_secret_key( + self, generated_app, monkeypatch + ): + """A random per-process default would silently break tokens on restart.""" + source, module = generated_app(["authentication"]) + assert "secrets.token_urlsafe" not in source + + monkeypatch.setattr(module, "SECRET_KEY", None) + with pytest.raises(RuntimeError, match="SECRET_KEY"): + module.create_access_token({"sub": "user-1"}) + + def test_auth_projects_declare_their_password_hashing_dependency( + self, monkeypatch, tmp_path + ): + """The generated main.py imports passlib, so it must be pinned.""" + project_dir = tmp_path / "api_auth_reqs" + project_dir.mkdir() + monkeypatch.setattr(tempfile, "mkdtemp", lambda prefix: str(project_dir)) + + gen = ProjectCodeGenerator(use_ai_generation=False) + video_analysis = { + "extracted_info": { + "title": "Auth API", + "technologies": ["python"], + "features": ["authentication"], + "project_type": "api", + }, + "success": True, + } + asyncio.run(gen.generate_project(video_analysis, {"type": "api"})) + + requirements = (project_dir / "requirements.txt").read_text() + assert "passlib" in requirements + assert "python-jose" in requirements