home-automation/tests/test_admin_cli.py

"""Tests for M4-T03 + M4-T07: scripts/admin_cli.py — CLI escape hatch.

Covers:
- reset-password: new password verifies, old password fails.
- unlock: clears specified/all throttle rows; prints count; idempotent.
- disable-totp: clears TOTP state + recovery codes without requiring any credential.
- reissue-totp: writes new secret; prints otpauth URI; secret changes.
- list-admin: prints user table without crashing.
- Non-existent user → friendly error + SystemExit with non-zero code.
- Data-safety: CLI must never reference user-data tables in delete/drop.

DB isolation: each test gets a fresh temporary SQLite database via
``cli_db`` fixture (mirrors the pattern in test_login_throttle.py).
"""

from __future__ import annotations

import subprocess
import sys
from datetime import UTC, datetime
from pathlib import Path

import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker

from app.db import reset_db_caches
from app.config import get_settings
from app.models.auth import AuthUser, RecoveryCode
from app.models.auth_throttle import LoginThrottle
from app.services.auth import hash_password, verify_password
from scripts.admin_cli import (
    build_parser,
    cmd_disable_totp,
    cmd_list_admin,
    cmd_reset_password,
    cmd_reissue_totp,
    cmd_unlock,
)


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


def _make_app_alembic_config(database_url: str) -> Config:
    config = Config("alembic_app.ini")
    config.set_main_option("sqlalchemy.url", database_url)
    return config


@pytest.fixture()
def cli_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    """Fresh migrated SQLite DB wired as the active app DB for the CLI.

    Monkeypatches APP_DATABASE_URL so that ``get_session_local()`` inside
    admin_cli reaches this temp file, not the real data store.
    """
    db_path = tmp_path / "cli_test.db"
    db_url = f"sqlite:///{db_path}"

    monkeypatch.setenv("APP_DATABASE_URL", db_url)
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "original-password")
    monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false")
    monkeypatch.setenv("AUTH_LOGIN_THROTTLE_ENABLED", "false")
    get_settings.cache_clear()
    reset_db_caches()

    command.upgrade(_make_app_alembic_config(db_url), "head")
    reset_db_caches()

    # Bootstrap the admin user manually so the test controls the initial hash.
    engine = create_engine(db_url, connect_args={"check_same_thread": False})
    SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, class_=Session)
    with SessionLocal() as session:
        # The Alembic upgrade does NOT create a user (that is app startup logic).
        # Insert the initial admin user here for tests that need one.
        existing = session.scalar(select(AuthUser).limit(1))
        if existing is None:
            user = AuthUser(
                username="admin",
                password_hash=hash_password("original-password"),
                is_active=True,
                force_password_change=False,
                created_at=datetime.now(UTC),
            )
            session.add(user)
            session.commit()
    engine.dispose()
    reset_db_caches()

    yield db_url

    get_settings.cache_clear()
    reset_db_caches()


@pytest.fixture()
def cli_session(cli_db: str) -> Session:
    """SQLAlchemy session bound to the CLI test DB."""
    engine = create_engine(cli_db, connect_args={"check_same_thread": False})
    SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, class_=Session)
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()
        engine.dispose()


def _add_throttle_rows(session: Session, rows: list[tuple[str, str]]) -> None:
    """Helper: insert (scope, key) throttle rows into the test DB."""
    now = datetime.now(UTC)
    for scope, key in rows:
        session.add(
            LoginThrottle(
                scope=scope,
                key=key,
                failures=5,
                first_failed_at=now,
                last_failed_at=now,
                next_allowed_at=None,
            )
        )
    session.commit()


def _throttle_count(session: Session) -> int:
    return session.query(LoginThrottle).count()


def _enable_totp_for_user(session: Session, username: str, secret: str = "JBSWY3DPEHPK3PXP") -> None:
    """Helper: set TOTP enabled + secret for a user and add some recovery codes."""
    user = session.scalar(select(AuthUser).where(AuthUser.username == username))
    assert user is not None, f"User '{username}' not found in _enable_totp_for_user"
    user.totp_secret = secret
    user.totp_enabled = True
    # Add two fake recovery codes so we can verify they get deleted.
    session.add(RecoveryCode(user_id=user.id, code_hash=hash_password("aaaa-bbbb")))
    session.add(RecoveryCode(user_id=user.id, code_hash=hash_password("cccc-dddd")))
    session.commit()
    session.expire_all()


def _recovery_code_count(session: Session, username: str) -> int:
    user = session.scalar(select(AuthUser).where(AuthUser.username == username))
    assert user is not None
    return session.query(RecoveryCode).filter(RecoveryCode.user_id == user.id).count()


# ---------------------------------------------------------------------------
# reset-password tests
# ---------------------------------------------------------------------------


class TestResetPassword:
    """reset-password sub-command."""

    def test_new_password_verifies(self, cli_db: str, cli_session: Session):
        """After reset, new password passes verify_password; old one fails."""
        parser = build_parser()
        args = parser.parse_args(["reset-password", "admin", "--password", "new-secret-123"])
        cmd_reset_password(args)

        # Re-query from the session (committed by CLI, so flush needed).
        cli_session.expire_all()
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert verify_password("new-secret-123", user.password_hash), (
            "New password should verify against stored hash"
        )
        assert not verify_password("original-password", user.password_hash), (
            "Old password must no longer verify"
        )

    def test_nonexistent_user_exits_nonzero(self, cli_db: str):
        """reset-password for an unknown username must SystemExit with code != 0."""
        parser = build_parser()
        args = parser.parse_args(
            ["reset-password", "ghost-user", "--password", "doesnt-matter"]
        )
        with pytest.raises(SystemExit) as exc_info:
            cmd_reset_password(args)
        assert exc_info.value.code != 0

    def test_empty_password_exits_nonzero(self, cli_db: str):
        """Passing an empty --password string must fail with non-zero exit."""
        parser = build_parser()
        args = parser.parse_args(["reset-password", "admin", "--password", ""])
        with pytest.raises(SystemExit) as exc_info:
            cmd_reset_password(args)
        assert exc_info.value.code != 0

    def test_multiple_resets_are_idempotent(self, cli_db: str, cli_session: Session):
        """Calling reset-password multiple times leaves the last password active."""
        parser = build_parser()

        args1 = parser.parse_args(["reset-password", "admin", "--password", "first-pass"])
        cmd_reset_password(args1)

        args2 = parser.parse_args(["reset-password", "admin", "--password", "second-pass"])
        cmd_reset_password(args2)

        cli_session.expire_all()
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert verify_password("second-pass", user.password_hash)
        assert not verify_password("first-pass", user.password_hash)


# ---------------------------------------------------------------------------
# unlock tests
# ---------------------------------------------------------------------------


class TestUnlock:
    """unlock sub-command."""

    def test_unlock_all_clears_all_rows(self, cli_db: str, cli_session: Session):
        """unlock --all should delete every row in auth_login_throttle."""
        _add_throttle_rows(
            cli_session,
            [("ip", "1.2.3.4"), ("ip", "5.6.7.8"), ("user", "admin")],
        )
        assert _throttle_count(cli_session) == 3

        parser = build_parser()
        args = parser.parse_args(["unlock", "--all"])
        cmd_unlock(args)

        cli_session.expire_all()
        assert _throttle_count(cli_session) == 0

    def test_unlock_ip_clears_only_target_ip(self, cli_db: str, cli_session: Session):
        """unlock --ip should remove only the matching scope='ip' row."""
        _add_throttle_rows(
            cli_session,
            [("ip", "10.0.0.1"), ("ip", "10.0.0.2"), ("user", "admin")],
        )

        parser = build_parser()
        args = parser.parse_args(["unlock", "--ip", "10.0.0.1"])
        cmd_unlock(args)

        cli_session.expire_all()
        assert _throttle_count(cli_session) == 2
        remaining = cli_session.scalars(select(LoginThrottle)).all()
        keys = [(r.scope, r.key) for r in remaining]
        assert ("ip", "10.0.0.1") not in keys
        assert ("ip", "10.0.0.2") in keys
        assert ("user", "admin") in keys

    def test_unlock_username_clears_only_target_user(self, cli_db: str, cli_session: Session):
        """unlock --username should remove only the matching scope='user' row."""
        _add_throttle_rows(
            cli_session,
            [("ip", "1.2.3.4"), ("user", "admin"), ("user", "other")],
        )

        parser = build_parser()
        args = parser.parse_args(["unlock", "--username", "admin"])
        cmd_unlock(args)

        cli_session.expire_all()
        assert _throttle_count(cli_session) == 2
        remaining = cli_session.scalars(select(LoginThrottle)).all()
        keys = [(r.scope, r.key) for r in remaining]
        assert ("user", "admin") not in keys
        assert ("ip", "1.2.3.4") in keys
        assert ("user", "other") in keys

    def test_unlock_all_idempotent(self, cli_db: str, cli_session: Session):
        """Calling unlock --all when no rows exist must not raise."""
        assert _throttle_count(cli_session) == 0

        parser = build_parser()
        args = parser.parse_args(["unlock", "--all"])
        cmd_unlock(args)  # must not raise

        cli_session.expire_all()
        assert _throttle_count(cli_session) == 0

    def test_unlock_ip_idempotent_when_absent(self, cli_db: str, cli_session: Session):
        """Calling unlock --ip for a non-existent IP is a no-op (0 rows deleted)."""
        parser = build_parser()
        args = parser.parse_args(["unlock", "--ip", "99.99.99.99"])
        cmd_unlock(args)  # must not raise; 0 rows deleted is fine

    def test_unlock_username_idempotent_when_absent(self, cli_db: str, cli_session: Session):
        """Calling unlock --username for a non-existent username is a no-op."""
        parser = build_parser()
        args = parser.parse_args(["unlock", "--username", "nobody"])
        cmd_unlock(args)  # must not raise

    def test_unlock_all_prints_row_count(self, cli_db: str, cli_session: Session, capsys):
        """unlock --all must print the number of rows deleted."""
        _add_throttle_rows(cli_session, [("ip", "1.2.3.4"), ("user", "admin")])

        parser = build_parser()
        args = parser.parse_args(["unlock", "--all"])
        cmd_unlock(args)

        captured = capsys.readouterr()
        assert "2" in captured.out

    def test_unlock_does_not_touch_auth_users(self, cli_db: str, cli_session: Session):
        """unlock must not delete or alter auth_users rows."""
        _add_throttle_rows(cli_session, [("ip", "1.2.3.4")])
        user_count_before = cli_session.query(AuthUser).count()

        parser = build_parser()
        args = parser.parse_args(["unlock", "--all"])
        cmd_unlock(args)

        cli_session.expire_all()
        assert cli_session.query(AuthUser).count() == user_count_before


# ---------------------------------------------------------------------------
# list-admin tests
# ---------------------------------------------------------------------------


class TestListAdmin:
    """list-admin sub-command."""

    def test_list_admin_prints_header_and_user(self, cli_db: str, capsys):
        """list-admin must print the admin user row."""
        parser = build_parser()
        args = parser.parse_args(["list-admin"])
        cmd_list_admin(args)

        captured = capsys.readouterr()
        assert "admin" in captured.out

    def test_list_admin_no_users(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys
    ):
        """list-admin on an empty user table should say 'No users found.'"""
        db_path = tmp_path / "empty_test.db"
        db_url = f"sqlite:///{db_path}"

        monkeypatch.setenv("APP_DATABASE_URL", db_url)
        monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
        monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "x")
        monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false")
        get_settings.cache_clear()
        reset_db_caches()
        command.upgrade(_make_app_alembic_config(db_url), "head")
        reset_db_caches()

        # Do NOT bootstrap a user — table is empty.
        parser = build_parser()
        args = parser.parse_args(["list-admin"])
        cmd_list_admin(args)

        captured = capsys.readouterr()
        assert "No users found" in captured.out

        get_settings.cache_clear()
        reset_db_caches()


# ---------------------------------------------------------------------------
# disable-totp tests
# ---------------------------------------------------------------------------


class TestDisableTotp:
    """disable-totp sub-command (M4-T07)."""

    def test_disable_totp_clears_state(self, cli_db: str, cli_session: Session):
        """After disable-totp: totp_enabled=False, totp_secret=None, recovery codes gone."""
        _enable_totp_for_user(cli_session, "admin")

        # Verify preconditions.
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert user.totp_enabled is True
        assert user.totp_secret is not None
        assert _recovery_code_count(cli_session, "admin") == 2

        # Run disable-totp — NO credential supplied (escape hatch semantics).
        parser = build_parser()
        args = parser.parse_args(["disable-totp", "admin"])
        cmd_disable_totp(args)

        # Verify state.
        cli_session.expire_all()
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert user.totp_enabled is False
        assert user.totp_secret is None
        assert _recovery_code_count(cli_session, "admin") == 0

    def test_disable_totp_no_credential_required(self, cli_db: str, cli_session: Session):
        """disable-totp must succeed without any recovery code or password argument."""
        _enable_totp_for_user(cli_session, "admin")

        # The argparse spec for disable-totp takes only username — no credential flags.
        # Parsing must succeed with just the username.
        parser = build_parser()
        args = parser.parse_args(["disable-totp", "admin"])
        # No exception means no credential was needed.
        cmd_disable_totp(args)  # must not raise

    def test_disable_totp_already_disabled_is_noop(self, cli_db: str, cli_session: Session):
        """disable-totp on a user who never had TOTP is a harmless no-op."""
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert user.totp_enabled is False

        parser = build_parser()
        args = parser.parse_args(["disable-totp", "admin"])
        cmd_disable_totp(args)  # must not raise

        cli_session.expire_all()
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert user.totp_enabled is False
        assert user.totp_secret is None

    def test_disable_totp_nonexistent_user_exits_nonzero(self, cli_db: str):
        """disable-totp for an unknown user must SystemExit with code != 0."""
        parser = build_parser()
        args = parser.parse_args(["disable-totp", "ghost-user"])
        with pytest.raises(SystemExit) as exc_info:
            cmd_disable_totp(args)
        assert exc_info.value.code != 0

    def test_disable_totp_prints_confirmation(
        self, cli_db: str, cli_session: Session, capsys
    ):
        """disable-totp must print a confirmation message mentioning the user."""
        _enable_totp_for_user(cli_session, "admin")

        parser = build_parser()
        args = parser.parse_args(["disable-totp", "admin"])
        cmd_disable_totp(args)

        captured = capsys.readouterr()
        assert "admin" in captured.out

    def test_disable_totp_does_not_touch_password(
        self, cli_db: str, cli_session: Session
    ):
        """disable-totp must not alter the user's password_hash."""
        _enable_totp_for_user(cli_session, "admin")

        user_before = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user_before is not None
        pw_hash_before = user_before.password_hash

        parser = build_parser()
        args = parser.parse_args(["disable-totp", "admin"])
        cmd_disable_totp(args)

        cli_session.expire_all()
        user_after = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user_after is not None
        assert user_after.password_hash == pw_hash_before


# ---------------------------------------------------------------------------
# reissue-totp tests
# ---------------------------------------------------------------------------


class TestReissueTotp:
    """reissue-totp sub-command (M4-T07)."""

    def test_reissue_totp_changes_secret(self, cli_db: str, cli_session: Session):
        """After reissue-totp, totp_secret must be a new non-empty value."""
        old_secret = "JBSWY3DPEHPK3PXP"
        _enable_totp_for_user(cli_session, "admin", secret=old_secret)

        parser = build_parser()
        args = parser.parse_args(["reissue-totp", "admin"])
        cmd_reissue_totp(args)

        cli_session.expire_all()
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert user.totp_secret is not None
        assert user.totp_secret != old_secret, "Secret must be regenerated"

    def test_reissue_totp_prints_uri(self, cli_db: str, cli_session: Session, capsys):
        """reissue-totp must print an otpauth:// URI and the secret."""
        _enable_totp_for_user(cli_session, "admin")

        parser = build_parser()
        args = parser.parse_args(["reissue-totp", "admin"])
        cmd_reissue_totp(args)

        captured = capsys.readouterr()
        assert "otpauth://" in captured.out
        assert "Secret" in captured.out

    def test_reissue_totp_totp_enabled_unchanged(self, cli_db: str, cli_session: Session):
        """reissue-totp must leave totp_enabled in its current state (True stays True)."""
        _enable_totp_for_user(cli_session, "admin")

        user_before = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user_before is not None
        assert user_before.totp_enabled is True

        parser = build_parser()
        args = parser.parse_args(["reissue-totp", "admin"])
        cmd_reissue_totp(args)

        cli_session.expire_all()
        user_after = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user_after is not None
        assert user_after.totp_enabled is True, "totp_enabled must remain unchanged"

    def test_reissue_totp_disabled_user_stays_disabled(
        self, cli_db: str, cli_session: Session
    ):
        """reissue-totp on a user with totp_enabled=False leaves it False."""
        # Don't enable TOTP — user is in disabled state.
        parser = build_parser()
        args = parser.parse_args(["reissue-totp", "admin"])
        cmd_reissue_totp(args)

        cli_session.expire_all()
        user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin"))
        assert user is not None
        assert user.totp_enabled is False, "totp_enabled must remain False"
        assert user.totp_secret is not None, "A new secret should have been written"

    def test_reissue_totp_nonexistent_user_exits_nonzero(self, cli_db: str):
        """reissue-totp for an unknown user must SystemExit with code != 0."""
        parser = build_parser()
        args = parser.parse_args(["reissue-totp", "ghost-user"])
        with pytest.raises(SystemExit) as exc_info:
            cmd_reissue_totp(args)
        assert exc_info.value.code != 0


# ---------------------------------------------------------------------------
# Argument-parser edge-cases
# ---------------------------------------------------------------------------


class TestParserEdgeCases:
    """Ensure argparse flags behave as expected."""

    def test_unlock_requires_flag(self):
        """unlock without any flag should exit with a parse error (non-zero)."""
        parser = build_parser()
        with pytest.raises(SystemExit) as exc_info:
            parser.parse_args(["unlock"])
        assert exc_info.value.code != 0

    def test_unlock_ip_and_all_mutually_exclusive(self):
        """--ip and --all together should produce a parse error."""
        parser = build_parser()
        with pytest.raises(SystemExit) as exc_info:
            parser.parse_args(["unlock", "--all", "--ip", "1.2.3.4"])
        assert exc_info.value.code != 0

    def test_unknown_command_exits_nonzero(self):
        """An unknown sub-command must exit non-zero."""
        parser = build_parser()
        with pytest.raises(SystemExit) as exc_info:
            parser.parse_args(["nonexistent-command"])
        assert exc_info.value.code != 0


# ---------------------------------------------------------------------------
# Module-level smoke: python -m scripts.admin_cli --help
# ---------------------------------------------------------------------------


def test_help_flag_exits_zero():
    """python -m scripts.admin_cli --help must exit 0."""
    result = subprocess.run(
        [sys.executable, "-m", "scripts.admin_cli", "--help"],
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0
    assert "reset-password" in result.stdout
    assert "unlock" in result.stdout
    assert "disable-totp" in result.stdout
    assert "reissue-totp" in result.stdout
    assert "list-admin" in result.stdout