105 lines
5.2 KiB
Python
105 lines
5.2 KiB
Python
"""Optional PostgreSQL contract tests.
|
|
|
|
These tests are intentionally gated by a dedicated disposable-test URL. They
|
|
never fall back to DATABASE_URL.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import uuid
|
|
from contextlib import nullcontext
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
|
|
|
|
TEST_URL = os.getenv("CLIENTFLOW_TEST_DATABASE_URL")
|
|
pytestmark = pytest.mark.skipif(not TEST_URL, reason="CLIENTFLOW_TEST_DATABASE_URL not set")
|
|
|
|
|
|
@pytest.fixture()
|
|
def pg_conn():
|
|
engine = create_engine(TEST_URL)
|
|
schema = "document_reconciliation_test_" + uuid.uuid4().hex
|
|
with engine.connect() as conn:
|
|
conn.execute(text(f'CREATE SCHEMA "{schema}"'))
|
|
conn.commit()
|
|
conn.execute(text(f'SET search_path TO "{schema}"'))
|
|
conn.execute(text("""CREATE TABLE opportunities(id UUID PRIMARY KEY);
|
|
CREATE TABLE commercial_documents(
|
|
id UUID PRIMARY KEY, opportunity_id UUID REFERENCES opportunities(id),
|
|
document_kind TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'related',
|
|
is_primary BOOLEAN NOT NULL DEFAULT FALSE, is_active BOOLEAN NOT NULL DEFAULT TRUE,
|
|
updated_at TIMESTAMPTZ DEFAULT now());
|
|
CREATE TABLE commercial_document_lines(
|
|
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), document_id UUID REFERENCES commercial_documents(id),
|
|
opportunity_item_id UUID, line_index INTEGER, local_product_id UUID,
|
|
jasmin_sales_item TEXT, description TEXT, quantity NUMERIC, unit TEXT,
|
|
unit_price NUMERIC, tax_schema TEXT, total_amount NUMERIC, payload JSONB,
|
|
created_at TIMESTAMPTZ DEFAULT now());
|
|
CREATE TABLE schema_migrations(version TEXT PRIMARY KEY);"""))
|
|
conn.commit()
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.rollback()
|
|
conn.execute(text("SET search_path TO public"))
|
|
conn.execute(text(f'DROP SCHEMA "{schema}" CASCADE'))
|
|
conn.commit()
|
|
engine.dispose()
|
|
|
|
|
|
def test_dedicated_url_is_the_only_postgres_test_source(pg_conn):
|
|
assert TEST_URL
|
|
assert str(pg_conn.execute(text("SELECT current_database()" )).scalar())
|
|
|
|
|
|
def test_migration_007_postgres_contracts_and_generic_writer(pg_conn, monkeypatch):
|
|
sql = Path("migrations/007_document_reconciliation_v2.sql").read_text()
|
|
pg_conn.exec_driver_sql(sql)
|
|
pg_conn.commit()
|
|
pg_conn.exec_driver_sql("SET search_path TO " + pg_conn.exec_driver_sql("SELECT current_schema()").scalar())
|
|
oid, first, second = [str(uuid.uuid4()) for _ in range(3)]
|
|
pg_conn.execute(text("INSERT INTO opportunities(id) VALUES(CAST(:id AS UUID))"), {"id": oid})
|
|
pg_conn.execute(text("""INSERT INTO commercial_documents(id,opportunity_id,document_kind)
|
|
VALUES(CAST(:a AS UUID),CAST(:o AS UUID),'invoice'),
|
|
(CAST(:b AS UUID),CAST(:o AS UUID),'invoice')"""), {"a": first, "b": second, "o": oid})
|
|
pg_conn.execute(text("""INSERT INTO opportunity_document_links
|
|
(opportunity_id,document_id,document_kind,relationship,source)
|
|
VALUES(CAST(:o AS UUID),CAST(:d AS UUID),'invoice','PRIMARY','test')"""), {"o": oid, "d": first})
|
|
pg_conn.commit()
|
|
with pytest.raises(Exception):
|
|
pg_conn.execute(text("""INSERT INTO opportunity_document_links
|
|
(opportunity_id,document_id,document_kind,relationship,source)
|
|
VALUES(CAST(:o AS UUID),CAST(:d AS UUID),'invoice','PRIMARY','test')"""), {"o": oid, "d": second})
|
|
pg_conn.rollback()
|
|
|
|
import app.commercial_service as commercial
|
|
class BoundEngine:
|
|
def begin(self): return nullcontext(pg_conn)
|
|
monkeypatch.setattr(commercial, "engine", BoundEngine())
|
|
monkeypatch.setattr(commercial, "ensure_commercial_schema", lambda: None)
|
|
commercial.add_document_lines(first, [{"description": "line"}])
|
|
ids = pg_conn.execute(text("""SELECT document_id::text, commercial_document_id::text
|
|
FROM commercial_document_lines""")).first()
|
|
assert ids == (first, first)
|
|
|
|
|
|
def test_migration_007_append_only_and_down(pg_conn):
|
|
pg_conn.exec_driver_sql(Path("migrations/007_document_reconciliation_v2.sql").read_text())
|
|
oid, did = str(uuid.uuid4()), str(uuid.uuid4())
|
|
pg_conn.execute(text("INSERT INTO opportunities(id) VALUES(CAST(:id AS UUID))"), {"id": oid})
|
|
pg_conn.execute(text("""INSERT INTO commercial_documents(id,opportunity_id,document_kind)
|
|
VALUES(CAST(:d AS UUID),CAST(:o AS UUID),'invoice')"""), {"d": did, "o": oid})
|
|
event = pg_conn.execute(text("""INSERT INTO opportunity_document_link_events
|
|
(opportunity_id,document_id,event_type,actor) VALUES(CAST(:o AS UUID),CAST(:d AS UUID),'TEST','test')
|
|
RETURNING id::text"""), {"o": oid, "d": did}).scalar()
|
|
with pytest.raises(Exception):
|
|
pg_conn.execute(text("UPDATE opportunity_document_link_events SET actor='changed' WHERE id=CAST(:id AS UUID)"), {"id": event})
|
|
pg_conn.rollback()
|
|
pg_conn.execute(text("SET clientflow.v2_consumers_active='off'"))
|
|
pg_conn.execute(text("SET clientflow.document_ledger_exported='on'"))
|
|
pg_conn.exec_driver_sql(Path("migrations/007_document_reconciliation_v2_down.sql").read_text())
|
|
assert pg_conn.execute(text("SELECT to_regclass('opportunity_document_links')")).scalar() is None
|