1666511475
- services/kb-postgres: pgvector/pgvector:pg16 na SOLARIA (:5433), named volume, init/001_envelope.sql (CREATE EXTENSION vector + zamrożona tabela envelope: id/source/ts/geo/raw_ref/entities), service.yaml, healthcheck, README z poprawnym mechanizmem deploy (deploy-node.sh składa dwa -f) - hosts/solaria/runtime/kb-postgres/docker-compose.override.yml: mem_limit 4g - inventory/topology.yaml + hosts/solaria/services.yaml: kb-postgres wpisany - packages/kb-mail: nowa konwencja shared lib (pip install /repo/packages/<lib>/) envelope.py — @dataclass Envelope, walidacja tz-aware ts db.py — insert_envelope / get_envelope (asyncpg, ON CONFLICT DO NOTHING) archive.py — save_eml append-only (asyncio.to_thread, FileExistsError na dup) tests: 15 unit pass + 5 integration (@pytest.mark.integration, wymaga KB_TEST_DSN) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""Sanity tests for SQL migration files — no DB or network required."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).parent.parent.parent.parent
|
|
INIT_DIR = REPO_ROOT / "services" / "kb-postgres" / "init"
|
|
|
|
|
|
def test_migration_001_exists():
|
|
assert (INIT_DIR / "001_envelope.sql").exists(), \
|
|
f"Expected {INIT_DIR / '001_envelope.sql'} to exist"
|
|
|
|
|
|
def test_migration_001_enables_vector_extension():
|
|
sql = (INIT_DIR / "001_envelope.sql").read_text()
|
|
assert "CREATE EXTENSION" in sql
|
|
assert "vector" in sql.lower()
|
|
|
|
|
|
def test_migration_001_creates_envelope_table():
|
|
sql = (INIT_DIR / "001_envelope.sql").read_text()
|
|
assert "CREATE TABLE" in sql
|
|
assert "envelope" in sql.lower()
|
|
|
|
|
|
def test_migration_001_has_all_frozen_columns():
|
|
sql = (INIT_DIR / "001_envelope.sql").read_text()
|
|
frozen_columns = ("id", "source", "ts", "geo", "raw_ref", "entities")
|
|
for col in frozen_columns:
|
|
assert col in sql, f"Frozen column '{col}' missing from 001_envelope.sql"
|
|
|
|
|
|
def test_migration_001_geo_is_nullable():
|
|
sql = (INIT_DIR / "001_envelope.sql").read_text()
|
|
# geo must NOT have NOT NULL constraint
|
|
lines = [l for l in sql.splitlines() if "geo" in l.lower()]
|
|
assert lines, "No 'geo' line found in migration SQL"
|
|
for line in lines:
|
|
assert "NOT NULL" not in line.upper(), \
|
|
f"geo must be nullable but found: {line.strip()}"
|
|
|
|
|
|
def test_migration_001_entities_has_default():
|
|
sql = (INIT_DIR / "001_envelope.sql").read_text()
|
|
lines = [l for l in sql.splitlines() if "entities" in l.lower()]
|
|
assert any("DEFAULT" in l.upper() for l in lines), \
|
|
"entities column must have a DEFAULT '[]'"
|