301 lines
10 KiB
Python
301 lines
10 KiB
Python
import os
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
import app.services.admin as admin_service
|
|
from app.services.admin import apply_rules_retroactively, change_password
|
|
from app.models.tables import Account, Category, CategoryRule, Transaction
|
|
|
|
|
|
# --- change_password(): reine Service-Tests (kein HTTP), gegen die
|
|
# FB_ENV_FILE-Datei aus client_with_env_file -----------------------------
|
|
|
|
def test_change_password_wrong_old_raises_without_touching_grafana_or_env(
|
|
client_with_env_file, env_file, monkeypatch
|
|
):
|
|
_client, password = client_with_env_file
|
|
path, _ = env_file
|
|
before = path.read_text()
|
|
calls = []
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password",
|
|
lambda *a, **kw: calls.append(a))
|
|
|
|
with pytest.raises(ValueError):
|
|
change_password("falsches-passwort", "neuesPasswort123")
|
|
|
|
assert calls == [] # Grafana darf bei falschem alten Passwort nie aufgerufen werden
|
|
assert path.read_text() == before
|
|
|
|
|
|
def test_change_password_calls_grafana_before_writing_env(
|
|
client_with_env_file, env_file, monkeypatch
|
|
):
|
|
_client, password = client_with_env_file
|
|
path, _ = env_file
|
|
order = []
|
|
|
|
def fake_sync(old, new, base_url):
|
|
order.append("grafana")
|
|
assert path.read_text() == before # .env noch unveraendert an diesem Punkt
|
|
|
|
before = path.read_text()
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password", fake_sync)
|
|
|
|
change_password(password, "neuesPasswort123")
|
|
|
|
order.append("env-written")
|
|
assert order == ["grafana", "env-written"]
|
|
assert "FB_PASSWORD='neuesPasswort123'" in path.read_text()
|
|
|
|
|
|
def test_change_password_grafana_failure_leaves_env_unchanged(
|
|
client_with_env_file, env_file, monkeypatch
|
|
):
|
|
_client, password = client_with_env_file
|
|
path, _ = env_file
|
|
before = path.read_text()
|
|
|
|
def fail(*a, **kw):
|
|
raise RuntimeError("Grafana-Passwortänderung fehlgeschlagen: boom")
|
|
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password", fail)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
change_password(password, "neuesPasswort123")
|
|
|
|
assert path.read_text() == before
|
|
|
|
|
|
def test_change_password_success_rewrites_env_in_place_preserving_inode(
|
|
client_with_env_file, env_file, monkeypatch
|
|
):
|
|
_client, password = client_with_env_file
|
|
path, _ = env_file
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password",
|
|
lambda *a, **kw: None)
|
|
|
|
inode_before = os.stat(path).st_ino
|
|
change_password(password, "neuesPasswort123")
|
|
inode_after = os.stat(path).st_ino
|
|
|
|
assert inode_before == inode_after # KRITISCH: kein rename, gleicher Inode
|
|
content = path.read_text()
|
|
assert "FB_PASSWORD='neuesPasswort123'" in content
|
|
assert "FB_GUI_PASSWORD_HASH='" in content
|
|
assert f"FB_GUI_PASSWORD_HASH='{password}'" not in content
|
|
|
|
|
|
def test_change_password_new_password_logs_in_old_does_not(
|
|
client_with_env_file, monkeypatch
|
|
):
|
|
client, password = client_with_env_file
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password",
|
|
lambda *a, **kw: None)
|
|
|
|
change_password(password, "neuesPasswort123")
|
|
|
|
r = client.post("/login", data={"username": "admin", "password": "neuesPasswort123"},
|
|
follow_redirects=False)
|
|
assert r.status_code == 303
|
|
|
|
client.cookies.clear()
|
|
r2 = client.post("/login", data={"username": "admin", "password": password},
|
|
follow_redirects=False)
|
|
assert r2.status_code == 401
|
|
|
|
|
|
# --- /admin GUI-Route: Auth-Gate + Formular-Fehlerpfade -------------------
|
|
|
|
def test_admin_page_requires_gui_session(client_with_env_file):
|
|
client, _password = client_with_env_file
|
|
r = client.get("/admin", follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert r.headers["location"] == "/login"
|
|
|
|
|
|
def test_admin_page_reachable_after_login(client_with_env_file):
|
|
client, password = client_with_env_file
|
|
client.post("/login", data={"username": "admin", "password": password})
|
|
r = client.get("/admin")
|
|
assert r.status_code == 200
|
|
assert "Passwort ändern" in r.text
|
|
assert "Regeln neu anwenden" in r.text
|
|
|
|
|
|
def test_admin_passwort_wrong_old_returns_400_with_message(
|
|
client_with_env_file, monkeypatch
|
|
):
|
|
client, password = client_with_env_file
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password",
|
|
lambda *a, **kw: None)
|
|
client.post("/login", data={"username": "admin", "password": password})
|
|
|
|
r = client.post("/admin/passwort", data={
|
|
"alt": "falsch", "neu": "neuesPasswort123", "neu2": "neuesPasswort123",
|
|
})
|
|
|
|
assert r.status_code == 400
|
|
assert "falsch" in r.text.lower()
|
|
|
|
|
|
def test_admin_passwort_too_short_returns_400(client_with_env_file):
|
|
client, password = client_with_env_file
|
|
client.post("/login", data={"username": "admin", "password": password})
|
|
|
|
r = client.post("/admin/passwort", data={
|
|
"alt": password, "neu": "kurz1", "neu2": "kurz1",
|
|
})
|
|
|
|
assert r.status_code == 400
|
|
assert "8 Zeichen" in r.text
|
|
|
|
|
|
def test_admin_passwort_mismatch_returns_400(client_with_env_file):
|
|
client, password = client_with_env_file
|
|
client.post("/login", data={"username": "admin", "password": password})
|
|
|
|
r = client.post("/admin/passwort", data={
|
|
"alt": password, "neu": "neuesPasswort123", "neu2": "andersPasswort123",
|
|
})
|
|
|
|
assert r.status_code == 400
|
|
assert "stimmt nicht" in r.text.lower() or "wiederholung" in r.text.lower()
|
|
|
|
|
|
def test_admin_passwort_grafana_failure_returns_400_env_unchanged(
|
|
client_with_env_file, env_file, monkeypatch
|
|
):
|
|
client, password = client_with_env_file
|
|
path, _ = env_file
|
|
before = path.read_text()
|
|
|
|
def fail(*a, **kw):
|
|
raise RuntimeError("Grafana-Passwortänderung fehlgeschlagen: boom")
|
|
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password", fail)
|
|
client.post("/login", data={"username": "admin", "password": password})
|
|
|
|
r = client.post("/admin/passwort", data={
|
|
"alt": password, "neu": "neuesPasswort123", "neu2": "neuesPasswort123",
|
|
})
|
|
|
|
assert r.status_code == 400
|
|
assert "grafana" in r.text.lower()
|
|
assert path.read_text() == before
|
|
|
|
|
|
def test_admin_passwort_grafana_read_timeout_returns_400_env_unchanged(
|
|
client_with_env_file, env_file, monkeypatch
|
|
):
|
|
"""F1 (Fable-Reject): ein Hang in der Grafana-Lesephase wirft ein rohes
|
|
TimeoutError statt eines urllib.error.URLError. Ohne expliziten Fang in
|
|
_sync_grafana_password schlaegt das bis in den Router durch (500 statt
|
|
deutscher 400-Fehlermeldung) - siehe app/services/admin.py."""
|
|
client, password = client_with_env_file
|
|
path, _ = env_file
|
|
before = path.read_text()
|
|
|
|
def hang(request, timeout=None):
|
|
raise TimeoutError("timed out")
|
|
|
|
monkeypatch.setattr("app.services.admin.urllib.request.urlopen", hang)
|
|
client.post("/login", data={"username": "admin", "password": password})
|
|
|
|
r = client.post("/admin/passwort", data={
|
|
"alt": password, "neu": "neuesPasswort123", "neu2": "neuesPasswort123",
|
|
})
|
|
|
|
assert r.status_code == 400
|
|
assert "grafana" in r.text.lower()
|
|
assert path.read_text() == before
|
|
|
|
|
|
def test_admin_passwort_success_returns_200_with_confirmation(
|
|
client_with_env_file, monkeypatch
|
|
):
|
|
client, password = client_with_env_file
|
|
calls = []
|
|
monkeypatch.setattr(admin_service, "_sync_grafana_password",
|
|
lambda *a, **kw: calls.append(a))
|
|
client.post("/login", data={"username": "admin", "password": password})
|
|
|
|
r = client.post("/admin/passwort", data={
|
|
"alt": password, "neu": "neuesPasswort123", "neu2": "neuesPasswort123",
|
|
})
|
|
|
|
assert r.status_code == 200
|
|
assert "erfolgreich" in r.text.lower()
|
|
assert calls # Grafana wurde tatsaechlich aufgerufen
|
|
|
|
# Session-Cookie bleibt nach der Aenderung gueltig (Secret unveraendert)
|
|
r2 = client.get("/admin")
|
|
assert r2.status_code == 200
|
|
|
|
|
|
# --- Regeln neu anwenden ---------------------------------------------------
|
|
|
|
def test_apply_rules_retroactively_categorizes_uncategorized_confirmed_tx(db):
|
|
account = Account(bank="Test", iban="DE02120300000000202099", name="Giro")
|
|
db.add(account)
|
|
db.flush()
|
|
category = Category(name="Lebensmittel")
|
|
db.add(category)
|
|
db.flush()
|
|
db.add(CategoryRule(pattern="rewe", category_id=category.id, priority=100))
|
|
tx_confirmed_uncategorized = Transaction(
|
|
account_id=account.id, booking_date=date(2026, 1, 5),
|
|
amount=Decimal("-10.00"), purpose="REWE Markt", counterparty="",
|
|
status="confirmed", dedup_hash="h1",
|
|
)
|
|
tx_draft = Transaction(
|
|
account_id=account.id, booking_date=date(2026, 1, 6),
|
|
amount=Decimal("-5.00"), purpose="REWE Markt", counterparty="",
|
|
status="draft", dedup_hash="h2",
|
|
)
|
|
tx_already_categorized = Transaction(
|
|
account_id=account.id, booking_date=date(2026, 1, 7),
|
|
amount=Decimal("-5.00"), purpose="REWE Markt", counterparty="",
|
|
status="confirmed", category_id=category.id, dedup_hash="h3",
|
|
)
|
|
db.add_all([tx_confirmed_uncategorized, tx_draft, tx_already_categorized])
|
|
db.commit()
|
|
|
|
hits = apply_rules_retroactively(db)
|
|
|
|
assert hits == 1
|
|
db.refresh(tx_confirmed_uncategorized)
|
|
db.refresh(tx_draft)
|
|
assert tx_confirmed_uncategorized.category_id == category.id
|
|
assert tx_draft.category_id is None # Entwuerfe bleiben unangetastet
|
|
|
|
|
|
def test_apply_category_rules_endpoint_requires_auth(client):
|
|
assert client.post("/api/category-rules/apply").status_code == 401
|
|
|
|
|
|
def test_apply_category_rules_endpoint_categorizes_and_returns_count(client, db):
|
|
account = Account(bank="Test", iban="DE02120300000000202098", name="Giro")
|
|
db.add(account)
|
|
db.flush()
|
|
category = Category(name="Lebensmittel")
|
|
db.add(category)
|
|
db.flush()
|
|
db.add(CategoryRule(pattern="rewe", category_id=category.id, priority=100))
|
|
tx = Transaction(
|
|
account_id=account.id, booking_date=date(2026, 1, 5),
|
|
amount=Decimal("-10.00"), purpose="REWE Markt", counterparty="",
|
|
status="confirmed", dedup_hash="h1",
|
|
)
|
|
db.add(tx)
|
|
db.commit()
|
|
|
|
r = client.post("/api/category-rules/apply",
|
|
headers={"Authorization": "Bearer test-key"})
|
|
|
|
assert r.status_code == 200
|
|
assert r.json() == {"categorized": 1}
|
|
db.refresh(tx)
|
|
assert tx.category_id == category.id
|