c9f9a1ca0e
PRD 5.16 — propagare design finalizata (system font stack, fara IBM Plex self-hostat): - US-001/002/008: tokeni --font-ui/--font-mono (system stack) + scala --fs-*; zero @font-face si zero /static/fonts/; landing aliniat la acelasi stack - US-003: RAR online = dot compact in antet + meniu burger; banda rosie DOAR pe blocat (invariant zero-silent-failures pastrat) - US-010: antet "ROMFAST AUTOPASS" + nume service + /login brandeit 2 coloane + badge plan; meniu burger cu separatoare; gate strict pe is_authenticated - US-011: selector tema pill icon+eticheta (reuse THEMES) - US-004/005/006/007: bug-fix editor prestatii (picker cod+denumire, add_extra in mod operatii, cod ales se salveaza fara "+", Renunta inchide via closest) - US-012/013: landing Autentificare->/login; wizard import colapsat + 4 pasi pe tokeni - fix VERIFY E2E: contoare duplicate pe 390px (inline display:flex batea @media) -> CSS + test-lock PRD 5.17 — tipuri de cont + trial Pro 30z + enforcement DUR: - US-001/002/008: accounts.tier + trial_until (migrare aditiva defensiva); app/plans.py sursa unica (PLANS, FREE_MONTHLY_LIMIT=60, effective_tier(now injectabil), monthly_usage, CONSUMED_STATUSES); create_account trial Pro 30z; CLI set-tier (protejat id=1, audit) - US-003/004/005: enforce volum 60/luna INAINTE de build_key pe ambele canale (PLAN_LIMITA_LUNARA, 3 niveluri + log_event); gate API Pro+ (PLAN_FARA_API 403 actionabil); valideaza/nomenclator raman permise; downgrade lazy; flag AUTOPASS_ENFORCE_PLANS (kill-switch) - US-006: badge plan antet + linie burger + consum N/60 + warn>=80% + 6 stari + copy RO pluralizat + banner one-time trial->Gratuit + pagina Cont Regresie: 1380 passed, 0 failed, 1 deselected (live). E2E browser pe 390/1280 confirmat. Backend trimitere (worker/masina stari/idempotenta/contract RAR) NEATINS. Lucrul 5.18 (corpus kNN) ramane separat, necomis. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
391 lines
14 KiB
Python
391 lines
14 KiB
Python
"""Teste US-001 (PRD 3.1): coloana accounts.active + helper-e cont in app/accounts.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def conn(monkeypatch):
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_accounts.db"))
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app.db import get_connection, init_db
|
|
init_db()
|
|
c = get_connection()
|
|
yield c
|
|
c.close()
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_create_account_returneaza_id(conn):
|
|
from app.accounts import create_account
|
|
acct_id = create_account(conn, "Service X")
|
|
assert isinstance(acct_id, int)
|
|
# AUTOINCREMENT peste default id=1 -> primul cont creat are id>=2 (nu atinge default).
|
|
assert acct_id >= 2
|
|
|
|
|
|
def test_create_account_activ_implicit(conn):
|
|
from app.accounts import create_account
|
|
acct_id = create_account(conn, "Service X")
|
|
row = conn.execute("SELECT active FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert row["active"] == 1
|
|
|
|
|
|
def test_create_account_inactiv(conn):
|
|
from app.accounts import create_account
|
|
acct_id = create_account(conn, "Service X", active=False)
|
|
row = conn.execute("SELECT active FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert row["active"] == 0
|
|
|
|
|
|
def test_create_account_name_gol_ridica_eroare(conn):
|
|
from app.accounts import create_account
|
|
with pytest.raises(ValueError):
|
|
create_account(conn, " ")
|
|
# nu a inserat nimic peste default
|
|
n = conn.execute("SELECT COUNT(*) AS n FROM accounts").fetchone()["n"]
|
|
assert n == 1
|
|
|
|
|
|
def test_create_account_cui_duplicat_respins(conn):
|
|
from app.accounts import create_account
|
|
first = create_account(conn, "Service A", cui="RO123")
|
|
with pytest.raises(ValueError) as exc:
|
|
create_account(conn, "Service B", cui="RO123")
|
|
# mesaj cu cauza + fix care numeste contul existent (A4)
|
|
msg = str(exc.value)
|
|
assert "RO123" in msg
|
|
assert str(first) in msg
|
|
|
|
|
|
def test_create_cui_null_multiplu_permis(conn):
|
|
from app.accounts import create_account
|
|
a = create_account(conn, "Fara CUI 1")
|
|
b = create_account(conn, "Fara CUI 2")
|
|
assert a != b
|
|
|
|
|
|
def test_create_cui_normalizat(conn):
|
|
from app.accounts import create_account
|
|
create_account(conn, "Service A", cui=" ro123 ")
|
|
# normalizat la RO123 -> duplicat respins indiferent de spatii/caz
|
|
with pytest.raises(ValueError):
|
|
create_account(conn, "Service B", cui="RO123")
|
|
row = conn.execute("SELECT cui FROM accounts WHERE name='Service A'").fetchone()
|
|
assert row["cui"] == "RO123"
|
|
|
|
|
|
def test_set_active_comuta(conn):
|
|
from app.accounts import create_account, set_active
|
|
acct_id = create_account(conn, "Service X")
|
|
set_active(conn, acct_id, False)
|
|
assert conn.execute("SELECT active FROM accounts WHERE id=?", (acct_id,)).fetchone()["active"] == 0
|
|
set_active(conn, acct_id, True)
|
|
assert conn.execute("SELECT active FROM accounts WHERE id=?", (acct_id,)).fetchone()["active"] == 1
|
|
|
|
|
|
def test_set_active_idempotent(conn):
|
|
from app.accounts import create_account, set_active
|
|
acct_id = create_account(conn, "Service X") # deja activ
|
|
set_active(conn, acct_id, True) # nu trebuie sa arunce
|
|
assert conn.execute("SELECT active FROM accounts WHERE id=?", (acct_id,)).fetchone()["active"] == 1
|
|
|
|
|
|
def test_set_active_inexistent_ridica(conn):
|
|
from app.accounts import set_active
|
|
with pytest.raises(ValueError):
|
|
set_active(conn, 9999, True)
|
|
|
|
|
|
def test_list_accounts_ordonat_fara_creds(conn):
|
|
from app.accounts import create_account, list_accounts
|
|
create_account(conn, "Service B")
|
|
create_account(conn, "Service A")
|
|
rows = list_accounts(conn)
|
|
ids = [r["id"] for r in rows]
|
|
assert ids == sorted(ids)
|
|
for r in rows:
|
|
assert "rar_creds_enc" not in r
|
|
assert set(r.keys()) == {"id", "name", "cui", "email", "active", "status", "created_at", "tier", "trial_until"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-001 (PRD 5.12): accounts.email + validari companie/email/CUI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_create_account_fara_email_ridica(conn):
|
|
"""create_account cu email="" ridica ValueError (email gol nu e acceptat)."""
|
|
from app.accounts import create_account
|
|
with pytest.raises(ValueError, match="email"):
|
|
create_account(conn, "Service X", cui="RO100", email="")
|
|
|
|
|
|
def test_create_account_fara_cui_ridica(conn):
|
|
"""create_account cu cui="" ridica ValueError (CUI gol nu e acceptat)."""
|
|
from app.accounts import create_account
|
|
with pytest.raises(ValueError, match="[Cc][Uu][Ii]|cod unic"):
|
|
create_account(conn, "Service X", cui="", email="test@test.com")
|
|
|
|
|
|
def test_email_normalizat_lowercase_trim(conn):
|
|
"""email e normalizat: trim + lower."""
|
|
from app.accounts import create_account
|
|
acct_id = create_account(conn, "Service X", cui="RO200", email=" Test@EXAMPLE.Com ")
|
|
row = conn.execute("SELECT email FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert row["email"] == "test@example.com"
|
|
|
|
|
|
def test_migrare_adauga_coloana_email_idempotent(conn):
|
|
"""_migrate e idempotent: ruleaza de doua ori fara eroare si coloana email exista."""
|
|
from app.db import _migrate
|
|
_migrate(conn) # a doua rulare (prima e in init_db)
|
|
cols = {r["name"] for r in conn.execute("PRAGMA table_info(accounts)").fetchall()}
|
|
assert "email" in cols
|
|
|
|
|
|
def test_account_is_complete_false_pe_legacy_incomplet(conn):
|
|
"""account_is_complete() returneaza False pe cont fara email sau fara CUI."""
|
|
from app.accounts import create_account, account_is_complete
|
|
# cont fara email si fara CUI
|
|
acct_id = create_account(conn, "Service Legacy")
|
|
row = conn.execute("SELECT * FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert account_is_complete(row) is False
|
|
|
|
# cont fara email, cu CUI
|
|
acct_id2 = create_account(conn, "Service Cu CUI", cui="RO300")
|
|
row2 = conn.execute("SELECT * FROM accounts WHERE id=?", (acct_id2,)).fetchone()
|
|
assert account_is_complete(row2) is False
|
|
|
|
# cont complet (cu email si CUI si name)
|
|
acct_id3 = create_account(conn, "Service Complet", cui="RO301", email="x@y.com")
|
|
row3 = conn.execute("SELECT * FROM accounts WHERE id=?", (acct_id3,)).fetchone()
|
|
assert account_is_complete(row3) is True
|
|
|
|
# contul sistem id=1 e EXCEPTAT (returneaza True indiferent)
|
|
row_sys = conn.execute("SELECT * FROM accounts WHERE id=1").fetchone()
|
|
assert account_is_complete(row_sys) is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5.17 US-001/US-008: schema tier + trial_until + set_tier + CLI set-tier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migrare_tier_trial_defensiva(conn):
|
|
"""_migrate adauga tier + trial_until pe conturi existente, e idempotenta."""
|
|
from app.db import _migrate
|
|
cols_before = {r["name"] for r in conn.execute("PRAGMA table_info(accounts)").fetchall()}
|
|
assert "tier" in cols_before
|
|
assert "trial_until" in cols_before
|
|
|
|
# a doua rulare: idempotenta (nu arunca)
|
|
_migrate(conn)
|
|
cols_after = {r["name"] for r in conn.execute("PRAGMA table_info(accounts)").fetchall()}
|
|
assert "tier" in cols_after
|
|
assert "trial_until" in cols_after
|
|
|
|
|
|
def test_cont_nou_tier_free_si_trial_30z(conn):
|
|
"""create_account seteaza tier='free' si trial_until = acum + ~30 zile."""
|
|
from datetime import datetime, timezone, timedelta
|
|
from app.accounts import create_account
|
|
|
|
before = datetime.now(timezone.utc)
|
|
acct_id = create_account(conn, "Service Trial", cui="RO_T1")
|
|
after = datetime.now(timezone.utc)
|
|
|
|
row = conn.execute("SELECT tier, trial_until FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert row["tier"] == "free"
|
|
assert row["trial_until"] is not None
|
|
|
|
tu = datetime.fromisoformat(row["trial_until"].replace(" ", "T"))
|
|
if tu.tzinfo is None:
|
|
tu = tu.replace(tzinfo=timezone.utc)
|
|
|
|
# trial_until trebuie sa fie intre now+29z si now+31z
|
|
assert tu >= before + timedelta(days=29)
|
|
assert tu <= after + timedelta(days=31)
|
|
|
|
|
|
def test_cont_nou_effective_tier_pro_in_trial(conn):
|
|
"""Cont nou are effective_tier='pro' (trial activ)."""
|
|
from datetime import datetime, timezone
|
|
from app.accounts import create_account
|
|
from app.plans import effective_tier
|
|
|
|
acct_id = create_account(conn, "Service Pro Trial", cui="RO_T2")
|
|
row = conn.execute("SELECT * FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
assert effective_tier(row, now) == "pro"
|
|
|
|
|
|
def test_set_tier_valid(conn):
|
|
"""set_tier seteaza tier-ul corect."""
|
|
from app.accounts import create_account, set_tier
|
|
|
|
acct_id = create_account(conn, "Service Tier", cui="RO_T3")
|
|
set_tier(conn, acct_id, "pro")
|
|
|
|
row = conn.execute("SELECT tier FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert row["tier"] == "pro"
|
|
|
|
|
|
def test_set_tier_cu_trial(conn):
|
|
"""set_tier cu trial_until seteaza ambele campuri."""
|
|
from app.accounts import create_account, set_tier
|
|
|
|
acct_id = create_account(conn, "Service Tier Trial", cui="RO_T4")
|
|
set_tier(conn, acct_id, "standard", trial_until="2026-12-31 00:00:00")
|
|
|
|
row = conn.execute("SELECT tier, trial_until FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert row["tier"] == "standard"
|
|
assert row["trial_until"] == "2026-12-31 00:00:00"
|
|
|
|
|
|
def test_set_tier_no_trial_sterge_trial_until(conn):
|
|
"""set_tier cu trial_until=None sterge trial-ul existent."""
|
|
from app.accounts import create_account, set_tier
|
|
|
|
acct_id = create_account(conn, "Service No Trial", cui="RO_T5")
|
|
# mai intai setam un trial
|
|
set_tier(conn, acct_id, "pro", trial_until="2026-12-31 00:00:00")
|
|
# acum stergem trial-ul
|
|
set_tier(conn, acct_id, "pro", trial_until=None)
|
|
|
|
row = conn.execute("SELECT tier, trial_until FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
assert row["tier"] == "pro"
|
|
assert row["trial_until"] is None
|
|
|
|
|
|
def test_set_tier_invalid_respins(conn):
|
|
"""set_tier cu tier invalid ridica ValueError cu mesaj clar."""
|
|
from app.accounts import create_account, set_tier
|
|
|
|
acct_id = create_account(conn, "Service Tier Invalid", cui="RO_T6")
|
|
with pytest.raises(ValueError, match="tier invalid"):
|
|
set_tier(conn, acct_id, "gold")
|
|
|
|
|
|
def test_set_tier_protejeaza_id1(conn):
|
|
"""set_tier pe contul de sistem id=1 ridica ValueError (protejat)."""
|
|
from app.accounts import set_tier
|
|
|
|
with pytest.raises(ValueError):
|
|
set_tier(conn, 1, "pro")
|
|
|
|
|
|
def test_set_tier_cont_inexistent_ridica(conn):
|
|
"""set_tier pe cont inexistent ridica ValueError."""
|
|
from app.accounts import set_tier
|
|
|
|
with pytest.raises(ValueError, match="inexistent"):
|
|
set_tier(conn, 9999, "pro")
|
|
|
|
|
|
def test_list_accounts_include_tier_trial(conn):
|
|
"""list_accounts include coloanele tier si trial_until."""
|
|
from app.accounts import create_account, list_accounts
|
|
|
|
create_account(conn, "Service List", cui="RO_L1")
|
|
rows = list_accounts(conn)
|
|
for r in rows:
|
|
assert "tier" in r
|
|
assert "trial_until" in r
|
|
|
|
|
|
def test_default_account_tier_free_fara_trial(conn):
|
|
"""Contul implicit id=1 (creat de schema) are tier='free' si trial_until=NULL."""
|
|
row = conn.execute("SELECT tier, trial_until FROM accounts WHERE id=1").fetchone()
|
|
assert row["tier"] == "free"
|
|
assert row["trial_until"] is None
|
|
|
|
|
|
def test_cli_set_tier(monkeypatch):
|
|
"""CLI set-tier seteaza tier-ul unui cont (--no-trial)."""
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_cli_tier.db"))
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
|
|
from tools.account import main
|
|
from app.db import init_db, get_connection
|
|
from app.accounts import create_account
|
|
|
|
init_db()
|
|
conn_tmp = get_connection()
|
|
acct_id = create_account(conn_tmp, "CLI Test", cui="RO_CLI1")
|
|
conn_tmp.close()
|
|
|
|
rc = main(["set-tier", "--account", str(acct_id), "--tier", "pro", "--no-trial"])
|
|
assert rc == 0
|
|
|
|
conn_tmp2 = get_connection()
|
|
row = conn_tmp2.execute("SELECT tier, trial_until FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
conn_tmp2.close()
|
|
assert row["tier"] == "pro"
|
|
assert row["trial_until"] is None
|
|
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_cli_set_tier_cu_trial_days(monkeypatch):
|
|
"""CLI set-tier cu --trial-days 14 seteaza trial_until = acum + 14z."""
|
|
from datetime import datetime, timezone, timedelta
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_cli_tier2.db"))
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
|
|
from tools.account import main
|
|
from app.db import init_db, get_connection
|
|
from app.accounts import create_account
|
|
|
|
init_db()
|
|
conn_tmp = get_connection()
|
|
acct_id = create_account(conn_tmp, "CLI Trial", cui="RO_CLI2")
|
|
conn_tmp.close()
|
|
|
|
before = datetime.now(timezone.utc)
|
|
rc = main(["set-tier", "--account", str(acct_id), "--tier", "pro", "--trial-days", "14"])
|
|
assert rc == 0
|
|
after = datetime.now(timezone.utc)
|
|
|
|
conn_tmp2 = get_connection()
|
|
row = conn_tmp2.execute("SELECT tier, trial_until FROM accounts WHERE id=?", (acct_id,)).fetchone()
|
|
conn_tmp2.close()
|
|
assert row["tier"] == "pro"
|
|
assert row["trial_until"] is not None
|
|
tu = datetime.fromisoformat(row["trial_until"].replace(" ", "T")).replace(tzinfo=timezone.utc)
|
|
assert tu >= before + timedelta(days=13)
|
|
assert tu <= after + timedelta(days=15)
|
|
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_cli_set_tier_invalid(monkeypatch):
|
|
"""CLI set-tier cu tier invalid: exit code != 0."""
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_cli_tier3.db"))
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
|
|
from tools.account import main
|
|
from app.db import init_db, get_connection
|
|
from app.accounts import create_account
|
|
|
|
init_db()
|
|
conn_tmp = get_connection()
|
|
acct_id = create_account(conn_tmp, "CLI Invalid", cui="RO_CLI3")
|
|
conn_tmp.close()
|
|
|
|
rc = main(["set-tier", "--account", str(acct_id), "--tier", "diamond"])
|
|
assert rc != 0
|
|
|
|
get_settings.cache_clear()
|