1c5b0cbc18
Inlocuieste crearea conturilor prin INSERT SQL manual cu un tool admin dedicat, simetric cu tools/apikey.py. Fundatia Etapei 3 (3.2/3.3). - app/accounts.py: create_account/set_active/list_accounts (helper pur, partajat CLI + viitor flux web 3.3). Normalizeaza CUI (trim+upper), prinde IntegrityError -> ValueError cu cauza+fix. - accounts.active (lifecycle cont) + index unic partial ux_accounts_cui (unicitate la nivel de index, fara fereastra de coliziune). Migrare idempotenta in _migrate. - tools/account.py: create (--name/--cui/--inactive/--with-key atomic), list [--pending], activate/deactivate --account N. Erori -> exit 2. - 20 teste noi (12 helper + 8 CLI); suita 299 passed. active e inert pana la gate-ul worker din 3.3 (documentat). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
125 lines
3.7 KiB
Python
125 lines
3.7 KiB
Python
"""Teste US-002 (PRD 3.1): CLI tools/account.py (create/list/activate/deactivate)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def env(monkeypatch):
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_tools_account.db"))
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
yield
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def _run(argv):
|
|
from tools.account import main
|
|
return main(argv)
|
|
|
|
|
|
def test_create_afiseaza_id(env, capsys):
|
|
rc = _run(["create", "--name", "Service X"])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "id=2" in out
|
|
assert "activ=da" in out
|
|
|
|
|
|
def test_create_inactive_in_asteptare(env, capsys):
|
|
rc = _run(["create", "--name", "Service X", "--inactive"])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "activ=nu" in out
|
|
|
|
|
|
def test_create_with_key_emite_cheie(env, capsys):
|
|
rc = _run(["create", "--name", "Service X", "--with-key"])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "rfak_" in out
|
|
# cheia exista in DB pentru contul nou
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
n = conn.execute(
|
|
"SELECT COUNT(*) AS n FROM api_keys k JOIN accounts a ON a.id=k.account_id "
|
|
"WHERE a.name='Service X'"
|
|
).fetchone()["n"]
|
|
finally:
|
|
conn.close()
|
|
assert n == 1
|
|
|
|
|
|
def test_create_cui_duplicat_exit_2(env, capsys):
|
|
assert _run(["create", "--name", "Service A", "--cui", "RO123"]) == 0
|
|
rc = _run(["create", "--name", "Service B", "--cui", "RO123"])
|
|
err = capsys.readouterr().err
|
|
assert rc == 2
|
|
assert "RO123" in err
|
|
|
|
|
|
def test_with_key_atomic_pe_cui_duplicat(env, capsys):
|
|
# cont initial care ocupa CUI
|
|
assert _run(["create", "--name", "Service A", "--cui", "RO123"]) == 0
|
|
capsys.readouterr()
|
|
# --with-key pe CUI duplicat: rollback -> niciun cont B, nicio cheie orfana
|
|
rc = _run(["create", "--name", "Service B", "--cui", "RO123", "--with-key"])
|
|
assert rc == 2
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
assert conn.execute("SELECT COUNT(*) AS n FROM accounts WHERE name='Service B'").fetchone()["n"] == 0
|
|
# o singura cheie n-a fost emisa (doar contul A nu are cheie)
|
|
assert conn.execute("SELECT COUNT(*) AS n FROM api_keys").fetchone()["n"] == 0
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_activate_comuta_starea(env, capsys):
|
|
_run(["create", "--name", "Service X", "--inactive"])
|
|
capsys.readouterr()
|
|
assert _run(["deactivate", "--account", "2"]) == 0
|
|
assert _run(["activate", "--account", "2"]) == 0
|
|
capsys.readouterr()
|
|
rc = _run(["list"])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "Service X" in out
|
|
# contul 2 apare activ
|
|
line = [ln for ln in out.splitlines() if "Service X" in ln][0]
|
|
assert "da" in line
|
|
|
|
|
|
def test_activate_inexistent_exit_2(env, capsys):
|
|
rc = _run(["activate", "--account", "9999"])
|
|
err = capsys.readouterr().err
|
|
assert rc == 2
|
|
assert "inexistent" in err
|
|
|
|
|
|
def test_list_afiseaza_activ(env, capsys):
|
|
_run(["create", "--name", "Service X"])
|
|
capsys.readouterr()
|
|
rc = _run(["list"])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "Service X" in out
|
|
assert "activ" in out # antet tabel
|
|
|
|
|
|
def test_list_pending_filtreaza(env, capsys):
|
|
_run(["create", "--name", "Activ SRL"])
|
|
_run(["create", "--name", "Asteptare SRL", "--inactive"])
|
|
capsys.readouterr()
|
|
rc = _run(["list", "--pending"])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "Asteptare SRL" in out
|
|
assert "Activ SRL" not in out
|