62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
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')"
|
|
)
|
|
|
|
|
|
def test_agent_dsn_escapes_password_with_space_and_quote():
|
|
import psycopg
|
|
from agent.config import AgentConfig
|
|
from agent.db import agent_dsn
|
|
|
|
cfg = AgentConfig(
|
|
db_host="localhost",
|
|
db_port=5432,
|
|
db_user="testuser",
|
|
db_password="pa ss'word", # password with space and single quote
|
|
agent_db_name="testdb",
|
|
firefly_url="http://firefly.test",
|
|
firefly_token="test-token",
|
|
heartbeat_interval_minutes=60,
|
|
)
|
|
|
|
dsn = agent_dsn(cfg)
|
|
# Parse DSN and verify password is preserved correctly
|
|
parsed = psycopg.conninfo.conninfo_to_dict(dsn)
|
|
assert parsed["password"] == "pa ss'word"
|