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