diff --git a/deliverables/README.md b/deliverables/README.md index 791d27b..7ec538a 100644 --- a/deliverables/README.md +++ b/deliverables/README.md @@ -15,6 +15,7 @@ | `notebooklm-access-suite/` | ชุดทักษะเข้าถึง NotebookLM (link share, artifact normalization) | | `ai-gateway-architecture-review/` | [AI Gateway Architecture Review — resilience & cost control](./ai-gateway-architecture-review/README.md) — Risk Register 32 จุดอ่อน, สถาปัตยกรรมที่ปรับปรุง (M1–M21), rollout 6 ระยะ | | `notebooklm-link-share/` | ทักษะแยก share ลิงก์ NotebookLM | +| `pm-backend/` | [Program Management Backend](./pm-backend/README.md) — FastAPI app 4 modules: PM CSV template, provider-neutral billing (stub/Stripe/Chargebee/Paddle), tool switcher, opt-in encryption at rest | ## ไฟล์อ้างอิงที่เกี่ยวข้อง diff --git a/deliverables/pm-backend/.gitignore b/deliverables/pm-backend/.gitignore new file mode 100644 index 0000000..407571c --- /dev/null +++ b/deliverables/pm-backend/.gitignore @@ -0,0 +1,11 @@ +.venv/ +__pycache__/ +*.pyc +.env +.pytest_cache/ + +# Generated at runtime — not part of the deliverable +pm_backend.db +pm_backend.db-wal +pm_backend.db-shm +tool_state.json diff --git a/deliverables/pm-backend/README.md b/deliverables/pm-backend/README.md new file mode 100644 index 0000000..623e2fc --- /dev/null +++ b/deliverables/pm-backend/README.md @@ -0,0 +1,233 @@ +# Program Management Backend (pm_backend) + +A single integrated FastAPI application with four equal modules: + +| Module | Purpose | Route prefix | +|--------|---------|--------------| +| **Program Management CSV** | Generate / validate program-management CSV templates | `/api/csv` | +| **Billing interoperability** | Provider-neutral billing (stub, Stripe, Chargebee, Paddle) | `/api/billing` | +| **Tool switcher** | Switch the active tool at runtime | `/api/tool` | +| **Encryption at rest** | Opt-in Fernet column encryption | `/api/programs` | + +--- + +## Quick start + +```bash +python -m venv .venv && source .venv/bin/activate # macOS/Linux +pip install -r requirements.txt + +uvicorn app.main:app --reload +``` + +Then open the interactive docs at . + +Run the tests: + +```bash +pytest -q # 58 unit tests; 15 sandbox integration tests skip by default +``` + +### Integration tests (opt-in) + +`tests/test_integration_*.py` exercise the real provider **sandbox** APIs +(Stripe test mode, Paddle sandbox, Chargebee test site). They are skipped +unless you opt in with `--run-integration` and supply credentials: + +```bash +# Stripe test mode +export STRIPE_TEST_API_KEY=sk_test_xxxxxxxxxxxx +# Paddle sandbox +export PADDLE_SANDBOX_API_KEY=pdl_sdbx_xxxxxxxxxxxx +export PADDLE_SANDBOX_BASE_URL=https://sandbox-api.paddle.com +# Chargebee test site +export CHARGEBEE_TEST_API_KEY=cb_xxxxxxxxxxxx +export CHARGEBEE_TEST_SITE=your-site-test + +pytest --run-integration -q +``` + +Safety guards in `conftest.py` refuse to run against production credentials — +a live Stripe key (`sk_live_*`), a non-sandbox Paddle host, or a Chargebee site +without a `-test` suffix will fail fast rather than touch a real account. +Guard behaviour is itself unit-tested in `tests/test_integration_guards.py` +(always runs, no network). Customers created by integration runs are deleted on +teardown. + +--- + +## Configuration + +All settings come from environment variables (an optional `.env` file in the +project root is also loaded). Defaults are shown. + +| Variable | Default | Description | +|----------|---------|-------------| +| `DATABASE_URL` | `sqlite:///./pm_backend.db` | SQLAlchemy database URL | +| `ENCRYPT_AT_REST` | `false` | Enable at-rest encryption of PII fields | +| `ENCRYPTION_KEY` | *(derived)* | Explicit Fernet key (see below) | +| `ENCRYPTION_SECRET` | `change-me` | Secret used to derive a key when `ENCRYPTION_KEY` unset | +| `BILLING_PROVIDER` | `stub` | `stub`, `stripe`, `chargebee`, or `paddle` | +| `BILLING_API_KEY` | *(empty)* | Provider API key (required for real providers) | +| `BILLING_BASE_URL` | *(empty)* | Override the provider API base (testing) | +| `ACTIVE_TOOL` | `pm_csv` | Initial tool for the tool switcher (`pm_csv`/`billing`/`settings`) | + +--- + +## Modules + +### 1. Program Management CSV + +Canonical 12-column sheet: + +``` +program_id, name, owner, status, start_date, end_date, +budget, spent, risk, health, milestone, notes +``` + +- `GET /api/csv/template` — download a blank CSV template. +- `POST /api/csv/generate` — generate CSV from a JSON list of rows. +- `POST /api/csv/validate` — validate rows against the schema and report + per-row errors (`status` and `health` are enumerated). + +### 2. Billing interoperability + +The app talks to a single **provider-neutral interface** +(`BillingProvider`): `create_customer`, `get_customer`, `create_invoice`, +`charge`, `health`. Only the adapter for the configured `BILLING_PROVIDER` +is loaded, so nothing else changes when you swap providers. + +Provider failures surface through a small error hierarchy that the HTTP layer +maps to status codes: + +| Error | HTTP | +|-------|------| +| `ProviderAuthError` | `401` — bad / missing key | +| `ProviderNotFoundError` | `404` — resource missing at the provider | +| `ProviderError` | `502` — upstream failure | + +Endpoints: + +- `GET /api/billing/provider` — provider name, status, mode. +- `POST /api/billing/customers` — create a customer. +- `POST /api/billing/invoices` — create + finalize an invoice for a customer. +- `POST /api/billing/charges` — create a PaymentIntent for a customer. + +Amounts are in **minor units** (e.g. `2500` = $25.00). + +#### Stripe + +Set the provider and a Stripe API key: + +```bash +export BILLING_PROVIDER=stripe +export BILLING_API_KEY=sk_test_xxxxxxxxxxxx # or sk_live_... +uvicorn app.main:app --reload +``` + +- Test vs live mode is **auto-detected** from the key prefix and reported in + `/health` and `/api/billing/provider`. +- The Stripe SDK is imported lazily and only required when this provider is + selected; `stripe>=8.0` is declared in `requirements.txt`. +- The adapter creates a draft invoice, adds one line item, and finalizes it — + mirroring a discrete charge in Stripe's data model. + +No live key? The default `stub` provider runs offline with no network or +credentials and is a reference for writing other adapters. Add a new provider +by implementing `BillingProvider` and registering it in `build_provider()`. + +#### Chargebee + +```bash +export BILLING_PROVIDER=chargebee +export BILLING_API_KEY=cb_xxxxxxxxxxx +export BILLING_BASE_URL=https://.chargebee.com/api/v2 +uvicorn app.main:app --reload +``` + +Chargebee authenticates against your **site subdomain**; both the API key and +the full `https://.chargebee.com/api/v2` base URL are required. Test vs +live mode is inferred from the site (`-test` suffix → `test`). The adapter +calls Chargebee's REST API over `requests` (lazy-imported) and sends amounts +in Chargebee's major units. + +#### Paddle + +```bash +export BILLING_PROVIDER=paddle +export BILLING_API_KEY=pdl_xxxxxxxxxxx +# optional — defaults to live; use the sandbox for testing: +export BILLING_BASE_URL=https://sandbox-api.paddle.com +uvicorn app.main:app --reload +``` + +Paddle Billing authenticates with a bearer token and sends amounts in **minor +units as strings** (`"2500"`). It is transaction-led: `create_invoice` creates +a manually-collected transaction (an invoice to send) and `charge` creates an +automatically-collected one (an immediate charge). Test vs live mode is +inferred from the base URL (`sandbox-api` → `test`). The adapter calls Paddle's +REST API over `requests` (lazy-imported). + +### 3. Tool switcher + +- `GET /api/tool` — current active tool + available tools. +- `POST /api/tool` `{"tool": "billing"}` — switch the active tool. + +The selection is persisted to `tool_state.json` and survives restarts. + +### 4. Encryption at rest (opt-in) + +PII-bearing string columns (currently the program `note` field) are stored +**plain-text by default**. Enable transparent Fernet encryption without +changing application code: + +```bash +export ENCRYPT_AT_REST=true +export ENCRYPTION_SECRET='a-long-random-secret' +``` + +When enabled, values are encrypted before they reach the database and +decrypted on read via a SQLAlchemy `TypeDecorator`. When disabled (default), +the `cryptography` package is never imported — no hard dependency. + +Generate and pin an explicit key instead of deriving one: + +```bash +python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +export ENCRYPTION_KEY='' +``` + +> **Rotating keys or changing the flag after data exists requires a +> migration** — rows written under one mode are not re-encrypted +> automatically. Back up before toggling on a live database. + +`/health` reports `encrypt_at_rest: true|false` so clients can verify. + +--- + +## Project layout + +``` +pm_backend/ +├── app/ +│ ├── main.py # FastAPI app + all routers +│ ├── config.py # env-driven settings +│ ├── db.py # engine, session, opt-in EncryptedString +│ ├── billing.py # provider-neutral interface + stub & Stripe adapters +│ ├── chargebee.py # Chargebee billing adapter (same interface) +│ ├── paddle.py # Paddle Billing adapter (same interface) +│ ├── pm_csv.py # program-management CSV template logic +│ ├── tool_switcher.py # active-tool state +│ └── repository.py # program record CRUD +├── tests/ +│ ├── test_app.py # API + opt-in encryption subprocess check +│ ├── test_stripe_adapter.py # Stripe adapter vs mocked SDK +│ ├── test_chargebee_adapter.py # Chargebee adapter vs mocked HTTP +│ ├── test_paddle_adapter.py # Paddle adapter vs mocked HTTP +│ ├── test_integration_guards.py # live-credential safety guards (no network) +│ ├── test_integration_stripe.py # real Stripe test-mode API (opt-in) +│ ├── test_integration_paddle.py # real Paddle sandbox API (opt-in) +│ └── test_integration_chargebee.py # real Chargebee test site (opt-in) +├── conftest.py # sys.path + --run-integration harness +└── requirements.txt +``` diff --git a/deliverables/pm-backend/app/__init__.py b/deliverables/pm-backend/app/__init__.py new file mode 100644 index 0000000..596fc7a --- /dev/null +++ b/deliverables/pm-backend/app/__init__.py @@ -0,0 +1,11 @@ +"""Program Management Backend. + +A single integrated FastAPI application containing four equal modules: + +1. Program Management CSV template generation (``pm_csv``) +2. Billing interoperability via a provider-neutral interface (``billing``) +3. A runtime tool switcher (``tool_switcher``) +4. Opt-in database encryption at rest (``db``), gated by ``ENCRYPT_AT_REST`` +""" + +__version__ = "1.0.0" diff --git a/deliverables/pm-backend/app/billing.py b/deliverables/pm-backend/app/billing.py new file mode 100644 index 0000000..1063eb6 --- /dev/null +++ b/deliverables/pm-backend/app/billing.py @@ -0,0 +1,281 @@ +"""Billing interoperability. + +A provider-neutral billing interface so the backend can talk to any billing +provider (Stripe, Chargebee, a custom gateway, ...) without the rest of the +app knowing which one is behind it. Only the adapter for the *selected* +provider (``BILLING_PROVIDER``) is loaded; the default ``stub`` adapter +requires no network and no credentials, so the app runs out of the box. + +Adapters signal failures through the provider-error hierarchy below so the +HTTP layer can map them to status codes without knowing provider internals: + ProviderAuthError -> 401/400 (bad key) + ProviderNotFoundError-> 404 (resource missing at the provider) + ProviderError -> 502 (upstream failure) +""" +from __future__ import annotations + +from typing import Protocol + +from pydantic import BaseModel, Field + + +class ProviderError(Exception): + """Base class for billing provider failures surfaced to the API layer.""" + + +class ProviderNotFoundError(ProviderError): + """A requested resource does not exist at the provider. -> HTTP 404.""" + + +class ProviderAuthError(ProviderError): + """The provider rejected the configured API key/credentials.""" + + +class Customer(BaseModel): + id: str + email: str = "" + name: str = "" + currency: str = "usd" + + +class Invoice(BaseModel): + id: str + customer_id: str + amount: int = Field(..., description="Amount in minor units") + currency: str = "usd" + status: str = "draft" + description: str = "" + + +class PaymentIntent(BaseModel): + id: str + customer_id: str + amount: int + currency: str = "usd" + status: str = "requires_confirmation" + + +class BillingProvider(Protocol): + """The interface every billing adapter implements.""" + + name: str + + def create_customer(self, email: str, name: str = "") -> Customer: ... + def get_customer(self, customer_id: str) -> Customer: ... + def create_invoice( + self, customer_id: str, amount: int, currency: str = "usd", + description: str = "", + ) -> Invoice: ... + def charge( + self, customer_id: str, amount: int, currency: str = "usd", + ) -> PaymentIntent: ... + def health(self) -> dict: ... + + +class StubBillingProvider: + """Offline adapter used by default. Useful for dev/tests and as the + reference implementation for writing real provider adapters.""" + + name = "stub" + + def __init__(self, base_url: str = "", api_key: str = "") -> None: + self.base_url = base_url + self.api_key = api_key + self._customers: dict[str, Customer] = {} + self._invoices: dict[str, Invoice] = {} + self._intents: dict[str, PaymentIntent] = {} + self._seq = 0 + + def _next_id(self, prefix: str) -> str: + self._seq += 1 + return f"{prefix}_{self._seq:06d}" + + def create_customer(self, email: str, name: str = "") -> Customer: + c = Customer(id=self._next_id("cus"), email=email, name=name) + self._customers[c.id] = c + return c + + def get_customer(self, customer_id: str) -> Customer: + if customer_id not in self._customers: + raise ProviderNotFoundError(f"customer not found: {customer_id}") + return self._customers[customer_id] + + def create_invoice( + self, customer_id: str, amount: int, currency: str = "usd", + description: str = "", + ) -> Invoice: + if customer_id not in self._customers: + raise ProviderNotFoundError(f"customer not found: {customer_id}") + inv = Invoice( + id=self._next_id("in_"), + customer_id=customer_id, + amount=amount, + currency=currency, + status="open", + description=description, + ) + self._invoices[inv.id] = inv + return inv + + def charge( + self, customer_id: str, amount: int, currency: str = "usd", + ) -> PaymentIntent: + if customer_id not in self._customers: + raise ProviderNotFoundError(f"customer not found: {customer_id}") + pi = PaymentIntent( + id=self._next_id("pi_"), + customer_id=customer_id, + amount=amount, + currency=currency, + status="succeeded", + ) + self._intents[pi.id] = pi + return pi + + def health(self) -> dict: + return { + "provider": self.name, + "status": "ok", + "mode": "offline", + "detail": "stub adapter (offline, no credentials required)", + } + + +class StripeBillingProvider: + """Real adapter backed by the stripe SDK. + + Only instantiated when BILLING_PROVIDER=stripe; the stripe package is + imported lazily so the app has no hard dependency on it unless the + provider is actually enabled. + + Raises ``ProviderAuthError`` on bad credentials and + ``ProviderNotFoundError`` when Stripe reports a missing resource, keeping + the HTTP layer provider-agnostic. + """ + + name = "stripe" + + def __init__(self, api_key: str, base_url: str = "") -> None: + if not api_key: + raise ProviderAuthError( + "BILLING_API_KEY is required for provider=stripe" + ) + import stripe # lazy import + + stripe.api_key = api_key + if base_url: + stripe.api_base = base_url + self._stripe = stripe + # "test" for sk_test_... keys, "live" for sk_live_... keys. + self._mode = "test" if api_key.startswith(("sk_test_", "rk_test_")) else "live" + + @staticmethod + def _translate(exc: Exception) -> ProviderError: + """Map a stripe library exception to a provider-error type.""" + # Imported lazily so this only runs on the Stripe path. + import stripe as _s + + if isinstance(exc, _s.error.AuthenticationError): + return ProviderAuthError(f"Stripe authentication failed: {exc}") + if isinstance(exc, _s.error.InvalidRequestError): + # InvalidRequestError is Stripe's signal for a missing resource + # (e.g. customer_cus_X does not exist) or a bad parameter. + return ProviderNotFoundError(f"Stripe request failed: {exc}") + return ProviderError(f"Stripe error: {exc}") + + def _customer(self, obj) -> Customer: + return Customer( + id=obj["id"], + email=obj.get("email", "") or "", + name=obj.get("name", "") or "", + ) + + def create_customer(self, email: str, name: str = "") -> Customer: + try: + params: dict = {"email": email} + if name: + params["name"] = name + return self._customer(self._stripe.Customer.create(**params)) + except Exception as exc: # stripe.error.* + raise self._translate(exc) from exc + + def get_customer(self, customer_id: str) -> Customer: + try: + return self._customer(self._stripe.Customer.retrieve(customer_id)) + except Exception as exc: + raise self._translate(exc) from exc + + def create_invoice( + self, customer_id: str, amount: int, currency: str = "usd", + description: str = "", + ) -> Invoice: + try: + # Draft invoice -> one line item -> finalize. This mirrors how a + # single, discrete charge is represented in real Stripe billing. + draft = self._stripe.Invoice.create(customer=customer_id) + self._stripe.InvoiceItem.create( + customer=customer_id, + amount=amount, + currency=currency, + description=description or "charge", + invoice=draft["id"], + ) + finalized = self._stripe.Invoice.finalize_invoice(draft["id"]) + return Invoice( + id=finalized["id"], + customer_id=customer_id, + amount=amount, + currency=currency, + status=finalized["status"], + description=description, + ) + except Exception as exc: + raise self._translate(exc) from exc + + def charge( + self, customer_id: str, amount: int, currency: str = "usd", + ) -> PaymentIntent: + try: + pi = self._stripe.PaymentIntent.create( + amount=amount, + currency=currency, + customer=customer_id, + automatic_payment_methods={"enabled": True}, + ) + return PaymentIntent( + id=pi["id"], + customer_id=customer_id, + amount=amount, + currency=currency, + status=pi["status"], + ) + except Exception as exc: + raise self._translate(exc) from exc + + def health(self) -> dict: + return { + "provider": self.name, + "status": "ok", + "mode": self._mode, + } + + +def build_provider(provider: str, api_key: str, base_url: str) -> BillingProvider: + """Factory that returns the adapter for the configured provider.""" + provider = (provider or "stub").lower() + if provider == "stub": + return StubBillingProvider(base_url=base_url, api_key=api_key) + if provider == "stripe": + return StripeBillingProvider(api_key=api_key, base_url=base_url) + if provider == "chargebee": + from .chargebee import ChargebeeBillingProvider + + return ChargebeeBillingProvider(api_key=api_key, base_url=base_url) + if provider == "paddle": + from .paddle import PaddleBillingProvider + + return PaddleBillingProvider(api_key=api_key, base_url=base_url) + raise ValueError( + f"Unsupported BILLING_PROVIDER: {provider!r} " + "(supported: stub, stripe, chargebee, paddle)" + ) diff --git a/deliverables/pm-backend/app/chargebee.py b/deliverables/pm-backend/app/chargebee.py new file mode 100644 index 0000000..a2d84b1 --- /dev/null +++ b/deliverables/pm-backend/app/chargebee.py @@ -0,0 +1,151 @@ +"""Chargebee billing adapter. + +A second real provider behind the same ``BillingProvider`` interface as the +Stripe adapter. Chargebee uses REST over the ``chargebee`` Python client; +here we call it via ``requests`` against the REST API so the adapter has no +extra hard dependency beyond the already-optional provider key. + +Base URL form: https://.chargebee.com/api/v2 +Auth: Basic with an empty password (Chargebee convention) + +Selected by setting BILLING_PROVIDER=chargebee. +""" +from __future__ import annotations + +import base64 + +from .billing import ( + Customer, + Invoice, + PaymentIntent, + ProviderAuthError, + ProviderError, + ProviderNotFoundError, +) + + +class ChargebeeBillingProvider: + """Chargebee adapter implementing the provider-neutral billing interface.""" + + name = "chargebee" + + def __init__(self, api_key: str, base_url: str = "") -> None: + if not api_key: + raise ProviderAuthError( + "BILLING_API_KEY is required for provider=chargebee" + ) + if not base_url: + raise ProviderAuthError( + "BILLING_BASE_URL (your Chargebee site, " + "https://.chargebee.com/api/v2) is required " + "for provider=chargebee" + ) + self._api_key = api_key + self._base_url = base_url.rstrip("/") + # test vs live: Chargebee sites end in "-test" or "-live" + self._mode = "test" if "-test" in base_url else "live" + + def _headers(self) -> dict: + token = base64.b64encode(f"{self._api_key}:".encode()).decode() + return {"Authorization": f"Basic {token}"} + + def _request(self, method: str, path: str, **kwargs) -> dict: + import requests # lazy import so the test layer can inject a fake + + try: + resp = requests.request( + method, + f"{self._base_url}{path}", + headers=self._headers(), + timeout=30, + **kwargs, + ) + except requests.RequestException as exc: + raise ProviderError(f"Chargebee request failed: {exc}") from exc + + if resp.status_code == 401: + raise ProviderAuthError("Chargebee authentication failed") + if resp.status_code == 404: + raise ProviderNotFoundError(f"Chargebee resource not found: {path}") + if not resp.ok: + raise ProviderError( + f"Chargebee error {resp.status_code}: {resp.text[:200]}" + ) + return resp.json() + + def create_customer(self, email: str, name: str = "") -> Customer: + payload = {"customer": {"email": email}} + if name: + payload["customer"]["first_name"] = name.split()[0] + if len(name.split()) > 1: + payload["customer"]["last_name"] = " ".join(name.split()[1:]) + data = self._request("POST", "/customers", json=payload) + c = data["customer"] + return Customer( + id=c["id"], + email=c.get("email", ""), + name=f"{c.get('first_name','')} {c.get('last_name','')}".strip(), + ) + + def get_customer(self, customer_id: str) -> Customer: + data = self._request("GET", f"/customers/{customer_id}") + c = data["customer"] + return Customer( + id=c["id"], + email=c.get("email", ""), + name=f"{c.get('first_name','')} {c.get('last_name','')}".strip(), + ) + + def create_invoice( + self, customer_id: str, amount: int, currency: str = "usd", + description: str = "", + ) -> Invoice: + # Chargebee bills via a subscription/charge; the simplest standing + # representation is an immediate one-off charge against the customer. + payload = { + "invoice": { + "customer_id": customer_id, + "currency_code": currency.upper(), + "addons": [ + { + "amount": amount / 100, # Chargebee uses major units + "description": description or "charge", + } + ], + } + } + data = self._request("POST", "/invoices", json=payload) + inv = data["invoice"] + return Invoice( + id=inv["id"], + customer_id=customer_id, + amount=amount, + currency=currency, + status=inv.get("status", "draft"), + description=description, + ) + + def charge( + self, customer_id: str, amount: int, currency: str = "usd", + ) -> PaymentIntent: + # Chargebee is subscription-led; a PaymentIntent equivalent is the + # "payment intent" created against the customer's payment method. + payload = { + "payment_intent": { + "customer_id": customer_id, + "amount": amount / 100, + "currency_code": currency.upper(), + } + } + data = self._request("POST", "/payment_intents", json=payload) + pi = data["payment_intent"] + return PaymentIntent( + id=pi["id"], + customer_id=customer_id, + amount=amount, + currency=currency, + status=pi.get("status", "inited"), + ) + + def health(self) -> dict: + return {"provider": self.name, "status": "ok", "mode": self._mode} diff --git a/deliverables/pm-backend/app/config.py b/deliverables/pm-backend/app/config.py new file mode 100644 index 0000000..666fc2f --- /dev/null +++ b/deliverables/pm-backend/app/config.py @@ -0,0 +1,90 @@ +"""Application configuration. + +Read from environment variables so the same code runs in dev, staging, and +production without edits. Supports an optional ``.env`` file (no third-party +dependency required). +""" +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from functools import lru_cache +from pathlib import Path + + +def _bool_env(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class Settings: + app_name: str = "Program Management Backend" + version: str = "1.0.0" + debug: bool = field(default_factory=lambda: _bool_env("DEBUG", False)) + + # --- Database -------------------------------------------------------- + database_url: str = field( + default_factory=lambda: os.getenv( + "DATABASE_URL", "sqlite:///./pm_backend.db" + ) + ) + # Opt-in encryption at rest. The app never hard-depends on an encryption + # library: when false, string fields are stored as plain text and the + # cryptography package is never imported. + encrypt_at_rest: bool = field( + default_factory=lambda: _bool_env("ENCRYPT_AT_REST", False) + ) + # Fernet key when encryption is enabled. Generate one with: + # python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" + # If unset, the app derives a stable key from ENCRYPTION_SECRET so it + # still boots out of the box. + encryption_key: str = field( + default_factory=lambda: os.getenv("ENCRYPTION_KEY", "") + ) + encryption_secret: str = field( + default_factory=lambda: os.getenv("ENCRYPTION_SECRET", "change-me") + ) + + # --- Billing --------------------------------------------------------- + # Provider-neutral interoperability. Only the selected provider's adapter + # is required at runtime; the rest of the app talks to one interface. + billing_provider: str = field( + default_factory=lambda: os.getenv("BILLING_PROVIDER", "stub").lower() + ) + billing_api_key: str = field( + default_factory=lambda: os.getenv("BILLING_API_KEY", "") + ) + billing_base_url: str = field( + default_factory=lambda: os.getenv("BILLING_BASE_URL", "") + ) + + # --- Tool switcher --------------------------------------------------- + active_tool: str = field( + default_factory=lambda: os.getenv("ACTIVE_TOOL", "pm_csv").lower() + ) + tools: tuple[str, ...] = ("pm_csv", "billing", "settings") + + +def load_dotenv(path: Path | None = None) -> None: + """Minimal .env loader.""" + env_file = path or Path(__file__).resolve().parent.parent / ".env" + if not env_file.exists(): + return + for line in env_file.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + key = key.strip() + value = value.strip().strip('"').strip("'") + if key and key not in os.environ: + os.environ[key] = value + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + load_dotenv() + return Settings() diff --git a/deliverables/pm-backend/app/db.py b/deliverables/pm-backend/app/db.py new file mode 100644 index 0000000..532bf36 --- /dev/null +++ b/deliverables/pm-backend/app/db.py @@ -0,0 +1,110 @@ +"""Database layer with OPT-IN encryption at rest. + +The app does NOT hard-depend on an encryption library. When +``ENCRYPT_AT_REST=true`` we use a SQLAlchemy TypeDecorator to transparently +encrypt string fields with Fernet before they hit the database and decrypt +them on read. When it is false (the default), the same fields are stored as +plain text and the ``cryptography`` package is never imported. +""" +from __future__ import annotations + +import base64 +import hashlib +import os +from typing import Any + +from sqlalchemy import Column, String, create_engine, event +from sqlalchemy.orm import declarative_base, sessionmaker +from sqlalchemy.types import TypeDecorator + +from .config import get_settings + + +def _build_fernet(secret: str, key: str) -> Any: + """Construct a Fernet instance from an explicit key or a derived secret.""" + from cryptography.fernet import Fernet # lazy import + + if key: + return Fernet(key.encode("utf-8")) + digest = hashlib.sha256(secret.encode("utf-8")).digest() + return Fernet(base64.urlsafe_b64encode(digest)) + + +class EncryptedString(TypeDecorator): + """Transparent at-rest encryption for string columns (opt-in).""" + + impl = String + cache_ok = True + + def __init__(self, *args, fernet=None, **kwargs): + super().__init__(*args, **kwargs) + self.fernet = fernet + + def process_bind_param(self, value, dialect): + if value is None or self.fernet is None: + return value + return self.fernet.encrypt(value.encode("utf-8")).decode("utf-8") + + def process_result_value(self, value, dialect): + if value is None or self.fernet is None: + return value + return self.fernet.decrypt(value.encode("utf-8")).decode("utf-8") + + +def _string_column(fernet: Any | None) -> Column: + """Plain or encrypted string column depending on the setting.""" + if fernet is not None: + return Column(EncryptedString(fernet=fernet)) + return Column(String) + + +settings = get_settings() + +_fernet: Any | None = None +if settings.encrypt_at_rest: + _fernet = _build_fernet(settings.encryption_secret, settings.encryption_key) + +engine = create_engine( + settings.database_url, + connect_args=( + {"check_same_thread": False} + if settings.database_url.startswith("sqlite") + else {} + ), +) + +if settings.database_url.startswith("sqlite"): + + @event.listens_for(engine, "connect") + def _set_sqlite_pragma(dbapi_connection, connection_record): + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA journal_mode=WAL") + cursor.close() + + +SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False) +Base = declarative_base() + + +class ProgramRecord(Base): + """A row in the program registry. ``note`` is the encrypted field.""" + + __tablename__ = "programs" + + id = Column(String, primary_key=True, default=lambda: os.urandom(8).hex()) + name = Column(String, nullable=False) + owner = Column(String, default="") + status = Column(String, default="planned") + note = _string_column(_fernet) # encrypted only when ENCRYPT_AT_REST=true + + +def init_db() -> None: + Base.metadata.create_all(bind=engine) + + +def get_db(): + db = SessionLocal() + try: + yield db + finally: + db.close() diff --git a/deliverables/pm-backend/app/main.py b/deliverables/pm-backend/app/main.py new file mode 100644 index 0000000..fadd139 --- /dev/null +++ b/deliverables/pm-backend/app/main.py @@ -0,0 +1,257 @@ +"""FastAPI application entrypoint. + +Bundles all four equal modules behind one ASGI app: + /api/csv - Program Management CSV template generation + /api/billing - provider-neutral billing interoperability + /api/tool - runtime tool switcher + /api/programs - program registry (exercises opt-in encryption at rest) +""" +from __future__ import annotations + +import io + +from fastapi import Depends, FastAPI, HTTPException, Query +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from . import billing, pm_csv, tool_switcher +from .billing import ( + ProviderAuthError, + ProviderError, + ProviderNotFoundError, +) +from .config import Settings, get_settings +from .db import get_db, init_db +from .repository import create_program, get_program, list_programs + +app = FastAPI( + title="Program Management Backend", + version=get_settings().version, + description=( + "Integrated backend with Program Management CSV generation, " + "billing interoperability, a tool switcher, and opt-in " + "database encryption at rest (ENCRYPT_AT_REST)." + ), +) + + +@app.on_event("startup") +def _startup() -> None: + init_db() + + +@app.get("/health", tags=["system"]) +def health(settings: Settings = Depends(get_settings)) -> dict: + return { + "status": "ok", + "app": settings.app_name, + "version": settings.version, + "encrypt_at_rest": settings.encrypt_at_rest, + "billing_provider": settings.billing_provider, + "active_tool": tool_switcher.current_tool(), + } + + +# -------------------------------------------------------------------------- +# Tool switcher +# -------------------------------------------------------------------------- + +class ToolSwitchIn(BaseModel): + tool: str = Field(..., description="One of pm_csv, billing, settings") + + +@app.get("/api/tool", tags=["tool-switcher"]) +def get_active_tool() -> dict: + return {"active_tool": tool_switcher.current_tool(), "tools": tool_switcher.list_tools()} + + +@app.post("/api/tool", tags=["tool-switcher"]) +def switch_tool(body: ToolSwitchIn) -> dict: + try: + active = tool_switcher.set_tool(body.tool) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return {"active_tool": active, "tools": tool_switcher.list_tools()} + + +# -------------------------------------------------------------------------- +# Program Management CSV +# -------------------------------------------------------------------------- + +class ProgramRowsIn(BaseModel): + rows: list[dict] = Field(default_factory=list) + + +@app.get("/api/csv/template", tags=["pm-csv"]) +def csv_template() -> StreamingResponse: + content = pm_csv.template_csv() + return StreamingResponse( + io.StringIO(content), + media_type="text/csv", + headers={ + "Content-Disposition": 'attachment; filename="program_template.csv"' + }, + ) + + +@app.post("/api/csv/generate", tags=["pm-csv"]) +def csv_generate(body: ProgramRowsIn) -> dict: + try: + rows = pm_csv.rows_to_csv(body.rows) + except Exception as exc: # pydantic validation + raise HTTPException(status_code=400, detail=str(exc)) from exc + return {"csv": rows, "count": len(body.rows)} + + +@app.post("/api/csv/validate", tags=["pm-csv"]) +def csv_validate(body: ProgramRowsIn) -> dict: + """Validate rows against the PM CSV schema and return errors per row.""" + errors: list[dict] = [] + for idx, raw in enumerate(body.rows): + try: + pm_csv.ProgramRow(**raw) + except Exception as exc: + errors.append({"row": idx, "error": str(exc)}) + return {"valid": not errors, "errors": errors, "count": len(body.rows)} + + +# -------------------------------------------------------------------------- +# Billing interoperability +# -------------------------------------------------------------------------- + +_provider_cache: dict[tuple, billing.BillingProvider] = {} + + +def _provider(settings: Settings = Depends(get_settings)) -> billing.BillingProvider: + """Return a cached billing provider so stub state persists between calls.""" + key = (settings.billing_provider, settings.billing_api_key, settings.billing_base_url) + if key not in _provider_cache: + try: + _provider_cache[key] = billing.build_provider( + provider=settings.billing_provider, + api_key=settings.billing_api_key, + base_url=settings.billing_base_url, + ) + except ProviderAuthError as exc: + raise HTTPException(status_code=401, detail=str(exc)) from exc + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return _provider_cache[key] + + +class CustomerCreateIn(BaseModel): + email: str = Field(..., min_length=3) + name: str = "" + + +class InvoiceCreateIn(BaseModel): + customer_id: str = Field(...) + amount: int = Field(..., gt=0, description="Amount in minor units (cents)") + currency: str = "usd" + description: str = "" + + +class ChargeIn(BaseModel): + customer_id: str = Field(...) + amount: int = Field(..., gt=0) + currency: str = "usd" + + +@app.get("/api/billing/provider", tags=["billing"]) +def billing_provider_info( + provider: billing.BillingProvider = Depends(_provider), +) -> dict: + return provider.health() + + +@app.post("/api/billing/customers", tags=["billing"]) +def billing_create_customer( + body: CustomerCreateIn, + provider: billing.BillingProvider = Depends(_provider), +) -> dict: + try: + c = provider.create_customer(body.email, body.name) + except ProviderNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except ProviderError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + return c.model_dump() + + +@app.post("/api/billing/invoices", tags=["billing"]) +def billing_create_invoice( + body: InvoiceCreateIn, + provider: billing.BillingProvider = Depends(_provider), +) -> dict: + try: + inv = provider.create_invoice( + body.customer_id, body.amount, body.currency, body.description + ) + except ProviderNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except ProviderError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + return inv.model_dump() + + +@app.post("/api/billing/charges", tags=["billing"]) +def billing_charge( + body: ChargeIn, + provider: billing.BillingProvider = Depends(_provider), +) -> dict: + try: + pi = provider.charge(body.customer_id, body.amount, body.currency) + except ProviderNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except ProviderError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + return pi.model_dump() + + +# -------------------------------------------------------------------------- +# Program registry (encryption at rest on the note column) +# -------------------------------------------------------------------------- + +class ProgramCreateIn(BaseModel): + name: str = Field(..., min_length=1) + owner: str = "" + status: str = "planned" + note: str = "" + + +class ProgramOut(BaseModel): + id: str + name: str + owner: str = "" + status: str = "" + note: str = "" + + +@app.post("/api/programs", tags=["programs"]) +def api_create_program( + body: ProgramCreateIn, db: Session = Depends(get_db) +) -> ProgramOut: + rec = create_program( + db, name=body.name, owner=body.owner, status=body.status, note=body.note + ) + return ProgramOut(id=rec.id, name=rec.name, owner=rec.owner, status=rec.status, note=rec.note) + + +@app.get("/api/programs", tags=["programs"]) +def api_list_programs( + db: Session = Depends(get_db), + limit: int = Query(100, ge=1, le=1000), +) -> list[ProgramOut]: + return [ + ProgramOut(id=r.id, name=r.name, owner=r.owner, status=r.status, note=r.note) + for r in list_programs(db)[:limit] + ] + + +@app.get("/api/programs/{program_id}", tags=["programs"]) +def api_get_program(program_id: str, db: Session = Depends(get_db)) -> ProgramOut: + rec = get_program(db, program_id) + if rec is None: + raise HTTPException(status_code=404, detail="program not found") + return ProgramOut(id=rec.id, name=rec.name, owner=rec.owner, status=rec.status, note=rec.note) diff --git a/deliverables/pm-backend/app/paddle.py b/deliverables/pm-backend/app/paddle.py new file mode 100644 index 0000000..16c1b08 --- /dev/null +++ b/deliverables/pm-backend/app/paddle.py @@ -0,0 +1,154 @@ +"""Paddle billing adapter. + +A third real provider behind the same ``BillingProvider`` interface as the +Stripe and Chargebee adapters. Talks to the Paddle Billing API (v2) over +``requests`` (lazy-imported) so it adds no hard dependency unless selected. + +Base URL: https://api.paddle.com (live) / https://sandbox-api.paddle.com (sandbox) +Auth: Authorization: Bearer +Amounts: Paddle Billing expects minor units as STRINGS (e.g. "2500"). + +Paddle is transaction-led: a *transaction* is the billable record, and an +invoice is generated from it. The adapter maps to the neutral interface as: + create_invoice -> a manually-collected transaction (an invoice to send) + charge -> an automatically-collected transaction (an immediate charge) + +Selected by setting BILLING_PROVIDER=paddle. +""" +from __future__ import annotations + +from .billing import ( + Customer, + Invoice, + PaymentIntent, + ProviderAuthError, + ProviderError, + ProviderNotFoundError, +) + +_DEFAULT_BASE = "https://api.paddle.com" +_SANDBOX_BASE = "https://sandbox-api.paddle.com" + + +class PaddleBillingProvider: + """Paddle Billing adapter implementing the neutral billing interface.""" + + name = "paddle" + + def __init__(self, api_key: str, base_url: str = "") -> None: + if not api_key: + raise ProviderAuthError( + "BILLING_API_KEY is required for provider=paddle" + ) + self._api_key = api_key + self._base_url = (base_url or _DEFAULT_BASE).rstrip("/") + # Sandbox endpoints signal test mode. + self._mode = "test" if "sandbox" in self._base_url else "live" + + def _headers(self) -> dict: + return { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + } + + def _request(self, method: str, path: str, **kwargs) -> dict: + import requests # lazy import so the test layer can inject a fake + + try: + resp = requests.request( + method, + f"{self._base_url}{path}", + headers=self._headers(), + timeout=30, + **kwargs, + ) + except requests.RequestException as exc: + raise ProviderError(f"Paddle request failed: {exc}") from exc + + if resp.status_code == 401: + raise ProviderAuthError("Paddle authentication failed") + if resp.status_code == 404: + raise ProviderNotFoundError(f"Paddle resource not found: {path}") + if not resp.ok: + raise ProviderError( + f"Paddle error {resp.status_code}: {resp.text[:200]}" + ) + + body = resp.json() + # Paddle wraps every payload in {"data": ...}. + return body.get("data", body) + + def _customer(self, c: dict) -> Customer: + return Customer( + id=c.get("id", ""), + email=c.get("email", "") or "", + name=c.get("name", "") or "", + ) + + def create_customer(self, email: str, name: str = "") -> Customer: + payload: dict = {"email": email} + if name: + payload["name"] = name + return self._customer(self._request("POST", "/customers", json=payload)) + + def get_customer(self, customer_id: str) -> Customer: + return self._customer(self._request("GET", f"/customers/{customer_id}")) + + def _transaction( + self, customer_id: str, amount: int, currency: str, + description: str, collection_mode: str, + ) -> dict: + payload = { + "customer_id": customer_id, + "collection_mode": collection_mode, + "items": [ + { + "quantity": 1, + "price": { + "description": description or "charge", + "unit_price": { + "amount": str(amount), # minor units, as a string + "currency_code": currency.upper(), + }, + "product": { + "name": description or "charge", + "tax_category": "standard", + }, + }, + } + ], + } + return self._request("POST", "/transactions", json=payload) + + def create_invoice( + self, customer_id: str, amount: int, currency: str = "usd", + description: str = "", + ) -> Invoice: + txn = self._transaction( + customer_id, amount, currency, description, "manual" + ) + return Invoice( + id=txn.get("id", ""), + customer_id=customer_id, + amount=amount, + currency=currency, + status=txn.get("status", "draft"), + description=description, + ) + + def charge( + self, customer_id: str, amount: int, currency: str = "usd", + ) -> PaymentIntent: + txn = self._transaction( + customer_id, amount, currency, "charge", "automatic" + ) + return PaymentIntent( + id=txn.get("id", ""), + customer_id=customer_id, + amount=amount, + currency=currency, + status=txn.get("status", "ready"), + ) + + def health(self) -> dict: + return {"provider": self.name, "status": "ok", "mode": self._mode} diff --git a/deliverables/pm-backend/app/pm_csv.py b/deliverables/pm-backend/app/pm_csv.py new file mode 100644 index 0000000..9e83d3b --- /dev/null +++ b/deliverables/pm-backend/app/pm_csv.py @@ -0,0 +1,126 @@ +"""Program Management CSV template generation. + +Provides a canonical CSV structure for managing programs (portfolios of +projects) and tools to (a) render a blank template, (b) fill a template from +a list of program rows, and (c) parse uploaded CSV into structured rows. + +Column model (a complete program-management sheet): + program_id, name, owner, status, start_date, end_date, + budget, spent, risk, health, milestone, notes +""" +from __future__ import annotations + +import csv +import io +from typing import Any, Sequence + +from pydantic import BaseModel, Field, field_validator + +CSV_COLUMNS = [ + "program_id", + "name", + "owner", + "status", + "start_date", + "end_date", + "budget", + "spent", + "risk", + "health", + "milestone", + "notes", +] + +VALID_STATUS = {"planned", "active", "on_hold", "completed", "cancelled"} +VALID_HEALTH = {"green", "amber", "red"} + + +class ProgramRow(BaseModel): + """A single row in the program management sheet.""" + + program_id: str = Field(default="", description="Unique program id") + name: str = Field(..., min_length=1, description="Program name") + owner: str = Field(default="", description="Owning individual/team") + status: str = Field(default="planned") + start_date: str = Field(default="") + end_date: str = Field(default="") + budget: float | None = Field(default=None, ge=0) + spent: float | None = Field(default=None, ge=0) + risk: str = Field(default="") + health: str = Field(default="green") + milestone: str = Field(default="") + notes: str = Field(default="") + + @field_validator("status") + @classmethod + def _status(cls, v: str) -> str: + v = v.strip().lower() + if v and v not in VALID_STATUS: + raise ValueError(f"status must be one of {sorted(VALID_STATUS)}") + return v + + @field_validator("health") + @classmethod + def _health(cls, v: str) -> str: + v = v.strip().lower() + if v and v not in VALID_HEALTH: + raise ValueError(f"health must be one of {sorted(VALID_HEALTH)}") + return v + + +def template_csv() -> str: + """Return a blank, ready-to-fill CSV template (headers + one example row + commented out is not possible in CSV, so we include one fully-formed + example row the caller can delete).""" + return rows_to_csv([ProgramRow(name="Example Program", status="active")]) + + +def rows_to_csv(rows: Sequence[ProgramRow | dict[str, Any]]) -> str: + """Serialize program rows into CSV text (header row included).""" + out = io.StringIO() + writer = csv.writer(out, lineterminator="\n") + writer.writerow(CSV_COLUMNS) + for row in rows: + p = row if isinstance(row, ProgramRow) else ProgramRow(**row) + writer.writerow( + [ + p.program_id, + p.name, + p.owner, + p.status, + p.start_date, + p.end_date, + "" if p.budget is None else f"{p.budget:g}", + "" if p.spent is None else f"{p.spent:g}", + p.risk, + p.health, + p.milestone, + p.notes, + ] + ) + return out.getvalue() + + +def parse_csv(text: str) -> list[ProgramRow]: + """Parse CSV text into validated ProgramRow objects.""" + reader = csv.DictReader(io.StringIO(text)) + if not reader.fieldnames: + raise ValueError("CSV is empty or has no header row") + missing = [c for c in CSV_COLUMNS if c not in reader.fieldnames] + if missing: + raise ValueError(f"Missing required columns: {', '.join(missing)}") + + rows: list[ProgramRow] = [] + for lineno, raw in enumerate(reader, start=2): + if not any((raw.get(c) or "").strip() for c in CSV_COLUMNS): + continue # skip fully-blank lines + # CSV cells for optional numeric fields arrive as "" — map to None. + clean = dict(raw) + for num_field in ("budget", "spent"): + if (clean.get(num_field) or "").strip() == "": + clean[num_field] = None + try: + rows.append(ProgramRow(**clean)) + except Exception as exc: # pydantic ValidationError + raise ValueError(f"Row {lineno}: {exc}") from exc + return rows diff --git a/deliverables/pm-backend/app/repository.py b/deliverables/pm-backend/app/repository.py new file mode 100644 index 0000000..2017833 --- /dev/null +++ b/deliverables/pm-backend/app/repository.py @@ -0,0 +1,34 @@ +"""Program repository: read/write ProgramRecord rows. + +These helpers exist so the API can create and fetch programs, transparently +exercising the optional at-rest encryption on the ``note`` column. The layer +is identical whether or not ENCRYPT_AT_REST is enabled — the difference lives +entirely in the SQLAlchemy column type selected in ``db``. +""" +from __future__ import annotations + +from sqlalchemy.orm import Session + +from .db import ProgramRecord + + +def create_program( + db: Session, + name: str, + owner: str = "", + status: str = "planned", + note: str = "", +) -> ProgramRecord: + record = ProgramRecord(name=name, owner=owner, status=status, note=note) + db.add(record) + db.commit() + db.refresh(record) + return record + + +def get_program(db: Session, program_id: str) -> ProgramRecord | None: + return db.get(ProgramRecord, program_id) + + +def list_programs(db: Session) -> list[ProgramRecord]: + return db.query(ProgramRecord).order_by(ProgramRecord.name).all() diff --git a/deliverables/pm-backend/app/tool_switcher.py b/deliverables/pm-backend/app/tool_switcher.py new file mode 100644 index 0000000..4a6e864 --- /dev/null +++ b/deliverables/pm-backend/app/tool_switcher.py @@ -0,0 +1,59 @@ +"""Runtime tool switcher. + +Lets a caller switch which "tool" the backend presents as active (pm_csv, +billing, settings) at runtime — useful for a UI tool switcher or for routing +an integration to a different feature area without redeploying. The chosen +tool is persisted so a restart keeps the selection. +""" +from __future__ import annotations + +import json +import os +from pathlib import Path + +from .config import get_settings + +_STATE_FILE = Path( + os.getenv("TOOL_STATE_FILE", str(Path(__file__).resolve().parent.parent / "tool_state.json")) +) + +KNOWN_TOOLS = ("pm_csv", "billing", "settings") + + +def _load_state() -> str: + try: + with open(_STATE_FILE, "r", encoding="utf-8") as fh: + data = json.load(fh) + return str(data.get("active_tool", "")).lower() + except (OSError, ValueError, TypeError): + return "" + + +def _save_state(tool: str) -> None: + _STATE_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(_STATE_FILE, "w", encoding="utf-8") as fh: + json.dump({"active_tool": tool}, fh, indent=2) + + +def current_tool() -> str: + """Return the active tool, preferring persisted state over env default.""" + persisted = _load_state() + if persisted in KNOWN_TOOLS: + return persisted + default = get_settings().active_tool.lower() + return default if default in KNOWN_TOOLS else KNOWN_TOOLS[0] + + +def list_tools() -> list[str]: + return list(KNOWN_TOOLS) + + +def set_tool(tool: str) -> str: + """Switch the active tool. Raises ValueError for unknown tools.""" + tool = tool.strip().lower() + if tool not in KNOWN_TOOLS: + raise ValueError( + f"Unknown tool {tool!r}; expected one of {KNOWN_TOOLS}" + ) + _save_state(tool) + return tool diff --git a/deliverables/pm-backend/conftest.py b/deliverables/pm-backend/conftest.py new file mode 100644 index 0000000..914ca01 --- /dev/null +++ b/deliverables/pm-backend/conftest.py @@ -0,0 +1,143 @@ +"""Shared pytest configuration. + +Adds the project root to ``sys.path`` so ``from app import ...`` resolves, and +wires the opt-in harness for integration tests that hit real provider +sandboxes (Stripe / Paddle / Chargebee). + +Integration tests are SKIPPED by default and only run with ``--run-integration`` +AND the relevant credentials in the environment. Guards below refuse to run +against production keys/sites, so a misconfigured environment fails loudly +instead of touching a live account. +""" +import os +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + + +# -------------------------------------------------------------------------- +# Opt-in harness +# -------------------------------------------------------------------------- + +def pytest_addoption(parser): + parser.addoption( + "--run-integration", + action="store_true", + default=False, + help="run integration tests that call real provider sandbox APIs", + ) + + +def pytest_configure(config): + config.addinivalue_line( + "markers", + "integration: hits a real provider sandbox API " + "(opt-in via --run-integration)", + ) + + +def pytest_collection_modifyitems(config, items): + if config.getoption("--run-integration"): + return + skip = pytest.mark.skip( + reason="integration test — pass --run-integration (and set credentials) to run" + ) + for item in items: + if "integration" in item.keywords: + item.add_marker(skip) + + +# -------------------------------------------------------------------------- +# Safety guards (pure functions so they can be unit-tested) +# -------------------------------------------------------------------------- + +class LiveCredentialError(ValueError): + """Raised when an integration test is pointed at production credentials.""" + + +def guard_stripe_key(key: str) -> str: + """Allow only Stripe TEST-mode secret/restricted keys.""" + key = (key or "").strip() + if key.startswith(("sk_live_", "rk_live_")): + raise LiveCredentialError( + "refusing to run integration tests with a LIVE Stripe key" + ) + if not key.startswith(("sk_test_", "rk_test_")): + raise LiveCredentialError( + "Stripe integration tests require a test-mode key " + "(sk_test_... or rk_test_...)" + ) + return key + + +def guard_paddle_sandbox(key: str, base_url: str) -> tuple[str, str]: + """Force the Paddle sandbox host and reject obviously-live keys.""" + key = (key or "").strip() + base_url = (base_url or "").strip() + if "live" in key.lower(): + raise LiveCredentialError( + "refusing to run integration tests with a LIVE Paddle key" + ) + if base_url != "https://sandbox-api.paddle.com": + raise LiveCredentialError( + "Paddle integration tests must target " + "https://sandbox-api.paddle.com" + ) + return key, base_url + + +def guard_chargebee_site(key: str, site: str) -> tuple[str, str]: + """Require a Chargebee ``*-test`` site.""" + key = (key or "").strip() + site = (site or "").strip() + if not site.endswith("-test"): + raise LiveCredentialError( + "refusing to run integration tests against a non-test Chargebee " + "site (expected '-test')" + ) + return key, f"https://{site}.chargebee.com/api/v2" + + +# -------------------------------------------------------------------------- +# Credential fixtures +# -------------------------------------------------------------------------- + +def _require_env(name: str) -> str: + value = os.getenv(name, "").strip() + if not value: + pytest.skip(f"{name} not set — skipping integration test") + return value + + +@pytest.fixture +def stripe_test_key(): + key = _require_env("STRIPE_TEST_API_KEY") + try: + return guard_stripe_key(key) + except LiveCredentialError as exc: + pytest.fail(str(exc)) + + +@pytest.fixture +def paddle_sandbox(): + key = _require_env("PADDLE_SANDBOX_API_KEY") + base = os.getenv("PADDLE_SANDBOX_BASE_URL", "https://sandbox-api.paddle.com") + try: + return guard_paddle_sandbox(key, base) + except LiveCredentialError as exc: + pytest.fail(str(exc)) + + +@pytest.fixture +def chargebee_test_site(): + key = _require_env("CHARGEBEE_TEST_API_KEY") + site = _require_env("CHARGEBEE_TEST_SITE") + try: + return guard_chargebee_site(key, site) + except LiveCredentialError as exc: + pytest.fail(str(exc)) diff --git a/deliverables/pm-backend/requirements.txt b/deliverables/pm-backend/requirements.txt new file mode 100644 index 0000000..c63fed0 --- /dev/null +++ b/deliverables/pm-backend/requirements.txt @@ -0,0 +1,19 @@ +fastapi>=0.110,<1.0 +uvicorn[standard]>=0.29 +pydantic>=2.6 +pydantic-settings>=2.2 +SQLAlchemy>=2.0,<2.1 + +# Optional: only required when ENCRYPT_AT_REST=true is set. +cryptography>=42.0 ; platform_python_implementation == "CPython" + +# Real billing provider. The Stripe adapter is lazy-imported, so the SDK is +# only required when BILLING_PROVIDER=stripe is set. +stripe>=8.0 + +# Used by the Chargebee adapter (BILLING_PROVIDER=chargebee). Lazy-imported. +requests>=2.31 + +# Test +pytest>=8.0 +httpx>=0.27 diff --git a/deliverables/pm-backend/tests/test_app.py b/deliverables/pm-backend/tests/test_app.py new file mode 100644 index 0000000..6630df1 --- /dev/null +++ b/deliverables/pm-backend/tests/test_app.py @@ -0,0 +1,176 @@ +"""Tests for the integrated PM backend. + +Runs with encryption disabled (default) and, in a subprocess, with +ENCRYPT_AT_REST=true to prove the opt-in path selects an encrypted column. +""" +import subprocess +import sys +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from app import main, pm_csv + + +@pytest.fixture(scope="module") +def client(): + # A context-managed TestClient fires the app's lifespan/startup handler, + # which runs init_db() so the program tables exist. + with TestClient(main.app) as c: + yield c + + +# -------------------------------------------------------------------------- +# Health + tool switcher +# -------------------------------------------------------------------------- + +def test_health(client): + r = client.get("/health") + assert r.status_code == 200 + body = r.json() + assert body["status"] == "ok" + assert "active_tool" in body + + +def test_tool_list_default(client): + r = client.get("/api/tool") + assert r.status_code == 200 + body = r.json() + assert "active_tool" in body + assert set(body["tools"]) == {"pm_csv", "billing", "settings"} + + +def test_tool_switch_roundtrip(client): + r = client.post("/api/tool", json={"tool": "billing"}) + assert r.status_code == 200 + assert r.json()["active_tool"] == "billing" + # restore + client.post("/api/tool", json={"tool": "pm_csv"}) + + +def test_tool_switch_unknown(client): + r = client.post("/api/tool", json={"tool": "nope"}) + assert r.status_code == 400 + + +# -------------------------------------------------------------------------- +# PM CSV +# -------------------------------------------------------------------------- + +def test_template_endpoint(client): + r = client.get("/api/csv/template") + assert r.status_code == 200 + assert "text/csv" in r.headers["content-type"] + text = r.text + assert "program_id" in text and "name" in text and "owner" in text + + +def test_rows_to_csv_roundtrip(): + rows = [ + {"name": "Alpha", "owner": "A", "status": "active", "budget": 1000}, + {"name": "Beta", "status": "planned", "risk": "med"}, + ] + csv_text = pm_csv.rows_to_csv(rows) + parsed = pm_csv.parse_csv(csv_text) + assert [p.name for p in parsed] == ["Alpha", "Beta"] + assert parsed[0].budget == 1000 + assert parsed[0].status == "active" + + +def test_generate_endpoint(client): + r = client.post( + "/api/csv/generate", + json={"rows": [{"name": "X", "status": "active"}]}, + ) + assert r.status_code == 200 + assert r.json()["count"] == 1 + + +def test_validate_reports_bad_status(client): + r = client.post( + "/api/csv/validate", + json={"rows": [{"name": "X", "status": "not-a-status"}]}, + ) + body = r.json() + assert body["valid"] is False + assert body["errors"] + + +# -------------------------------------------------------------------------- +# Billing (stub provider) +# -------------------------------------------------------------------------- + +def test_billing_provider_info(client): + r = client.get("/api/billing/provider") + assert r.status_code == 200 + assert r.json()["provider"] == "stub" + + +def test_billing_customer_invoice_charge_flow(client): + c = client.post( + "/api/billing/customers", json={"email": "a@b.com", "name": "A"} + ).json() + assert c["id"].startswith("cus_") + inv = client.post( + "/api/billing/invoices", + json={"customer_id": c["id"], "amount": 2500, "currency": "usd"}, + ).json() + assert inv["amount"] == 2500 and inv["status"] == "open" + pi = client.post( + "/api/billing/charges", + json={"customer_id": c["id"], "amount": 2500, "currency": "usd"}, + ).json() + assert pi["status"] == "succeeded" + + +def test_billing_unknown_customer_404(client): + r = client.post( + "/api/billing/invoices", + json={"customer_id": "missing", "amount": 100}, + ) + assert r.status_code == 404 + + +# -------------------------------------------------------------------------- +# Program registry (exercises optional encryption on 'note') +# -------------------------------------------------------------------------- + +def test_program_create_list_get(client): + created = client.post( + "/api/programs", + json={"name": "Launch", "owner": "Nattapong", "note": "secret value"}, + ).json() + pid = created["id"] + assert created["note"] == "secret value" + listed = client.get("/api/programs").json() + assert any(p["id"] == pid for p in listed) + fetched = client.get(f"/api/programs/{pid}").json() + assert fetched["note"] == "secret value" + + +# -------------------------------------------------------------------------- +# Encryption at rest round-trip in a clean subprocess +# -------------------------------------------------------------------------- + +def test_encrypt_at_rest_stores_ciphertext(): + """With ENCRYPT_AT_REST=true the note column is an EncryptedString type and + a Fernet instance is active (ciphertext on disk, decrypted on read).""" + result = subprocess.run( + [sys.executable, "-c", ( + "import os; os.environ['ENCRYPT_AT_REST']='true';" + "os.environ['ENCRYPTION_SECRET']='test-secret';" + "os.environ['DATABASE_URL']='sqlite:///:memory:';" + "from app import db;" + "col=db.ProgramRecord.__table__.c.note;" + "print(type(col.type).__name__);" + "print('encrypted' if db._fernet is not None else 'plain')" + )], + capture_output=True, + text=True, + cwd=str(Path(__file__).resolve().parent.parent), + ) + assert result.returncode == 0, result.stderr + out = result.stdout.strip().splitlines() + assert out[0] == "EncryptedString" + assert out[1] == "encrypted" diff --git a/deliverables/pm-backend/tests/test_chargebee_adapter.py b/deliverables/pm-backend/tests/test_chargebee_adapter.py new file mode 100644 index 0000000..695fd85 --- /dev/null +++ b/deliverables/pm-backend/tests/test_chargebee_adapter.py @@ -0,0 +1,168 @@ +"""Tests for the Chargebee billing adapter, verified against a mocked HTTP +layer (we do not call Chargebee's live API without a key). A fake ``requests`` +records the outbound calls and returns canned Chargebee responses so we assert +the adapter builds the right requests and maps errors correctly. +""" +import sys +from types import ModuleType + +import pytest + +from app.billing import ( + ProviderAuthError, + ProviderNotFoundError, + build_provider, +) +from app.chargebee import ChargebeeBillingProvider + + +# --- Fake requests layer ---------------------------------------------------- + +class _FakeResponse: + def __init__(self, status_code=200, json_data=None): + self.status_code = status_code + self._json = json_data or {} + self.ok = 200 <= status_code < 300 + self.text = str(json_data) + + def json(self): + return self._json + + +class FakeRequests(ModuleType): + """Module stand-in: records calls, returns configurable responses. The + adapter's own status-code handling raises the provider errors, so this + fake only returns queued responses (never raises).""" + + responses: list[_FakeResponse] = [] + calls: list[tuple] = [] + # Present so the adapter's ``except requests.RequestException`` clause + # resolves without error (it is only evaluated on a transport failure). + RequestException = Exception + + @classmethod + def reset(cls, *responses): + cls.responses = list(responses) or [_FakeResponse()] + cls.calls = [] + + @classmethod + def request(cls, method, url, **kwargs): + cls.calls.append((method, url, kwargs)) + return cls.responses.pop(0) if cls.responses else _FakeResponse() + + +@pytest.fixture +def fake_requests(): + real = sys.modules.get("requests") + fake = FakeRequests("requests") + sys.modules["requests"] = fake + FakeRequests.reset() + try: + yield fake + finally: + if real is not None: + sys.modules["requests"] = real + else: + sys.modules.pop("requests", None) + + +def _provider(): + return ChargebeeBillingProvider( + api_key="cb_test_key", + base_url="https://acme-test.chargebee.com/api/v2", + ) + + +# --- Construction ----------------------------------------------------------- + +def test_requires_api_key(): + with pytest.raises(ProviderAuthError): + ChargebeeBillingProvider(api_key="", base_url="https://x/api/v2") + + +def test_requires_base_url(): + with pytest.raises(ProviderAuthError): + ChargebeeBillingProvider(api_key="cb_key", base_url="") + + +def test_detects_mode_from_site(): + assert _provider()._mode == "test" + live = ChargebeeBillingProvider( + api_key="cb_key", base_url="https://acme.chargebee.com/api/v2" + ) + assert live._mode == "live" + + +def test_factory_registers_chargebee(fake_requests): + p = build_provider( + provider="chargebee", api_key="cb_key", + base_url="https://x-test.chargebee.com/api/v2", + ) + assert p.name == "chargebee" + + +# --- Customer --------------------------------------------------------------- + +def test_create_customer(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"customer": { + "id": "cb_cus_1", "email": "a@b.com", + "first_name": "Alice", "last_name": "Smith", + }})) + p = _provider() + c = p.create_customer("a@b.com", name="Alice Smith") + assert c.id == "cb_cus_1" + assert c.name == "Alice Smith" + # Assert it POSTed to /customers with the right JSON body. + method, url, kwargs = FakeRequests.calls[0] + assert method == "POST" and url.endswith("/customers") + assert kwargs["json"]["customer"]["email"] == "a@b.com" + + +def test_get_customer_ok(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"customer": { + "id": "cb_cus_9", "email": "e@x.com", "first_name": "X", + }})) + c = _provider().get_customer("cb_cus_9") + assert c.id == "cb_cus_9" + assert c.email == "e@x.com" + + +def test_get_missing_customer_raises_not_found(fake_requests): + FakeRequests.reset(_FakeResponse(404, {"error": "not found"})) + # 404 -> ProviderNotFoundError (mapped to HTTP 404 by the router) + with pytest.raises(ProviderNotFoundError): + _provider().get_customer("nope") + + +def test_bad_key_raises_auth_error(fake_requests): + FakeRequests.reset(_FakeResponse(401)) + with pytest.raises(ProviderAuthError): + _provider().get_customer("cb_cus_1") + + +# --- Invoice / charge ------------------------------------------------------- + +def test_create_invoice(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"invoice": { + "id": "cb_inv_1", "status": "posted", + }})) + inv = _provider().create_invoice("cb_cus_1", 2500, "usd", "Setup") + method, url, kwargs = FakeRequests.calls[0] + assert method == "POST" and url.endswith("/invoices") + # Amount is sent in major units (25.00) with currency uppercased. + assert kwargs["json"]["invoice"]["addons"][0]["amount"] == 25.0 + assert kwargs["json"]["invoice"]["currency_code"] == "USD" + assert inv.id == "cb_inv_1" + assert inv.amount == 2500 + + +def test_charge_creates_payment_intent(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"payment_intent": { + "id": "cb_pi_1", "status": "inited", + }})) + pi = _provider().charge("cb_cus_1", 1200, "usd") + method, url, kwargs = FakeRequests.calls[0] + assert method == "POST" and url.endswith("/payment_intents") + assert kwargs["json"]["payment_intent"]["amount"] == 12.0 + assert pi.id == "cb_pi_1" + assert pi.amount == 1200 diff --git a/deliverables/pm-backend/tests/test_integration_chargebee.py b/deliverables/pm-backend/tests/test_integration_chargebee.py new file mode 100644 index 0000000..99dcce4 --- /dev/null +++ b/deliverables/pm-backend/tests/test_integration_chargebee.py @@ -0,0 +1,85 @@ +"""Integration tests that hit the real Chargebee TEST site. + +Opt-in: these run only with ``pytest --run-integration`` AND test-site +credentials in ``CHARGEBEE_TEST_API_KEY`` / ``CHARGEBEE_TEST_SITE``. The safety +guard in ``conftest.py`` requires a ``*-test`` site, so this cannot touch a +production Chargebee account. Customers created here are deleted on teardown. +""" +import base64 + +import pytest +import requests + +from app.billing import ProviderNotFoundError +from app.chargebee import ChargebeeBillingProvider + +pytestmark = pytest.mark.integration + + +def _basic_auth(key: str) -> str: + token = base64.b64encode(f"{key}:".encode()).decode() + return f"Basic {token}" + + +@pytest.fixture +def provider(chargebee_test_site): + key, base = chargebee_test_site + p = ChargebeeBillingProvider(api_key=key, base_url=base) + created: list[str] = [] + yield p, created + headers = {"Authorization": _basic_auth(key)} + for cid in created: + try: + # Chargebee deletes via POST /customers//delete + requests.post( + f"{base}/customers/{cid}/delete", headers=headers, timeout=15 + ) + except requests.RequestException: + pass + + +def test_health_reports_test_mode(provider): + p, _ = provider + health = p.health() + assert health["provider"] == "chargebee" + assert health["mode"] == "test" + + +def test_create_and_get_customer(provider): + p, created = provider + c = p.create_customer("integration-cus@example.com", name="Integration Test") + created.append(c.id) + assert c.id + assert c.email == "integration-cus@example.com" + + fetched = p.get_customer(c.id) + assert fetched.id == c.id + assert fetched.email == c.email + + +def test_create_invoice(provider): + p, created = provider + c = p.create_customer("integration-inv@example.com") + created.append(c.id) + inv = p.create_invoice( + c.id, amount=2500, currency="usd", description="Integration invoice" + ) + assert inv.customer_id == c.id + assert inv.amount == 2500 + assert inv.id + + +def test_charge(provider): + p, created = provider + c = p.create_customer("integration-pi@example.com") + created.append(c.id) + pi = p.charge(c.id, amount=1200, currency="usd") + assert pi.customer_id == c.id + assert pi.amount == 1200 + assert pi.id + + +def test_missing_customer_raises_not_found(provider): + p, _ = provider + with pytest.raises(ProviderNotFoundError): + p.get_customer("__does_not_exist__") diff --git a/deliverables/pm-backend/tests/test_integration_guards.py b/deliverables/pm-backend/tests/test_integration_guards.py new file mode 100644 index 0000000..0187667 --- /dev/null +++ b/deliverables/pm-backend/tests/test_integration_guards.py @@ -0,0 +1,78 @@ +"""Unit tests for the integration-test safety guards. + +These are plain unit tests (no network, no opt-in needed): they prove the +guards refuse live credentials/sites and accept test ones, so a misconfigured +environment can never point an integration test at a production account. +""" +import pytest + +from conftest import ( + LiveCredentialError, + guard_chargebee_site, + guard_paddle_sandbox, + guard_stripe_key, +) + + +# --- Stripe --------------------------------------------------------------- + +def test_stripe_accepts_test_secret_key(): + assert guard_stripe_key("sk_test_abc123") == "sk_test_abc123" + + +def test_stripe_accepts_test_restricted_key(): + assert guard_stripe_key("rk_test_abc123") == "rk_test_abc123" + + +def test_stripe_rejects_live_secret_key(): + with pytest.raises(LiveCredentialError, match="LIVE"): + guard_stripe_key("sk_live_abc123") + + +def test_stripe_rejects_live_restricted_key(): + with pytest.raises(LiveCredentialError, match="LIVE"): + guard_stripe_key("rk_live_abc123") + + +def test_stripe_rejects_unknown_key_shape(): + with pytest.raises(LiveCredentialError, match="test-mode"): + guard_stripe_key("nonsense") + + +def test_stripe_rejects_empty_key(): + with pytest.raises(LiveCredentialError): + guard_stripe_key("") + + +# --- Paddle --------------------------------------------------------------- + +def test_paddle_accepts_sandbox(): + key, base = guard_paddle_sandbox("pdl_sdbx_abc", "https://sandbox-api.paddle.com") + assert base == "https://sandbox-api.paddle.com" + + +def test_paddle_rejects_live_base_url(): + with pytest.raises(LiveCredentialError, match="sandbox-api"): + guard_paddle_sandbox("pdl_abc", "https://api.paddle.com") + + +def test_paddle_rejects_live_key_marker(): + with pytest.raises(LiveCredentialError, match="LIVE"): + guard_paddle_sandbox("pdl_live_abc", "https://sandbox-api.paddle.com") + + +# --- Chargebee ------------------------------------------------------------ + +def test_chargebee_accepts_test_site(): + key, base = guard_chargebee_site("cb_key", "acme-test") + assert base == "https://acme-test.chargebee.com/api/v2" + + +def test_chargebee_rejects_production_site(): + with pytest.raises(LiveCredentialError, match="non-test"): + guard_chargebee_site("cb_key", "acme") + + +def test_chargebee_rejects_empty_site(): + with pytest.raises(LiveCredentialError): + guard_chargebee_site("cb_key", "") diff --git a/deliverables/pm-backend/tests/test_integration_paddle.py b/deliverables/pm-backend/tests/test_integration_paddle.py new file mode 100644 index 0000000..a000813 --- /dev/null +++ b/deliverables/pm-backend/tests/test_integration_paddle.py @@ -0,0 +1,75 @@ +"""Integration tests that hit the real Paddle SANDBOX API. + +Opt-in: these run only with ``pytest --run-integration`` AND sandbox +credentials in ``PADDLE_SANDBOX_API_KEY``. The safety guard in ``conftest.py`` +forces the sandbox host and rejects live keys, so this cannot touch a live +Paddle account. Customers created here are deleted on teardown. +""" +import pytest +import requests + +from app.billing import ProviderNotFoundError +from app.paddle import PaddleBillingProvider + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def provider(paddle_sandbox): + key, base = paddle_sandbox + p = PaddleBillingProvider(api_key=key, base_url=base) + created: list[str] = [] + yield p, created + headers = {"Authorization": f"Bearer {key}"} + for cid in created: + try: + requests.delete(f"{base}/customers/{cid}", headers=headers, timeout=15) + except requests.RequestException: + pass + + +def test_health_reports_test_mode(provider): + p, _ = provider + health = p.health() + assert health["provider"] == "paddle" + assert health["mode"] == "test" + + +def test_create_and_get_customer(provider): + p, created = provider + c = p.create_customer("integration-cus@example.com", name="Integration Test") + created.append(c.id) + assert c.id.startswith("ctm_") + assert c.email == "integration-cus@example.com" + + fetched = p.get_customer(c.id) + assert fetched.id == c.id + assert fetched.email == c.email + + +def test_create_invoice(provider): + p, created = provider + c = p.create_customer("integration-inv@example.com") + created.append(c.id) + inv = p.create_invoice( + c.id, amount=2500, currency="usd", description="Integration invoice" + ) + assert inv.customer_id == c.id + assert inv.amount == 2500 + assert inv.id + + +def test_charge(provider): + p, created = provider + c = p.create_customer("integration-pi@example.com") + created.append(c.id) + pi = p.charge(c.id, amount=1200, currency="usd") + assert pi.customer_id == c.id + assert pi.amount == 1200 + assert pi.id + + +def test_missing_customer_raises_not_found(provider): + p, _ = provider + with pytest.raises(ProviderNotFoundError): + p.get_customer("ctm_does_not_exist_12345") diff --git a/deliverables/pm-backend/tests/test_integration_stripe.py b/deliverables/pm-backend/tests/test_integration_stripe.py new file mode 100644 index 0000000..8421106 --- /dev/null +++ b/deliverables/pm-backend/tests/test_integration_stripe.py @@ -0,0 +1,75 @@ +"""Integration tests that hit the real Stripe TEST API. + +Opt-in: these run only with ``pytest --run-integration`` AND a test-mode key in +``STRIPE_TEST_API_KEY``. The safety guard in ``conftest.py`` refuses live keys, +so this can never touch a production account. Customers created here are +deleted on teardown. +""" +import pytest + +from app.billing import ProviderNotFoundError, StripeBillingProvider + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def provider(stripe_test_key): + p = StripeBillingProvider(api_key=stripe_test_key) + created: list[str] = [] + yield p, created + # Best-effort teardown so repeated runs don't accumulate test customers. + import stripe + + for cid in created: + try: + stripe.Customer.delete(cid) + except Exception: + pass + + +def test_health_reports_test_mode(provider): + p, _ = provider + health = p.health() + assert health["provider"] == "stripe" + assert health["mode"] == "test" + + +def test_create_and_get_customer(provider): + p, created = provider + c = p.create_customer("integration+cus@example.com", name="Integration Test") + created.append(c.id) + assert c.id.startswith("cus_") + assert c.email == "integration+cus@example.com" + assert c.name == "Integration Test" + + fetched = p.get_customer(c.id) + assert fetched.id == c.id + assert fetched.email == c.email + + +def test_create_invoice(provider): + p, created = provider + c = p.create_customer("integration+inv@example.com") + created.append(c.id) + inv = p.create_invoice( + c.id, amount=2500, currency="usd", description="Integration invoice" + ) + assert inv.customer_id == c.id + assert inv.amount == 2500 + assert inv.status in {"open", "draft", "paid"} + + +def test_charge(provider): + p, created = provider + c = p.create_customer("integration+pi@example.com") + created.append(c.id) + pi = p.charge(c.id, amount=1200, currency="usd") + assert pi.customer_id == c.id + assert pi.amount == 1200 + assert pi.status + + +def test_missing_customer_raises_not_found(provider): + p, _ = provider + with pytest.raises(ProviderNotFoundError): + p.get_customer("cus_does_not_exist_12345") diff --git a/deliverables/pm-backend/tests/test_paddle_adapter.py b/deliverables/pm-backend/tests/test_paddle_adapter.py new file mode 100644 index 0000000..5dbef5b --- /dev/null +++ b/deliverables/pm-backend/tests/test_paddle_adapter.py @@ -0,0 +1,163 @@ +"""Tests for the Paddle billing adapter, verified against a mocked HTTP layer +(no live Paddle key). A fake ``requests`` records outbound calls and returns +canned Paddle Billing responses (wrapped in ``{"data": ...}``) so we assert the +adapter builds correct requests and maps errors to the neutral hierarchy. +""" +import sys +from types import ModuleType + +import pytest + +from app.billing import ProviderAuthError, ProviderNotFoundError, build_provider +from app.paddle import PaddleBillingProvider + + +class _FakeResponse: + def __init__(self, status_code=200, json_data=None): + self.status_code = status_code + self._json = json_data or {} + self.ok = 200 <= status_code < 300 + self.text = str(json_data) + + def json(self): + return self._json + + +class FakeRequests(ModuleType): + """Returns queued responses; the adapter's own status handling raises.""" + + responses: list[_FakeResponse] = [] + calls: list[tuple] = [] + RequestException = Exception + + @classmethod + def reset(cls, *responses): + cls.responses = list(responses) or [_FakeResponse()] + cls.calls = [] + + @classmethod + def request(cls, method, url, **kwargs): + cls.calls.append((method, url, kwargs)) + return cls.responses.pop(0) if cls.responses else _FakeResponse() + + +@pytest.fixture +def fake_requests(): + real = sys.modules.get("requests") + fake = FakeRequests("requests") + sys.modules["requests"] = fake + FakeRequests.reset() + try: + yield fake + finally: + if real is not None: + sys.modules["requests"] = real + else: + sys.modules.pop("requests", None) + + +def _provider(): + return PaddleBillingProvider(api_key="pdl_test_key") + + +# --- Construction ----------------------------------------------------------- + +def test_requires_api_key(): + with pytest.raises(ProviderAuthError): + PaddleBillingProvider(api_key="") + + +def test_defaults_to_live_mode(): + assert _provider()._mode == "live" + + +def test_sandbox_url_is_test_mode(): + p = PaddleBillingProvider( + api_key="pdl_key", base_url="https://sandbox-api.paddle.com" + ) + assert p._mode == "test" + + +def test_factory_registers_paddle(fake_requests): + p = build_provider(provider="paddle", api_key="pdl_key", base_url="") + assert p.name == "paddle" + + +# --- Customer --------------------------------------------------------------- + +def test_create_customer(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"data": { + "id": "ctm_1", "email": "a@b.com", "name": "Alice", + }})) + c = _provider().create_customer("a@b.com", name="Alice") + assert c.id == "ctm_1" + assert c.name == "Alice" + method, url, kwargs = FakeRequests.calls[0] + assert method == "POST" and url.endswith("/customers") + assert kwargs["json"] == {"email": "a@b.com", "name": "Alice"} + # Bearer auth header is set. + assert kwargs["headers"]["Authorization"] == "Bearer pdl_test_key" + + +def test_get_customer_ok(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"data": { + "id": "ctm_9", "email": "e@x.com", + }})) + c = _provider().get_customer("ctm_9") + assert c.id == "ctm_9" + assert c.email == "e@x.com" + + +def test_get_missing_customer_raises_not_found(fake_requests): + FakeRequests.reset(_FakeResponse(404, {"error": "not found"})) + with pytest.raises(ProviderNotFoundError): + _provider().get_customer("nope") + + +def test_bad_key_raises_auth_error(fake_requests): + FakeRequests.reset(_FakeResponse(401)) + with pytest.raises(ProviderAuthError): + _provider().get_customer("ctm_1") + + +# --- Transaction / invoice / charge ----------------------------------------- + +def test_create_invoice_uses_manual_collection(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"data": { + "id": "txn_1", "status": "draft", + }})) + inv = _provider().create_invoice("ctm_1", 2500, "usd", "Setup") + method, url, kwargs = FakeRequests.calls[0] + assert method == "POST" and url.endswith("/transactions") + body = kwargs["json"] + assert body["collection_mode"] == "manual" + assert body["customer_id"] == "ctm_1" + # Amount is a minor-unit STRING with uppercased currency. + price = body["items"][0]["price"] + assert price["unit_price"]["amount"] == "2500" + assert price["unit_price"]["currency_code"] == "USD" + assert price["description"] == "Setup" + assert inv.id == "txn_1" + assert inv.amount == 2500 + assert inv.status == "draft" + + +def test_charge_uses_automatic_collection(fake_requests): + FakeRequests.reset(_FakeResponse(200, {"data": { + "id": "txn_2", "status": "ready", + }})) + pi = _provider().charge("ctm_1", 1200, "usd") + body = FakeRequests.calls[0][2]["json"] + assert body["collection_mode"] == "automatic" + assert body["items"][0]["price"]["unit_price"]["amount"] == "1200" + assert pi.id == "txn_2" + assert pi.amount == 1200 + assert pi.status == "ready" + + +def test_upstream_error_is_provider_error(fake_requests): + from app.billing import ProviderError + + FakeRequests.reset(_FakeResponse(500, {"error": "boom"})) + with pytest.raises(ProviderError): + _provider().get_customer("ctm_1") diff --git a/deliverables/pm-backend/tests/test_stripe_adapter.py b/deliverables/pm-backend/tests/test_stripe_adapter.py new file mode 100644 index 0000000..6d28382 --- /dev/null +++ b/deliverables/pm-backend/tests/test_stripe_adapter.py @@ -0,0 +1,222 @@ +"""Tests for the real Stripe billing adapter, verified with a mocked SDK. + +We cannot call Stripe's live API without a test-mode key, so we assert the +adapter drives the Stripe SDK correctly by replacing the SDK's resource +classes with fakes that mimic real Stripe objects (dict-like, keyed access). +This validates the adapter's calls, argument translation, and error mapping +against the semantics Stripe actually returns. +""" +from types import ModuleType +import sys + +import pytest + +from app.billing import ( + ProviderAuthError, + ProviderNotFoundError, + StripeBillingProvider, + build_provider, +) + + +# --- Fake Stripe SDK -------------------------------------------------------- + +class FakeCustomer(dict): + @classmethod + def create(cls, **kw): + return cls(id="cus_test123", email=kw.get("email", ""), name=kw.get("name", "")) + + @classmethod + def retrieve(cls, customer_id): + if customer_id == "missing": + raise _make_stripe_error("InvalidRequestError", "No such customer") + return cls(id=customer_id, email="e@x.com", name="X") + + +class FakeInvoiceItem(dict): + @classmethod + def create(cls, **kw): + return cls(id="ii_test", **kw) + + +class FakeInvoice(dict): + @classmethod + def create(cls, **kw): + return cls(id="in_test_draft", customer=kw.get("customer", ""), status="draft") + + @classmethod + def finalize_invoice(cls, invoice_id): + return cls(id=invoice_id, status="open") + + +class FakePaymentIntent(dict): + @classmethod + def create(cls, **kw): + return cls( + id="pi_test123", + amount=kw.get("amount", 0), + currency=kw.get("currency", "usd"), + customer=kw.get("customer", ""), + status="requires_payment_method", + ) + + +class _FakeErrors: + class Error(Exception): + pass + + class AuthenticationError(Error): + pass + + class InvalidRequestError(Error): + pass + + class CardError(Error): + pass + + +class FakeStripe(ModuleType): + api_key = "sk_test_dummy" + api_base = "https://api.stripe.com" + Customer = FakeCustomer + Invoice = FakeInvoice + InvoiceItem = FakeInvoiceItem + PaymentIntent = FakePaymentIntent + error = _FakeErrors + + +def _make_stripe_error(kind, msg): + exc = getattr(_FakeErrors, kind)(msg) + # Stripe SDK error objects expose an attribute-style payload. + exc.error = type("E", (), {"message": msg})() + return exc + + +@pytest.fixture +def fake_stripe_module(): + """Inject a fake Stripe module into sys.modules so the adapter's lazy + ``import stripe`` picks it up, and restore the real module after.""" + real = sys.modules.get("stripe") + fake = FakeStripe("stripe") + sys.modules["stripe"] = fake + try: + yield fake + finally: + if real is not None: + sys.modules["stripe"] = real + else: + sys.modules.pop("stripe", None) + + +def _provider(fake_stripe_module) -> StripeBillingProvider: + return StripeBillingProvider(api_key="sk_test_dummy") + + +# --- Construction ----------------------------------------------------------- + +def test_requires_api_key(): + with pytest.raises(ProviderAuthError): + StripeBillingProvider(api_key="") + + +def test_missing_key_at_factory_raises(): + with pytest.raises(ProviderAuthError): + build_provider(provider="stripe", api_key="", base_url="") + + +def test_detects_test_mode(fake_stripe_module): + assert _provider(fake_stripe_module)._mode == "test" + + +def test_detects_live_mode(fake_stripe_module): + p = StripeBillingProvider(api_key="sk_live_abc") + assert p._mode == "live" + + +# --- Customer lifecycle ----------------------------------------------------- + +def test_create_customer(fake_stripe_module): + p = _provider(fake_stripe_module) + c = p.create_customer("a@b.com", name="Alice") + assert c.id == "cus_test123" + assert c.email == "a@b.com" + assert c.name == "Alice" + + +def test_get_customer_ok(fake_stripe_module): + c = _provider(fake_stripe_module).get_customer("cus_known") + assert c.id == "cus_known" + + +def test_get_missing_customer_raises_not_found(fake_stripe_module): + with pytest.raises(ProviderNotFoundError): + _provider(fake_stripe_module).get_customer("missing") + + +# --- Invoice lifecycle ------------------------------------------------------ + +def test_create_invoice_flow(fake_stripe_module): + """create_invoice must: create draft -> add line item -> finalize.""" + # Replace the fake resource methods with mocks so we can assert the + # adapter really drives all three Stripe calls with correct arguments. + sm = fake_stripe_module + from unittest.mock import Mock + + sm.Invoice.create = Mock(return_value=sm.Invoice(id="in_draft", status="draft")) + sm.InvoiceItem.create = Mock(return_value=sm.InvoiceItem(id="ii_1")) + sm.Invoice.finalize_invoice = Mock( + return_value=sm.Invoice(id="in_draft", status="open") + ) + + p = _provider(fake_stripe_module) + inv = p.create_invoice("cus_test123", amount=2500, currency="usd", + description="Setup") + + sm.Invoice.create.assert_called_once_with(customer="cus_test123") + sm.InvoiceItem.create.assert_called_once_with( + customer="cus_test123", amount=2500, currency="usd", + description="Setup", invoice="in_draft", + ) + sm.Invoice.finalize_invoice.assert_called_once_with("in_draft") + assert inv.id == "in_draft" + assert inv.status == "open" + assert inv.amount == 2500 + + +# --- Charges ---------------------------------------------------------------- + +def test_charge_creates_payment_intent(fake_stripe_module): + pi = _provider(fake_stripe_module).charge("cus_test123", 1200, "usd") + assert pi.id == "pi_test123" + assert pi.amount == 1200 + assert pi.status == "requires_payment_method" + + +def test_missing_customer_charge_maps_to_provider_error(fake_stripe_module): + # The adapter should surface Stripe errors; our fake customer validation + # only triggers on get_customer, so simulate by patching create. + fake_stripe_module.PaymentIntent = type( + "BadPI", (), {"create": staticmethod(lambda **kw: (_ for _ in ()).throw( + _make_stripe_error("InvalidRequestError", "No such customer: cus_x") + ))} + ) + with pytest.raises(ProviderNotFoundError): + _provider(fake_stripe_module).charge("cus_x", 100, "usd") + + +# --- Error translation ------------------------------------------------------ + +def test_auth_error_maps_to_provider_auth(fake_stripe_module): + fake_stripe_module.Customer = type( + "BadCust", (), {"create": staticmethod(lambda **kw: (_ for _ in ()).throw( + _make_stripe_error("AuthenticationError", "Invalid API key") + ))} + ) + with pytest.raises(ProviderAuthError): + _provider(fake_stripe_module).create_customer("a@b.com") + + +def test_health_reports_mode(fake_stripe_module): + h = _provider(fake_stripe_module).health() + assert h["provider"] == "stripe" + assert h["mode"] == "test"