jonatanritter
1791435a8d
Product is the only tenant table this service owns (porte table: organization_id -> product_id, everything below it -- tenant_ref/ branch_ref -- stays an opaque string owned by the consuming product, never a row here). Bcrypt-hashed API keys (passlib, same CryptContext shape as the auto's auth.service), generated once by scripts/create_product.py and never persisted in the clear. require_product (X-Api-Key -> Product, 401 on missing/wrong/soft-deleted) is the porte adaptation of the auto's require_permission — every /v1/* route will depend on it instead of a JWT bearer token. Also ports shared/base_model.py and shared/errors.py verbatim (Global Constraints: soft delete mixins, structured 409 bodies). products migration + real "alembic upgrade head" round-trip test (tests/migrations/, new shared _helpers.py instead of the auto's ad hoc cross-file _run_psql reuse). 13 tests green via `make k8s-test`.
50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
"""Shared plumbing for the "real migration" tests (Global Constraints:
|
|
every migration ships with a test that runs `alembic upgrade head` as a
|
|
REAL subprocess against a disposable database -- never just
|
|
`Base.metadata.create_all`, which builds the schema straight from the
|
|
current model definitions and so can never catch a migration that drifted
|
|
from them). Mirrors the technique `auto/backend/tests/migrations/` uses
|
|
(e.g. `test_seed_cadastros_peca_pessoa_editar_permission.py`'s `_run_psql`),
|
|
centralized here instead of re-imported test-file-to-test-file -- this repo
|
|
starts that convention fresh rather than porting the auto's ad hoc
|
|
cross-file reuse."""
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
REPO_DIR = Path(__file__).resolve().parents[2]
|
|
|
|
_PSQL = "psql"
|
|
|
|
|
|
def run_psql(*args: str) -> subprocess.CompletedProcess:
|
|
"""Runs `psql -U postgres -h localhost <args>` against the pod's
|
|
Postgres sidecar (trust auth -- see `auto/k8s/test-runner.yaml`'s
|
|
header comment for why localhost/no-password is assumed) and asserts
|
|
it succeeded."""
|
|
result = subprocess.run(
|
|
[_PSQL, "-U", "postgres", "-h", "localhost", *args],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
assert result.returncode == 0, f"psql failed: {result.stderr}"
|
|
return result
|
|
|
|
|
|
def run_alembic(database_url: str, *args: str) -> subprocess.CompletedProcess:
|
|
"""Runs `uv run alembic <args>` as a subprocess with `DATABASE_URL`
|
|
pointed at `database_url` (never the service's own `settings.
|
|
database_url` default) -- `alembic/env.py` reads `DATABASE_URL` via
|
|
`fiscal_svc.core.config.settings`, which layers real env vars over
|
|
`.env`."""
|
|
env = {**os.environ, "DATABASE_URL": database_url}
|
|
return subprocess.run(
|
|
["uv", "run", "alembic", *args],
|
|
cwd=str(REPO_DIR),
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
)
|