"""Integration tests for PostgreSQL role copy and script running.

Run against the local postgres-source / postgres-target docker-compose
services via `make integration_test`. Excluded from the default lint-and-test
run.
"""

import psycopg
import psycopg.rows
import pytest
from psycopg import sql
from src.logic import postgresql


def _connect(credentials):
    """Open an autocommit connection to the given database."""
    return psycopg.connect(
        host=credentials['host'],
        user=credentials['username'],
        password=credentials['password'],
        dbname='postgres',
        autocommit=True,
    )


def _role_row(credentials, rolname):
    """Return selected pg_roles columns for a role, or None if absent."""
    with _connect(credentials) as conn:
        with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
            cur.execute(
                'select rolname, rolcanlogin '
                '  from pg_roles where rolname = %s',
                (rolname,))
            return cur.fetchone()


def _is_member_of(credentials, child, parent):
    """Return True if child is a direct member of parent."""
    with _connect(credentials) as conn:
        with conn.cursor() as cur:
            cur.execute(
                'select 1 from pg_auth_members m '
                '  join pg_roles c on c.oid = m.member '
                '  join pg_roles p on p.oid = m.roleid '
                ' where c.rolname = %s and p.rolname = %s',
                (child, parent))
            return cur.fetchone() is not None


def test_copy_users_replicates_roles(
        seeded_source, target_credentials):
    """copy_users replicates login/group roles, attributes and memberships."""
    created = postgresql.copy_users(seeded_source, target_credentials)

    assert created >= 3

    login = _role_row(target_credentials, 'app_login')
    group = _role_row(target_credentials, 'app_group')
    owner = _role_row(target_credentials, 'app_owner')

    assert login is not None and login['rolcanlogin'] is True
    # Group/NOLOGIN role must be copied (it is a membership parent).
    assert group is not None and group['rolcanlogin'] is False
    assert owner is not None

    # Membership edge is replayed.
    assert _is_member_of(target_credentials, 'app_login', 'app_group')


def test_existing_password_preserved_on_alter(
        seeded_source, target_credentials):
    """ALTER with no supplied password preserves the target's own password."""
    # Pre-create the login role on the target with its own password, so
    # copy_users takes the ALTER path for it.
    with _connect(target_credentials) as conn:
        with conn.cursor() as cur:
            cur.execute(sql.SQL(
                'create role app_login login password {}').format(
                    sql.Literal('target_secret')))

    # No password map: app_login is absent, so its password is left intact.
    postgresql.copy_users(seeded_source, target_credentials)

    login_creds = {
        'host': target_credentials['host'],
        'username': 'app_login',
        'password': 'target_secret',
    }
    # Connecting proves the ALTER did not wipe the existing password.
    with _connect(login_creds) as conn:
        with conn.cursor() as cur:
            cur.execute('select current_user')
            assert cur.fetchone()[0] == 'app_login'


def test_copy_users_applies_password_from_map(
        seeded_source, target_credentials):
    """A new login role gets the password supplied in role_passwords."""
    postgresql.copy_users(
        seeded_source, target_credentials, {'app_login': 'new_dev_pw'})

    login_creds = {
        'host': target_credentials['host'],
        'username': 'app_login',
        'password': 'new_dev_pw',
    }
    # Connecting with the supplied password proves it was applied on CREATE.
    with _connect(login_creds) as conn:
        with conn.cursor() as cur:
            cur.execute('select current_user')
            assert cur.fetchone()[0] == 'app_login'


def test_password_overwritten_on_alter(seeded_source, target_credentials):
    """A supplied password overwrites a pre-existing target password."""
    # Pre-create app_login with a different password so copy_users alters it.
    with _connect(target_credentials) as conn:
        with conn.cursor() as cur:
            cur.execute(sql.SQL(
                'create role app_login login password {}').format(
                    sql.Literal('old_secret')))

    postgresql.copy_users(
        seeded_source, target_credentials, {'app_login': 'new_pw'})

    new_creds = {
        'host': target_credentials['host'],
        'username': 'app_login',
        'password': 'new_pw',
    }
    # The new password works.
    with _connect(new_creds) as conn:
        with conn.cursor() as cur:
            cur.execute('select current_user')
            assert cur.fetchone()[0] == 'app_login'

    # The old password no longer authenticates.
    old_creds = {**new_creds, 'password': 'old_secret'}
    with pytest.raises(psycopg.OperationalError):
        _connect(old_creds)


def test_copy_users_upserts_object_owning_role(
        seeded_source, target_credentials):
    """An existing role that owns objects is altered, not dropped.

    Pre-create app_owner on the target and have it own a table; a drop-based
    implementation would fail here, the upsert must succeed.
    """
    with _connect(target_credentials) as conn:
        with conn.cursor() as cur:
            cur.execute(sql.SQL(
                'create role app_owner login password {}').format(
                    sql.Literal('old_secret')))
            cur.execute('create table owned_by_app (id int)')
            cur.execute('alter table owned_by_app owner to app_owner')

    # Should not raise despite app_owner owning owned_by_app on the target.
    postgresql.copy_users(seeded_source, target_credentials)

    # The owning role was altered (not dropped) and still exists.
    assert _role_row(target_credentials, 'app_owner') is not None


def test_copy_users_is_idempotent(seeded_source, target_credentials):
    """Running copy_users twice succeeds (second run takes the ALTER path)."""
    postgresql.copy_users(seeded_source, target_credentials)
    created = postgresql.copy_users(seeded_source, target_credentials)
    assert created >= 3
    assert _is_member_of(target_credentials, 'app_login', 'app_group')


def test_copy_users_as_non_superuser(
        seeded_source, non_superuser_target_credentials, target_credentials):
    """copy_users works when the target connection is not a superuser.

    Reproduces the RDS restriction locally: a CREATEROLE-only role cannot set
    SUPERUSER/REPLICATION/BYPASSRLS, so emitting those clauses would raise
    "must be superuser to alter superusers". The roles still get copied.
    """
    created = postgresql.copy_users(
        seeded_source, non_superuser_target_credentials)

    assert created >= 3
    # Verify with master creds that the roles landed on the target.
    assert _role_row(target_credentials, 'app_login') is not None
    assert _role_row(target_credentials, 'app_group') is not None
    assert _is_member_of(target_credentials, 'app_login', 'app_group')


def test_run_scripts_executes_sql(target_credentials, tmp_path):
    """run_scripts runs every .sql file in lexicographical order."""
    # The docker-compose target only has the `postgres` database, so the
    # scripts declare it; the parse-and-connect path is still exercised.
    (tmp_path / '01-create.sql').write_text(
        '-- @database: postgres\ncreate table sanitise_test (id int);')
    (tmp_path / '02-insert.sql').write_text(
        '-- @database: postgres\n'
        'insert into sanitise_test (id) values (1), (2);')

    try:
        count = postgresql.run_scripts(target_credentials, str(tmp_path))
        assert count == 2

        with _connect(target_credentials) as conn:
            with conn.cursor() as cur:
                cur.execute('select count(*) from sanitise_test')
                assert cur.fetchone()[0] == 2
    finally:
        with _connect(target_credentials) as conn:
            with conn.cursor() as cur:
                cur.execute('drop table if exists sanitise_test')


def test_run_scripts_targets_named_database(target_credentials, tmp_path):
    """A script's @database directive routes it to that database.

    Creates a second database, points a script at it, and asserts the object
    lands there and not in the default `postgres` database — proving the
    directive (not the connection default) chooses the database.
    """
    with _connect(target_credentials) as conn:
        with conn.cursor() as cur:
            cur.execute('drop database if exists sanitise_target_db')
            cur.execute('create database sanitise_target_db')

    (tmp_path / '01-create.sql').write_text(
        '-- @database: sanitise_target_db\n'
        'create table routed_test (id int);')

    target_db_creds = {**target_credentials, 'dbname': 'sanitise_target_db'}
    try:
        postgresql.run_scripts(target_credentials, str(tmp_path))

        # The table exists in the routed database...
        with psycopg.connect(
                host=target_db_creds['host'],
                user=target_db_creds['username'],
                password=target_db_creds['password'],
                dbname='sanitise_target_db',
                autocommit=True) as conn:
            with conn.cursor() as cur:
                cur.execute("select to_regclass('routed_test')")
                assert cur.fetchone()[0] is not None
        # ...and not in the default postgres database.
        with _connect(target_credentials) as conn:
            with conn.cursor() as cur:
                cur.execute("select to_regclass('routed_test')")
                assert cur.fetchone()[0] is None
    finally:
        with _connect(target_credentials) as conn:
            with conn.cursor() as cur:
                cur.execute('drop database if exists sanitise_target_db')