192 lines
8.8 KiB
Python
192 lines
8.8 KiB
Python
import email
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import mailer
|
|
from app.main import app
|
|
|
|
H = {"X-Requested-With": "budget-app"}
|
|
|
|
|
|
class FakeSMTP:
|
|
sent: list = []
|
|
fail_login = False
|
|
|
|
def __init__(self, host, port, timeout=None, context=None):
|
|
self.host, self.port = host, port
|
|
|
|
def ehlo(self):
|
|
pass
|
|
|
|
def starttls(self, context=None):
|
|
pass
|
|
|
|
def login(self, user, password):
|
|
if FakeSMTP.fail_login or password != "smtp-secret":
|
|
import smtplib
|
|
raise smtplib.SMTPAuthenticationError(535, b"bad credentials")
|
|
|
|
def send_message(self, msg, from_addr=None, to_addrs=None):
|
|
FakeSMTP.sent.append((from_addr, list(to_addrs), msg))
|
|
return {}
|
|
|
|
def quit(self):
|
|
pass
|
|
|
|
|
|
class FakeIMAP:
|
|
appended: list = []
|
|
folders = [b'(\\HasNoChildren) "/" "INBOX"', b'(\\HasNoChildren \\Sent) "/" "Itens Enviados"',
|
|
b'(\\HasNoChildren) "/" "Sent"']
|
|
|
|
def __init__(self, host, port, ssl_context=None, timeout=None):
|
|
pass
|
|
|
|
def login(self, user, password):
|
|
assert password == "smtp-secret"
|
|
|
|
def list(self):
|
|
return "OK", list(FakeIMAP.folders)
|
|
|
|
def append(self, folder, flags, date, data):
|
|
FakeIMAP.appended.append((folder, flags, data))
|
|
return "OK", [b"APPEND completed"]
|
|
|
|
def logout(self):
|
|
pass
|
|
|
|
|
|
@pytest.fixture()
|
|
def fake_mail(monkeypatch):
|
|
FakeSMTP.sent, FakeIMAP.appended, FakeSMTP.fail_login = [], [], False
|
|
monkeypatch.setattr(mailer.smtplib, "SMTP", FakeSMTP)
|
|
monkeypatch.setattr(mailer.smtplib, "SMTP_SSL", FakeSMTP)
|
|
monkeypatch.setattr(mailer.imaplib, "IMAP4_SSL", FakeIMAP)
|
|
monkeypatch.setattr(mailer.imaplib, "IMAP4", FakeIMAP)
|
|
|
|
|
|
def configure(client, **extra):
|
|
body = {"enabled": True, "smtp_host": "mail.local", "smtp_port": 587, "smtp_security": "starttls",
|
|
"smtp_username": "budget@local.test", "smtp_password": "smtp-secret",
|
|
"from_address": "budget@local.test", "from_name": "Budget", "save_to_sent": True,
|
|
"imap_host": "mail.local", "imap_port": 993, "imap_security": "ssl", **extra}
|
|
r = client.put("/api/mail/settings", json=body)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
|
|
def test_settings_hide_and_encrypt_passwords(client, fake_mail):
|
|
out = configure(client)
|
|
assert out["smtp_password_set"] and "smtp_password" not in out and out["ready"]
|
|
from app.db import SessionLocal
|
|
from app.models import MailSettings
|
|
with SessionLocal() as db:
|
|
stored = db.get(MailSettings, 1).smtp_password
|
|
assert stored and "smtp-secret" not in stored and mailer.decrypt(stored) == "smtp-secret"
|
|
# saving again without a password keeps it
|
|
configure(client, smtp_password="")
|
|
assert client.post("/api/mail/test", json={}).json() == {"smtp": "ok", "imap": "ok", "sent_folder": "Itens Enviados"}
|
|
|
|
|
|
def test_send_report_saves_sent_copy_and_logs(client, imported, fake_mail):
|
|
configure(client)
|
|
r = client.post(f"/api/years/{imported}/email", json={
|
|
"to": ["cfo@example.com; ops@example.com"], "cc": ["boss@example.com"], "subject": "Budget 2027",
|
|
"message": "Segue o orçamento.", "lang": "pt", "copy_me": True,
|
|
"reports": [{"kind": "cost-report", "fmt": "pdf"}, {"kind": "budget", "fmt": "xlsx"}]})
|
|
assert r.status_code == 200, r.text
|
|
log = r.json()
|
|
assert log["status"] == "sent" and log["sent_copy"] == "saved" and log["sent_copy_detail"] == "Itens Enviados"
|
|
assert log["to"] == ["cfo@example.com", "ops@example.com"]
|
|
assert log["cc"] == ["boss@example.com", "admin@test.local"]
|
|
|
|
sender, rcpts, msg = FakeSMTP.sent[0]
|
|
assert sender == "budget@local.test" and len(rcpts) == 4
|
|
names = [p.get_filename() for p in msg.iter_attachments()]
|
|
assert names == ["Budget_IT_2027.xlsx", "Capex_e_Custos_IT_2027.pdf"]
|
|
folder, flags, data = FakeIMAP.appended[0]
|
|
assert folder == '"Itens Enviados"' and flags == r"(\Seen)"
|
|
copy = email.message_from_bytes(data)
|
|
assert copy["Message-ID"] == msg["Message-ID"] and copy["Subject"] == "Budget 2027"
|
|
|
|
rows = client.get("/api/mail/log").json()
|
|
assert rows["total"] == 1 and rows["rows"][0]["user"] == "Administrator"
|
|
assert [a["name"] for a in rows["rows"][0]["attachments"]] == names
|
|
assert "cfo@example.com" in client.get("/api/mail/log.csv").text
|
|
acts = client.get("/api/activity?entity=mail").json()["rows"]
|
|
assert acts[0]["action"] == "send" and "cfo@example.com" in acts[0]["summary"]
|
|
|
|
|
|
def test_failed_send_is_logged(client, imported, fake_mail):
|
|
configure(client)
|
|
FakeSMTP.fail_login = True
|
|
r = client.post(f"/api/years/{imported}/email", json={"to": ["a@example.com"],
|
|
"reports": [{"kind": "budget", "fmt": "pdf"}]})
|
|
assert r.status_code == 502 and r.json()["detail"] == "smtp_auth"
|
|
row = client.get("/api/mail/log?status=failed").json()["rows"][0]
|
|
assert row["error"].startswith("smtp_auth") and row["sent_copy"] == "off"
|
|
assert not FakeIMAP.appended
|
|
|
|
|
|
def test_sent_folder_found_by_name(fake_mail, monkeypatch):
|
|
monkeypatch.setattr(FakeIMAP, "folders", [b'(\\HasNoChildren) "." "INBOX"', b'(\\HasNoChildren) "." "INBOX.Sent"'])
|
|
assert mailer.find_sent_folder(FakeIMAP("h", 1), "") == "INBOX.Sent"
|
|
with pytest.raises(mailer.MailError) as exc:
|
|
mailer.find_sent_folder(FakeIMAP("h", 1), "Outbox")
|
|
assert exc.value.code == "sent_folder_not_found"
|
|
|
|
|
|
def test_mail_rules(client, imported, fake_mail):
|
|
# not configured yet
|
|
r = client.post(f"/api/years/{imported}/email", json={"to": ["a@example.com"],
|
|
"reports": [{"kind": "budget", "fmt": "pdf"}]})
|
|
assert r.status_code == 409
|
|
configure(client)
|
|
bad = client.post(f"/api/years/{imported}/email", json={"to": ["not-an-address"],
|
|
"reports": [{"kind": "budget", "fmt": "pdf"}]})
|
|
assert bad.status_code == 400 and bad.json()["detail"].startswith("invalid_address")
|
|
assert client.post(f"/api/years/{imported}/email", json={"to": ["a@example.com"], "reports": []}).status_code == 400
|
|
|
|
client.post("/api/users", json={"email": "v@t.local", "name": "Viewer", "role": "viewer",
|
|
"password": "password-123"})
|
|
viewer = TestClient(app)
|
|
viewer.headers.update(H)
|
|
viewer.post("/api/auth/login", json={"email": "v@t.local", "password": "password-123"})
|
|
# any signed-in user may email a report; only admins see the settings and the logs
|
|
assert viewer.get("/api/mail/status").json()["enabled"] is True
|
|
assert viewer.post(f"/api/years/{imported}/email", json={
|
|
"to": ["a@example.com"], "reports": [{"kind": "budget", "fmt": "xlsx"}]}).status_code == 200
|
|
for url in ("/api/mail/settings", "/api/mail/log", "/api/mail/log.csv", "/api/activity", "/api/activity.csv"):
|
|
assert viewer.get(url).status_code == 403, url
|
|
assert viewer.put("/api/mail/settings", json={"enabled": False}).status_code == 403
|
|
|
|
|
|
def test_only_admin_deletes_budgets(client, imported):
|
|
client.post("/api/users", json={"email": "ap@t.local", "name": "Approver", "role": "approver",
|
|
"password": "password-123", "department_id": 1})
|
|
ap = TestClient(app)
|
|
ap.headers.update(H)
|
|
ap.post("/api/auth/login", json={"email": "ap@t.local", "password": "password-123"})
|
|
assert ap.get(f"/api/years/{imported}").json()["actions"]["delete"] is False
|
|
assert ap.delete(f"/api/years/{imported}").status_code == 403
|
|
# an admin can delete it even after approval
|
|
assert client.post(f"/api/years/{imported}/status", json={"action": "submit"}).status_code == 200
|
|
assert client.get(f"/api/years/{imported}").json()["actions"]["delete"] is True
|
|
assert client.delete(f"/api/years/{imported}").status_code == 200
|
|
assert client.get(f"/api/years/{imported}").status_code == 404
|
|
row = client.get("/api/activity?action=delete&entity=year").json()["rows"][0]
|
|
assert row["user"] == "Administrator" and "IT 2027" in row["summary"] and row["budget_exists"] is False
|
|
|
|
|
|
def test_activity_log_records_who_does_what(client, imported):
|
|
client.post("/api/auth/login", json={"email": "admin@test.local", "password": "wrong"})
|
|
client.get(f"/api/years/{imported}/reports/budget.xlsx")
|
|
data = client.get("/api/activity").json()
|
|
actions = [(r["entity"], r["action"]) for r in data["rows"]]
|
|
assert ("report", "download") in actions and ("user", "login_failed") in actions and ("year", "import") in actions
|
|
assert data["rows"][0]["budget"] == "IT 2027" or data["rows"][0]["entity"] == "user"
|
|
only = client.get("/api/activity?entity=report").json()
|
|
assert only["total"] == 1 and only["rows"][0]["budget"] == "IT 2027"
|
|
assert "budget.xlsx" in client.get("/api/activity.csv?entity=report").text
|