vnctalk-prosody/tests/test_06_postgres.py

"""PostgreSQL database verification tests.

Prosody's mod_storage_sql keeps ALL stores in just two tables:
  - prosody         (keyval / map stores):
        host, user, store, key, type, value          -- indexed by (host,user,store,key)
  - prosodyarchive  (archive stores):
        sort_id, host, user, store, key, "when", "with", type, value

Individual stores (roster, vcard, archive, muc_log, kick, activity, ...) are
NOT separate tables — they are rows distinguished by the `store` column.
See https://prosody.im/doc/developers/modules/mod_storage_sql

Requires: --pg-host, --pg-user, --pg-password (and optionally --pg-db, --pg-port)
"""
import pytest


async def _store_has_rows(conn, table, store):
    """Return True if the given store has at least one row in the table."""
    row = await conn.fetchrow(
        f'SELECT 1 FROM {table} WHERE store = $1 LIMIT 1', store
    )
    return row is not None


@pytest.mark.asyncio
class TestPostgresSchema:
    """Verify the Prosody PostgreSQL schema is present."""

    async def test_connection(self, pg_connection):
        """Basic connectivity to PostgreSQL."""
        row = await pg_connection.fetchrow("SELECT 1 AS one")
        assert row["one"] == 1

    async def test_prosody_tables_exist(self, pg_connection):
        """The two mod_storage_sql tables must exist."""
        tables = await pg_connection.fetch(
            """
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
            """
        )
        table_names = {t["table_name"] for t in tables}
        assert "prosody" in table_names, f"'prosody' keyval table missing. Got: {sorted(table_names)}"
        assert "prosodyarchive" in table_names, f"'prosodyarchive' table missing. Got: {sorted(table_names)}"

    async def test_archive_table_exists(self, pg_connection):
        """The archive table (prosodyarchive) must exist and hold the MAM 'archive' store."""
        row = await pg_connection.fetchrow(
            "SELECT to_regclass('public.prosodyarchive') AS t"
        )
        assert row["t"] is not None, "prosodyarchive table not found"
        assert await _store_has_rows(pg_connection, "prosodyarchive", "archive"), (
            "No rows with store='archive' — MAM archive is empty or storage misconfigured"
        )

    async def test_muc_log_store_exists(self, pg_connection):
        """MUC logging is the 'muc_log' store inside prosodyarchive (not a separate table)."""
        assert await _store_has_rows(pg_connection, "prosodyarchive", "muc_log"), (
            "No rows with store='muc_log' — mod_muc_mam not logging, or muc_log_all_rooms off"
        )


@pytest.mark.asyncio
class TestPostgresMAM:
    """Verify MAM data is written to PostgreSQL correctly."""

    async def test_archive_has_required_columns(self, pg_connection):
        """prosodyarchive must have the columns Prosody's archive storage expects."""
        cols = await pg_connection.fetch(
            """
            SELECT column_name
            FROM information_schema.columns
            WHERE table_schema = 'public' AND table_name = 'prosodyarchive'
            ORDER BY ordinal_position
            """
        )
        col_names = [c["column_name"] for c in cols]
        required = ["host", "user", "store", "key", "when", "with", "value"]
        for r in required:
            assert r in col_names, f"prosodyarchive missing column '{r}'. Got: {col_names}"

    async def test_muc_log_stored_under_conference_host(self, pg_connection, xmpp_config):
        """MUC archives belong to the MUC component host, with the room node as 'user'."""
        if not await _store_has_rows(pg_connection, "prosodyarchive", "muc_log"):
            pytest.skip("No muc_log rows to inspect")
        row = await pg_connection.fetchrow(
            """
            SELECT host, "user" FROM prosodyarchive WHERE store = 'muc_log' LIMIT 1
            """
        )
        muc_domain = xmpp_config.get("muc_domain")
        if muc_domain:
            assert row["host"] == muc_domain, (
                f"muc_log row host '{row['host']}' is not the MUC component '{muc_domain}'"
            )
        else:
            assert "conference" in row["host"], f"unexpected muc_log host: {row['host']}"
        assert row["user"], "muc_log row has empty 'user' (room node) column"


@pytest.mark.asyncio
class TestPostgresDataIntegrity:
    """Cross-check XMPP state with PostgreSQL state."""

    async def test_vcard_store_exists(self, pg_connection):
        """vCards are the 'vcard' store in the prosody keyval table (needed for avatar fallback)."""
        assert await _store_has_rows(pg_connection, "prosody", "vcard"), (
            "No rows with store='vcard' — no vCards have been stored"
        )


@pytest.mark.asyncio
class TestPostgresStoreUserInversion:
    """Priority 3 from AUDIT: REST-injected messages are archived under the sender's user row."""

    async def test_rest_message_archived_under_sender(self, pg_connection, xmpp_client, xmpp_config, rest_injector):
        """After REST injection, the archive row's 'user' column must be the sender, not the recipient."""
        import asyncio
        import uuid

        msg_id = f"rest-pg-{uuid.uuid4().hex[:12]}"
        from_jid = xmpp_client.boundjid.bare
        to_jid = f"foreign-{uuid.uuid4().hex[:8]}@foreign.example.com"
        body = f"PostgreSQL store_user test {msg_id}"

        status, _ = await rest_injector.inject_message(msg_id, from_jid, to_jid, body)
        assert status in (201, 422), f"REST injection failed with {status}"

        # Wait for the async DB write
        await asyncio.sleep(2)

        # 'key' is the generated archive UID, not the message id — the message id
        # and body live inside the serialized stanza in 'value', so match on that.
        sender_localpart = from_jid.split("@")[0]
        rows = await pg_connection.fetch(
            """
            SELECT "user", "with", value
            FROM prosodyarchive
            WHERE store = 'archive' AND value LIKE $1
            """,
            f"%{msg_id}%",
        )

        assert len(rows) > 0, f"No archive row found for REST-injected message {msg_id}"
        row = rows[0]
        assert row["user"] == sender_localpart, (
            f"Archive row owned by wrong user: expected '{sender_localpart}' (sender), "
            f"got '{row['user']}'. mod_mam.lua store_user inversion patch is missing."
        )
        assert row["with"] == to_jid, (
            f"Archive 'with' column mismatch: expected '{to_jid}', got '{row['with']}'"
        )


@pytest.mark.asyncio
class TestPostgresKickStore:
    """Verify mod_vnc_track_kicks writes to the kick archive store."""

    async def test_kick_store_exists(self, pg_connection):
        """Kicks are the 'kick' archive store (mod_vnc_track_kicks: open_store('kick','archive'))."""
        assert await _store_has_rows(pg_connection, "prosodyarchive", "kick"), (
            "No rows with store='kick' — mod_vnc_track_kicks not loaded or no kicks recorded"
        )

    async def test_kick_writes_fresh_row(self, pg_connection, xmpp_client, second_client, xmpp_config):
        """A real vnc-muc-kick must append a fresh row to the kick archive.

        register.lib.lua fires `vnc-muc-kick` from handle_unregister_iq (the
        xmpp:vnctalk:unregister IQ), which mod_vnc_track_kicks persists as a
        kick-archive row keyed by room node with `with` = the kicked user's
        bare JID. We snapshot max(sort_id), trigger the IQ, and assert a new
        row landed with a higher sort_id.
        """
        import asyncio
        import uuid
        import slixmpp
        import xml.etree.ElementTree as ET

        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")
        if xmpp_client.boundjid.bare == second_client.boundjid.bare:
            pytest.xfail("Need --xmpp-jid2 with a distinct account")

        room_jid = f"kickpg_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick1 = "owner"
        member_bare = second_client.boundjid.bare
        room_node = room_jid.split("@")[0]

        await xmpp_client.join_muc(room_jid, nick1)
        await xmpp_client.configure_muc(room_jid)
        await asyncio.sleep(0.5)
        await xmpp_client.set_muc_affiliation(room_jid, member_bare, "member")
        await asyncio.sleep(0.5)

        before = await pg_connection.fetchval(
            'SELECT COALESCE(MAX(sort_id), 0) FROM prosodyarchive WHERE store = $1', "kick"
        )

        try:
            iq = second_client.make_iq_set(ito=room_jid)
            ET.SubElement(iq.xml, "{xmpp:vnctalk:unregister}query")
            # handle_unregister_iq fires vnc-muc-kick but sends no IQ reply,
            # so the send times out by design — the kick row is still written.
            try:
                await iq.send(timeout=5)
            except slixmpp.exceptions.IqTimeout:
                pass

            # Allow the async archive write to land.
            row = None
            for _ in range(20):
                await asyncio.sleep(0.25)
                row = await pg_connection.fetchrow(
                    """
                    SELECT sort_id, host, "user", "with", value
                    FROM prosodyarchive
                    WHERE store = 'kick' AND sort_id > $1
                    ORDER BY sort_id DESC LIMIT 1
                    """,
                    before,
                )
                if row is not None:
                    break

            assert row is not None, (
                "No new kick row appended after unregister IQ — "
                "mod_vnc_track_kicks vnc-muc-kick handler did not persist"
            )
            assert row["sort_id"] > before
            assert row["with"] == member_bare, (
                f"kick row 'with' is '{row['with']}', expected '{member_bare}'"
            )
            assert row["host"] == muc_domain, (
                f"kick row host '{row['host']}' is not the MUC component '{muc_domain}'"
            )
            assert row["user"] == room_node, (
                f"kick row 'user' is '{row['user']}', expected room node '{room_node}'"
            )
        finally:
            await xmpp_client.leave_muc(room_jid, nick1)
            await xmpp_client.destroy_muc(room_jid)


@pytest.mark.asyncio
class TestPostgresActivityStore:
    """Verify mod_vnc_lastactivity writes to the activity map store."""

    async def test_activity_store_exists(self, pg_connection):
        """Activity is the 'activity' map store in the prosody keyval table
        (mod_vnc_lastactivity: open_store('activity','map'))."""
        assert await _store_has_rows(pg_connection, "prosody", "activity"), (
            "No rows with store='activity' — mod_vnc_lastactivity not loaded or no activity recorded"
        )