tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity

M4-T03: add admin_cli escape hatch (reset-password, unlock, list-admin)

Browse Source
This commit is contained in:
tliu93
2026-06-21 21:28:41 +02:00
parent e480a84a44
commit 5481fbc99c
3 changed files with 641 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_admin_cli.py
+390
View File
@@ -0,0 +1,390 @@
"""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