team-tryouts/app/supporting_scripts/schema_report.py

"""Compare a live database against the models (DB-001).

Why this exists
---------------
`db.create_all()` creates missing tables and never ALTERs an existing one. A
column added to a model months ago is therefore simply absent from any
database that already had the table, and nothing says so: the application
starts, and the first query touching that column fails at runtime. The audit
called the accumulated result "les dérives" and could not measure it, because
measuring it needs the production database.

This script measures it. It is **read-only** — it opens a connection, reads
the catalogue, prints a report and exits. It issues no DDL and no DML, and
takes no locks beyond what reading `information_schema` takes.

It is the prerequisite for everything in the DB wave: `DB-002` asks for an
initial Alembic migration describing the **real** schema rather than the
models', and this is what tells you what the real schema is.

Usage
-----
    # Against whatever DATABASE_URL points at
    python app/supporting_scripts/schema_report.py

    # Against a restored copy, which is the safe way to do it first
    python app/supporting_scripts/schema_report.py \
        --url postgresql://user:pass@host:5432/restored_copy

    # Also look for the seeded admin/password account (SEC-003)
    python app/supporting_scripts/schema_report.py --check-seed-accounts

    # Find Discord identities that must be reconciled before UNIQUE (SEC-012)
    python app/supporting_scripts/schema_report.py --check-discord-identities

Exit codes
----------
    0  the live schema matches the models
    1  drift or requested data risk found — the report says what
    2  could not connect or read the catalogue

Reading the output
------------------
Findings are grouped by what they cost you:

  BLOCKING   the application will fail at runtime — a table or column the
             models use and the database does not have.
  RISK       the database has something the models do not describe. Harmless
             to the running application, but an Alembic autogenerate would
             propose to DROP it, which is how a corrective migration deletes
             a column somebody still needed.
  DIFFERENCE type, nullability, default or constraint disagreements. Each one
             needs a human: some are dialect spelling, some are real.
"""

import argparse
import os
import sys

# Importable as a script from the project root, like the other supporting
# scripts: `python app/supporting_scripts/schema_report.py`.
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))

from sqlalchemy import create_engine, inspect  # noqa: E402
from sqlalchemy.exc import SQLAlchemyError  # noqa: E402

BLOCKING = 'BLOCKING'
RISK = 'RISK'
DIFFERENCE = 'DIFFERENCE'


class Finding:
    """One disagreement between the models and the live database."""

    def __init__(self, severity, table, detail, consequence=''):
        self.severity = severity
        self.table = table
        self.detail = detail
        self.consequence = consequence

    def __str__(self):
        line = f'  [{self.severity:10}] {self.table}: {self.detail}'
        if self.consequence:
            line += f'\n               → {self.consequence}'
        return line

    def __repr__(self):  # pragma: no cover — debugging aid
        return f'<Finding {self.severity} {self.table} {self.detail}>'


def model_metadata():
    """The schema the models describe.

    Imports app.models for its side effect: importing the modules is what
    registers every table on the shared metadata.
    """
    from app.extensions import db
    from app.models import User  # noqa: F401 — registers the whole model package

    return db.metadata


def _type_of(column_type, dialect):
    """A type as this dialect spells it, so the two sides are comparable.

    Comparing `String(200)` with `VARCHAR(200)` as strings would report every
    column as different. Compiling both against the same dialect makes the
    comparison mean something.
    """
    try:
        return column_type.compile(dialect=dialect)
    except Exception:  # noqa: BLE001 — an uncompilable type is still reportable
        return str(column_type)


def compare_tables(metadata, inspector):
    """Tables the models expect against tables the database has."""
    findings = []
    model_tables = set(metadata.tables)
    live_tables = set(inspector.get_table_names())

    for name in sorted(model_tables - live_tables):
        findings.append(
            Finding(
                BLOCKING,
                name,
                'table is missing from the database',
                'every query against this model fails. create_all() would '
                'create it — which is why the absence can survive unnoticed '
                'on a machine where AUTO_CREATE_TABLES is on.',
            )
        )

    for name in sorted(live_tables - model_tables):
        findings.append(
            Finding(
                RISK,
                name,
                'table exists in the database and in no model',
                'an Alembic autogenerate would propose to DROP it. Decide '
                'before running one: it may be a leftover, or it may be the '
                'only copy of something.',
            )
        )

    return findings, sorted(model_tables & live_tables)


def compare_columns(metadata, inspector, table_name, dialect):
    """Column-by-column, for one table."""
    findings = []
    model_columns = {c.name: c for c in metadata.tables[table_name].columns}
    live_columns = {c['name']: c for c in inspector.get_columns(table_name)}

    for name in sorted(set(model_columns) - set(live_columns)):
        column = model_columns[name]
        findings.append(
            Finding(
                BLOCKING,
                table_name,
                f'column "{name}" is in the model and not in the database',
                'this is exactly what create_all() cannot fix: it never '
                'ALTERs. Any query selecting or writing this column fails.'
                + (
                    ''
                    if column.nullable
                    else ' The column is NOT NULL, so the '
                    'corrective migration needs a default or a backfill.'
                ),
            )
        )

    for name in sorted(set(live_columns) - set(model_columns)):
        findings.append(
            Finding(
                RISK,
                table_name,
                f'column "{name}" is in the database and not in any model',
                'an autogenerated migration would propose to DROP it, taking '
                'its data. Check whether something outside the application '
                'reads it before agreeing.',
            )
        )

    for name in sorted(set(model_columns) & set(live_columns)):
        model_column, live_column = model_columns[name], live_columns[name]

        model_type = _type_of(model_column.type, dialect)
        live_type = _type_of(live_column['type'], dialect)
        if model_type != live_type:
            findings.append(
                Finding(
                    DIFFERENCE,
                    table_name,
                    f'column "{name}" type: model says {model_type}, database says {live_type}',
                    'a narrower column in the database silently truncates or '
                    'rejects; a wider one is usually harmless.',
                )
            )

        if bool(model_column.nullable) != bool(live_column.get('nullable', True)):
            findings.append(
                Finding(
                    DIFFERENCE,
                    table_name,
                    f'column "{name}" nullability: model says '
                    f'{"NULL" if model_column.nullable else "NOT NULL"}, database says '
                    f'{"NULL" if live_column.get("nullable", True) else "NOT NULL"}',
                    'a NOT NULL the database does not enforce is a constraint '
                    'the application only believes it has.',
                )
            )

    return findings


def compare_constraints(metadata, inspector, table_name):
    """Unique constraints, indexes and foreign keys.

    Named constraints are compared by the columns they cover rather than by
    name: the same rule declared under two names is the same rule, and
    reporting it as a difference would bury the ones that matter.
    """
    findings = []
    table = metadata.tables[table_name]

    def column_sets(entries, key):
        return {tuple(sorted(entry[key] or [])) for entry in entries}

    model_unique = {
        tuple(sorted(c.name for c in constraint.columns))
        for constraint in table.constraints
        if constraint.__class__.__name__ == 'UniqueConstraint'
    }
    live_unique = column_sets(inspector.get_unique_constraints(table_name), 'column_names')
    for columns in sorted(model_unique - live_unique):
        findings.append(
            Finding(
                DIFFERENCE,
                table_name,
                f'unique constraint on {list(columns)} is declared and absent from the database',
                'the application believes duplicates are impossible here. '
                'They are not, and two concurrent requests will prove it.',
            )
        )

    model_fks = {
        tuple(sorted(fk.parent.name for fk in constraint.elements))
        for constraint in table.foreign_key_constraints
    }
    live_fks = column_sets(inspector.get_foreign_keys(table_name), 'constrained_columns')
    for columns in sorted(model_fks - live_fks):
        findings.append(
            Finding(
                DIFFERENCE,
                table_name,
                f'foreign key on {list(columns)} is declared and absent from the database',
                'orphan rows are possible, and ON DELETE behaviour is not '
                'being enforced by the database at all.',
            )
        )

    model_indexes = {tuple(sorted(c.name for c in index.columns)) for index in table.indexes}
    live_indexes = column_sets(inspector.get_indexes(table_name), 'column_names')
    for columns in sorted(model_indexes - live_indexes):
        findings.append(
            Finding(
                DIFFERENCE,
                table_name,
                f'index on {list(columns)} is declared and absent from the database',
                'correctness is unaffected; the queries that rely on it are '
                'doing sequential scans.',
            )
        )

    return findings


def collect_findings(engine):
    """Every disagreement between the models and this database."""
    metadata = model_metadata()
    inspector = inspect(engine)

    findings, shared_tables = compare_tables(metadata, inspector)
    for table_name in shared_tables:
        findings.extend(compare_columns(metadata, inspector, table_name, engine.dialect))
        findings.extend(compare_constraints(metadata, inspector, table_name))
    return findings


def find_seed_accounts(engine):
    """Accounts matching the credentials clear_db.py used to seed (SEC-003).

    The script was removed from the deployment, but it had already been run:
    the audit could not tell whether an `admin` account with the password
    `password` still exists in production, and that question cannot be
    answered from the repository.

    Returns:
        list[tuple]: (username, role, whether the known password matches).
    """
    from sqlalchemy import text

    from app.extensions import check_password

    with engine.connect() as connection:
        rows = connection.execute(
            text('SELECT username, role, password_hash FROM users WHERE username = :name'),
            {'name': 'admin'},
        ).fetchall()

    results = []
    for username, role, password_hash in rows:
        try:
            matches = check_password(password_hash, 'password')
        except Exception:  # noqa: BLE001 — an unreadable hash is not a match
            matches = False
        results.append((username, role, matches))
    return results


def find_duplicate_discord_identities(engine):
    """Discord snowflakes claimed by more than one account (SEC-012).

    New links are now refused in application code, but existing production
    rows predate that guard. These groups must be reconciled before Alembic
    can add the database-level UNIQUE constraint.

    Returns:
        list[tuple]: (discord_user_id, comma-separated usernames, count).
    """
    from sqlalchemy import text

    with engine.connect() as connection:
        rows = connection.execute(
            text(
                'SELECT discord_user_id, COUNT(*) AS account_count '
                'FROM users '
                "WHERE discord_user_id IS NOT NULL AND discord_user_id <> '' "
                'GROUP BY discord_user_id HAVING COUNT(*) > 1 '
                'ORDER BY discord_user_id'
            )
        ).fetchall()

        duplicates = []
        for discord_user_id, account_count in rows:
            usernames = connection.execute(
                text(
                    'SELECT username FROM users '
                    'WHERE discord_user_id = :discord_user_id ORDER BY username'
                ),
                {'discord_user_id': discord_user_id},
            ).scalars()
            duplicates.append((discord_user_id, ', '.join(usernames), account_count))
    return duplicates


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.split('\n')[0])
    parser.add_argument(
        '--url',
        default=os.getenv('DATABASE_URL'),
        help='Database URL. Defaults to DATABASE_URL. Point it at a restored copy the first time.',
    )
    parser.add_argument(
        '--check-seed-accounts',
        action='store_true',
        help='Also look for the admin/password account seeded by clear_db.py (SEC-003).',
    )
    parser.add_argument(
        '--check-discord-identities',
        action='store_true',
        help='Find duplicate Discord IDs that block the SEC-012 UNIQUE constraint.',
    )
    args = parser.parse_args(argv)

    if not args.url:
        print('No database URL. Pass --url or set DATABASE_URL.', file=sys.stderr)
        return 2

    from app.app import normalise_database_url

    try:
        engine = create_engine(normalise_database_url(args.url))
        findings = collect_findings(engine)
    except SQLAlchemyError as exc:
        print(f'Could not read the schema: {exc}', file=sys.stderr)
        return 2

    print('=' * 78)
    print('Schema report — models vs live database (DB-001)')
    print('=' * 78)

    if not findings:
        print('\nNo drift. The live schema matches the models.')
    for severity in (BLOCKING, RISK, DIFFERENCE):
        group = [f for f in findings if f.severity == severity]
        if not group:
            continue
        print(f'\n{severity} — {len(group)} finding(s)')
        for finding in group:
            print(finding)

    if args.check_seed_accounts:
        print('\n' + '=' * 78)
        print('Seeded accounts (SEC-003)')
        print('=' * 78)
        try:
            accounts = find_seed_accounts(engine)
        except SQLAlchemyError as exc:
            print(f'Could not check: {exc}')
        else:
            if not accounts:
                print('No account named "admin".')
            for username, role, matches in accounts:
                verdict = (
                    'PASSWORD IS STILL "password" — change it now'
                    if matches
                    else 'password has been changed'
                )
                print(f'  {username} ({role}): {verdict}')

    duplicate_discord_identities = []
    if args.check_discord_identities:
        print('\n' + '=' * 78)
        print('Duplicate Discord identities (SEC-012)')
        print('=' * 78)
        try:
            duplicate_discord_identities = find_duplicate_discord_identities(engine)
        except SQLAlchemyError as exc:
            print(f'Could not check: {exc}')
        else:
            if not duplicate_discord_identities:
                print('No Discord identity is shared by multiple accounts.')
            for discord_user_id, usernames, account_count in duplicate_discord_identities:
                print(f'  {discord_user_id}: {account_count} accounts ({usernames})')

    blocking = sum(1 for f in findings if f.severity == BLOCKING)
    print(f'\n{len(findings)} finding(s), {blocking} blocking.')
    return 1 if findings or duplicate_discord_identities else 0


if __name__ == '__main__':  # pragma: no cover
    sys.exit(main())