home-automation/tests/test_admin_cli.py

"""Tests for M4-T03: 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.
- 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
from app.models.auth_throttle import LoginThrottle
from app.services.auth import hash_password, verify_password
from scripts.admin_cli import (
    build_parser,
    cmd_list_admin,
    cmd_reset_password,
    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()


# ---------------------------------------------------------------------------
# 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()


# ---------------------------------------------------------------------------
# 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 "list-admin" in result.stdout