import sqlite3

import pytest

from src.passkey_store import PasskeyStore
from src.session_store import SessionStore, SessionStoreError


def test_global_access_revocation_rolls_back_every_credential_and_session_change(tmp_path):
    database = tmp_path / "sessions.sqlite3"
    sessions = SessionStore(database, clock=lambda: 1_000.0)
    sessions.activate(
        "phone-session",
        2_000,
        management_id="phone-management-id",
        device_label="Phone",
    )
    sessions.mint_step_up(
        "phone-session",
        action="revoke_all_sessions",
        target="all",
        ttl_seconds=90,
    )
    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
    passkeys.register(
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=0,
        device_label="Phone",
        management_id="phone-management-id",
    )
    passkeys.issue_challenge(
        b"pending-challenge",
        session_id="phone-session",
        purpose="authentication",
        action="sign_in",
        target="dashboard",
    )
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TRIGGER block_session_delete BEFORE DELETE ON active_sessions "
            "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
        )

    with pytest.raises(SessionStoreError):
        passkeys.revoke_all_access()

    with sqlite3.connect(database) as connection:
        assert connection.execute("SELECT COUNT(*) FROM passkey_credentials").fetchone() == (1,)
        assert connection.execute("SELECT COUNT(*) FROM passkey_challenges").fetchone() == (1,)
        assert connection.execute("SELECT COUNT(*) FROM step_up_grants").fetchone() == (1,)
        assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (1,)


def test_device_access_revocation_rolls_back_target_and_preserves_other_devices(tmp_path):
    database = tmp_path / "sessions.sqlite3"
    sessions = SessionStore(database, clock=lambda: 1_000.0)
    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
    for label in ("Phone", "Laptop"):
        slug = label.lower()
        sessions.activate(
            f"{slug}-session",
            2_000,
            management_id=f"{slug}-management-id",
            device_label=label,
        )
        sessions.mint_step_up(
            f"{slug}-session",
            action="close_issue",
            target="stackchain/dashboard#1",
            ttl_seconds=90,
        )
        passkeys.register(
            credential_id=f"{slug}-credential".encode(),
            public_key=f"{slug}-public-key".encode(),
            sign_count=0,
            device_label=label,
            management_id=f"{slug}-management-id",
        )
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TRIGGER block_phone_delete BEFORE DELETE ON active_sessions "
            "WHEN OLD.management_id = 'phone-management-id' "
            "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
        )

    with pytest.raises(SessionStoreError):
        passkeys.revoke_device_access("phone-management-id")

    with sqlite3.connect(database) as connection:
        assert connection.execute(
            "SELECT management_id FROM active_sessions ORDER BY management_id"
        ).fetchall() == [("laptop-management-id",), ("phone-management-id",)]
        assert connection.execute(
            "SELECT management_id FROM passkey_credentials ORDER BY management_id"
        ).fetchall() == [("laptop-management-id",), ("phone-management-id",)]
        assert connection.execute("SELECT COUNT(*) FROM step_up_grants").fetchone() == (2,)