diff --git a/pyproject.toml b/pyproject.toml index 673a8a6..5882411 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ tests = [ "flask-marshmallow", "marshmallow-sqlalchemy", "psycopg2-binary", + "pytest-cov" ] [project.urls] diff --git a/src/utils_flask_sqla/referential.py b/src/utils_flask_sqla/referential.py new file mode 100644 index 0000000..1dabb45 --- /dev/null +++ b/src/utils_flask_sqla/referential.py @@ -0,0 +1,99 @@ +import csv +import os + +from sqlalchemy import ( + inspect as sa_inspect, + func, + exists, + select, + table as sa_table, + column as sa_column, +) + +CSV_FIELDNAMES = [ + "ref_table", + "table_name", + "schema", + "fk_column", + "fk_value", + "nb_affected_lines", +] + + +def get_referencing_tables(table_name, db, schema="", exclude_tables=[]): + """Trouve toutes les tables qui ont des FK pointant vers table_name, + en excluant les tables listées dans exclude_tables (même schéma).""" + exclude_tables = set(exclude_tables or []) + inspector = sa_inspect(db.engine) + referencing_tables = [] + + for other_schema in inspector.get_schema_names(): + if other_schema in ("pg_catalog", "information_schema", "pg_toast"): + continue + try: + for other_table in inspector.get_table_names(schema=other_schema): + if other_table != table_name and other_table in exclude_tables: + continue + for fk in inspector.get_foreign_keys(other_table, schema=other_schema): + if fk["referred_table"] == f"{schema}.{table_name}": + referencing_tables.append( + { + "schema": other_schema, + "table": other_table, + "fk_column": fk["constrained_columns"][0], + "ref_column": fk["referred_columns"][0], + } + ) + except Exception as e: + print(f"Impossible d'inspecter le schéma {other_schema}: {e}") + + return referencing_tables + + +def collect_orphan_rows(ref_table, new_ref_table, pk_col, db, schema="", exclude_tables=None): + """ + Retourne la liste des lignes orphelines pour une table du référentiel : + valeurs de pk_col présentes dans les tables référençantes mais absentes de new_ref_table. + """ + referencing = get_referencing_tables( + ref_table, db, schema=schema, exclude_tables=exclude_tables + ) + rows = [] + for ref in referencing: + fk_col = ref["fk_column"] + src = sa_table(ref["table"], sa_column(fk_col), schema=ref["schema"]) + ref_t = sa_table(new_ref_table, sa_column(pk_col), schema=schema) + subq = select(1).select_from(ref_t).where(ref_t.c[pk_col] == src.c[fk_col]) + stmt = ( + select(src.c[fk_col], func.count().label("nb_lines")) + .where(src.c[fk_col].isnot(None)) + .where(~exists(subq)) + .group_by(src.c[fk_col]) + .order_by(src.c[fk_col]) + ) + for fk_value, nb_lines in db.session.execute(stmt).fetchall(): + rows.append( + { + "ref_table": ref_table, + "table_name": ref["table"], + "schema": ref["schema"], + "fk_column": fk_col, + "fk_value": fk_value, + "nb_affected_lines": nb_lines, + } + ) + return rows + + +def export_orphans_to_csv(rows, output_path): + """Écrit la liste de lignes orphelines dans un CSV.""" + if not rows: + return 0 + dirname = os.path.dirname(output_path) + if dirname: + os.makedirs(dirname, exist_ok=True) + with open(output_path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=CSV_FIELDNAMES) + writer.writeheader() + writer.writerows(rows) + return len(rows) diff --git a/src/utils_flask_sqla/tests/psql_tests/__init__.py b/src/utils_flask_sqla/tests/psql_tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/utils_flask_sqla/tests/psql_tests/conftest.py b/src/utils_flask_sqla/tests/psql_tests/conftest.py new file mode 100644 index 0000000..9e6b846 --- /dev/null +++ b/src/utils_flask_sqla/tests/psql_tests/conftest.py @@ -0,0 +1,88 @@ +import importlib.util +import os +from pathlib import Path + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.exc import OperationalError + +from utils_flask_sqla.revision import alter_table_with_dependent_views + +# --------------------------------------------------------------------------- +# Integration tests against a real PostgreSQL database. +# +# These exercise the actual `pg_capture_dependent_views` / `pg_drop_dependent_views` +# / `pg_recreate_dependent_views` SQL functions (created by migration revision +# 1d09a9b67970) against a real dependency chain: a table, a materialized view +# built on that table, and a plain view built on that materialized view. +# +# They require a running PostgreSQL server. Point TEST_PG_DATABASE_URI at it, +# e.g.: +# docker run --rm -d -p 5432:5432 -e POSTGRES_USER=geonatadmin \ +# -e POSTGRES_PASSWORD=geonatadmin -e POSTGRES_DB=geonature2db \ +# postgis/postgis:15-3.4 +# Tests are skipped automatically if no server is reachable. +# --------------------------------------------------------------------------- + +PG_URI = os.environ.get( + "TEST_PG_DATABASE_URI", + "postgresql+psycopg2://geonatadmin:geonatadmin@localhost:5432/geonature2db", +) + +MIGRATION_FILE = ( + Path(__file__).resolve().parent.parent.parent + / "migrations" + / "versions" + / "1d09a9b67970_add_functions_to_fetch_dependent_view_.py" +) + + +def _load_migration_module(): + spec = importlib.util.spec_from_file_location( + "utils_flask_sqla_test_dependent_views_migration", MIGRATION_FILE + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture(scope="module") +def pg_engine(): + from alembic.operations import Operations + from alembic.runtime.migration import MigrationContext + + engine = create_engine(PG_URI) + try: + with engine.connect() as conn: + conn.execute(text("SELECT 1")) + except OperationalError as exc: + engine.dispose() + pytest.skip(f"PostgreSQL test database not reachable at {PG_URI}: {exc}") + + migration = _load_migration_module() + with engine.begin() as conn: + with Operations.context(MigrationContext.configure(conn)): + migration.upgrade() + + yield engine + + with engine.begin() as conn: + with Operations.context(MigrationContext.configure(conn)): + migration.downgrade() + engine.dispose() + + +@pytest.fixture +def pg_conn(pg_engine): + """A connection wrapping the whole test in one rolled-back transaction. + + PostgreSQL DDL is transactional, so creating the schema/table/views inside + this transaction and rolling it back at teardown leaves no trace, without + needing per-test unique names. + """ + with pg_engine.connect() as conn: + trans = conn.begin() + try: + yield conn + finally: + trans.rollback() diff --git a/src/utils_flask_sqla/tests/psql_tests/test_referential.py b/src/utils_flask_sqla/tests/psql_tests/test_referential.py new file mode 100644 index 0000000..be1a7c0 --- /dev/null +++ b/src/utils_flask_sqla/tests/psql_tests/test_referential.py @@ -0,0 +1,112 @@ +import pytest + + +from tempfile import NamedTemporaryFile + +import pytest +from flask import Flask +from flask_sqlalchemy import SQLAlchemy +from sqlalchemy import ForeignKey, Integer, Table, Unicode, insert +from utils_flask_sqla.referential import collect_orphan_rows, get_referencing_tables +from utils_flask_sqla.tests.psql_tests.conftest import PG_URI +from utils_flask_sqla.tests.utils import TestSession + +db = SQLAlchemy() + + +table_1 = Table( + "public.table1", + db.metadata, + db.Column("pk", Integer, primary_key=True), + db.Column("name", Unicode), +) +table_2 = Table( + "public.table2", + db.metadata, + db.Column("pk", Integer, primary_key=True), + db.Column("fk", Integer, ForeignKey(table_1.c.pk)), +) +table_1_new = Table( + "public.table1_new", + db.metadata, + db.Column("pk", Integer, primary_key=True), +) + + +@pytest.fixture(scope="session") +def _app(): + app = Flask(__name__) + app.config["SQLALCHEMY_DATABASE_URI"] = PG_URI + db.init_app(app) + with app.app_context(): + db.create_all() + yield + + +@pytest.fixture(scope="session") +def _session(_app): + db.session.session_factory.class_ = TestSession + db.session.remove() + return db.session + + +@pytest.fixture(scope="session") +def app(_app, _session): + pass + + +@pytest.fixture(scope="session") +def data(app): + with db.session.begin_nested(): + db.session.execute( + insert(table_1).values( + [ + {"pk": 1, "name": "ligne1"}, + {"pk": 2, "name": "ligne2"}, + {"pk": 3, "name": "ligne3"}, + ] + ) + ) + db.session.execute( + insert(table_2).values( + [ + {"pk": 1, "fk": 1}, + {"pk": 2, "fk": 2}, + {"pk": 3, "fk": 3}, + ] + ) + ) + + +@pytest.mark.usefixtures("app") +class TestReferential: + + def test_get_get_referencing_tables(data): + referenced_tables = get_referencing_tables("table1", db, "public") + assert any([table_def["table"] == "public.table2" for table_def in referenced_tables]) + + def test_collect_orphan_rows_no_orphans(self, data): + rows = collect_orphan_rows("table1", "public.table1", "pk", db, schema="public") + assert rows == [] + + def test_collect_orphan_rows_detects_missing_rows(self, data): + db.session.execute(insert(table_1_new).values([{"pk": 1}, {"pk": 2}])) + + rows = collect_orphan_rows("table1", "public.table1_new", "pk", db, schema="public") + + assert rows == [ + { + "ref_table": "table1", + "table_name": "public.table2", + "schema": "public", + "fk_column": "fk", + "fk_value": 3, + "nb_affected_lines": 1, + } + ] + + def test_collect_orphan_rows_excludes_tables(self, data): + rows = collect_orphan_rows( + "table1", "public.table1", "pk", db, schema="public", exclude_tables=["table2"] + ) + assert rows == [] diff --git a/src/utils_flask_sqla/tests/test_revision.py b/src/utils_flask_sqla/tests/psql_tests/test_revision.py similarity index 64% rename from src/utils_flask_sqla/tests/test_revision.py rename to src/utils_flask_sqla/tests/psql_tests/test_revision.py index 91b5b53..407616b 100644 --- a/src/utils_flask_sqla/tests/test_revision.py +++ b/src/utils_flask_sqla/tests/psql_tests/test_revision.py @@ -8,86 +8,6 @@ from utils_flask_sqla.revision import alter_table_with_dependent_views -# --------------------------------------------------------------------------- -# Integration tests against a real PostgreSQL database. -# -# These exercise the actual `pg_capture_dependent_views` / `pg_drop_dependent_views` -# / `pg_recreate_dependent_views` SQL functions (created by migration revision -# 1d09a9b67970) against a real dependency chain: a table, a materialized view -# built on that table, and a plain view built on that materialized view. -# -# They require a running PostgreSQL server. Point TEST_PG_DATABASE_URI at it, -# e.g.: -# docker run --rm -p 5432:5432 -e POSTGRES_USER=geonatadmin \ -# -e POSTGRES_PASSWORD=geonatadmin -e POSTGRES_DB=geonature2db \ -# postgis/postgis:15-3.4 -# Tests are skipped automatically if no server is reachable. -# --------------------------------------------------------------------------- - -PG_URI = os.environ.get( - "TEST_PG_DATABASE_URI", - "postgresql+psycopg2://geonatadmin:geonatadmin@localhost:5432/geonature2db", -) - -MIGRATION_FILE = ( - Path(__file__).resolve().parents[1] - / "migrations" - / "versions" - / "1d09a9b67970_add_functions_to_fetch_dependent_view_.py" -) - - -def _load_migration_module(): - spec = importlib.util.spec_from_file_location( - "utils_flask_sqla_test_dependent_views_migration", MIGRATION_FILE - ) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module - - -@pytest.fixture(scope="module") -def pg_engine(): - from alembic.operations import Operations - from alembic.runtime.migration import MigrationContext - - engine = create_engine(PG_URI) - try: - with engine.connect() as conn: - conn.execute(text("SELECT 1")) - except OperationalError as exc: - engine.dispose() - pytest.skip(f"PostgreSQL test database not reachable at {PG_URI}: {exc}") - - migration = _load_migration_module() - with engine.begin() as conn: - with Operations.context(MigrationContext.configure(conn)): - migration.upgrade() - - yield engine - - with engine.begin() as conn: - with Operations.context(MigrationContext.configure(conn)): - migration.downgrade() - engine.dispose() - - -@pytest.fixture -def pg_conn(pg_engine): - """A connection wrapping the whole test in one rolled-back transaction. - - PostgreSQL DDL is transactional, so creating the schema/table/views inside - this transaction and rolling it back at teardown leaves no trace, without - needing per-test unique names. - """ - with pg_engine.connect() as conn: - trans = conn.begin() - try: - yield conn - finally: - trans.rollback() - - SCHEMA = "test_revision"