diff --git a/agent/src/agent/db.py b/agent/src/agent/db.py new file mode 100644 index 0000000..9d500b4 --- /dev/null +++ b/agent/src/agent/db.py @@ -0,0 +1,70 @@ +"""agentdb access: bootstrap, migrations, connections.""" + +from __future__ import annotations + +from pathlib import Path + +import psycopg + +from agent.config import AgentConfig + +_MIGRATIONS_DIR = Path(__file__).parent / "migrations" + + +def _dsn(cfg: AgentConfig, dbname: str) -> str: + return ( + f"host={cfg.db_host} port={cfg.db_port} dbname={dbname} " + f"user={cfg.db_user} password={cfg.db_password}" + ) + + +def admin_dsn(cfg: AgentConfig) -> str: + """DSN for the maintenance database that always exists.""" + return _dsn(cfg, "postgres") + + +def agent_dsn(cfg: AgentConfig) -> str: + return _dsn(cfg, cfg.agent_db_name) + + +def connect(cfg: AgentConfig) -> psycopg.Connection: + return psycopg.connect(agent_dsn(cfg)) + + +def ensure_database(cfg: AgentConfig) -> None: + """Create the agent database if it does not exist. Idempotent.""" + with psycopg.connect(admin_dsn(cfg), autocommit=True) as conn: + exists = conn.execute( + "SELECT 1 FROM pg_database WHERE datname = %s", (cfg.agent_db_name,) + ).fetchone() + if not exists: + # CREATE DATABASE cannot be parameterized; the name comes from + # our own config, quoted defensively via psycopg's identifier API. + from psycopg import sql + + conn.execute( + sql.SQL("CREATE DATABASE {}").format(sql.Identifier(cfg.agent_db_name)) + ) + + +def run_migrations(cfg: AgentConfig) -> list[str]: + """Apply pending migrations in filename order. Returns names applied.""" + applied: list[str] = [] + with connect(cfg) as conn: + conn.execute( + "CREATE TABLE IF NOT EXISTS schema_migrations (" + " name TEXT PRIMARY KEY," + " applied_at TIMESTAMPTZ NOT NULL DEFAULT now())" + ) + done = { + row[0] for row in conn.execute("SELECT name FROM schema_migrations") + } + for path in sorted(_MIGRATIONS_DIR.glob("*.sql")): + if path.name in done: + continue + conn.execute(path.read_text()) + conn.execute( + "INSERT INTO schema_migrations (name) VALUES (%s)", (path.name,) + ) + applied.append(path.name) + return applied diff --git a/agent/src/agent/migrations/0001_initial.sql b/agent/src/agent/migrations/0001_initial.sql new file mode 100644 index 0000000..e661137 --- /dev/null +++ b/agent/src/agent/migrations/0001_initial.sql @@ -0,0 +1,47 @@ +-- agentdb schema v1 (see spec: agentdb schema section) + +CREATE TABLE sync_runs ( + id BIGSERIAL PRIMARY KEY, + connector TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL, + finished_at TIMESTAMPTZ, + status TEXT NOT NULL CHECK (status IN ('running', 'ok', 'error')), + items_upserted INTEGER NOT NULL DEFAULT 0, + detail TEXT +); + +CREATE INDEX sync_runs_connector_id_idx ON sync_runs (connector, id DESC); + +CREATE TABLE documents ( + id BIGSERIAL PRIMARY KEY, + path TEXT NOT NULL UNIQUE, + sha256 CHAR(64) NOT NULL UNIQUE, + institution TEXT NOT NULL, + account TEXT, + doc_type TEXT NOT NULL, + period TEXT, + source TEXT NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE bills ( + id BIGSERIAL PRIMARY KEY, + payee TEXT NOT NULL, + expected_amount NUMERIC(12,2), + due_date DATE NOT NULL, + funding_account TEXT, + autopay BOOLEAN NOT NULL DEFAULT false, + status TEXT NOT NULL DEFAULT 'upcoming' + CHECK (status IN ('upcoming', 'approved', 'paid', 'skipped', 'overdue')) +); + +CREATE TABLE approvals ( + id BIGSERIAL PRIMARY KEY, + bill_id BIGINT NOT NULL REFERENCES bills (id), + amount NUMERIC(12,2) NOT NULL, + requested_at TIMESTAMPTZ NOT NULL DEFAULT now(), + decided_at TIMESTAMPTZ, + decision TEXT CHECK (decision IN ('approved', 'denied')), + telegram_message_id TEXT, + executed_at TIMESTAMPTZ +); diff --git a/agent/tests/conftest.py b/agent/tests/conftest.py new file mode 100644 index 0000000..2661a90 --- /dev/null +++ b/agent/tests/conftest.py @@ -0,0 +1,29 @@ +import itertools + +import pytest +from testcontainers.postgres import PostgresContainer + +from agent.config import AgentConfig + +_db_counter = itertools.count() + + +@pytest.fixture(scope="session") +def pg(): + with PostgresContainer("postgres:16-alpine") as container: + yield container + + +@pytest.fixture +def cfg(pg) -> AgentConfig: + """AgentConfig pointing at a fresh, uniquely named database per test.""" + return AgentConfig( + db_host=pg.get_container_host_ip(), + db_port=int(pg.get_exposed_port(5432)), + db_user=pg.username, + db_password=pg.password, + agent_db_name=f"agentdb_test_{next(_db_counter)}", + firefly_url="http://firefly.test", + firefly_token="test-token", + heartbeat_interval_minutes=60, + ) diff --git a/agent/tests/test_db.py b/agent/tests/test_db.py new file mode 100644 index 0000000..82414db --- /dev/null +++ b/agent/tests/test_db.py @@ -0,0 +1,39 @@ +from agent.db import connect, ensure_database, run_migrations + +EXPECTED_TABLES = {"schema_migrations", "sync_runs", "documents", "bills", "approvals"} + + +def test_ensure_database_creates_and_is_idempotent(cfg): + ensure_database(cfg) + ensure_database(cfg) # second call must not raise + with connect(cfg) as conn: + assert conn.execute("SELECT 1").fetchone()[0] == 1 + + +def test_run_migrations_applies_each_migration_exactly_once(cfg): + ensure_database(cfg) + + first = run_migrations(cfg) + assert first == ["0001_initial.sql"] + + second = run_migrations(cfg) + assert second == [] + + with connect(cfg) as conn: + rows = conn.execute( + "SELECT tablename FROM pg_tables WHERE schemaname = 'public'" + ).fetchall() + assert EXPECTED_TABLES <= {r[0] for r in rows} + + +def test_sync_runs_rejects_bad_status(cfg): + import psycopg + import pytest + + ensure_database(cfg) + run_migrations(cfg) + with connect(cfg) as conn, pytest.raises(psycopg.errors.CheckViolation): + conn.execute( + "INSERT INTO sync_runs (connector, started_at, status)" + " VALUES ('x', now(), 'bogus')" + )