200 lines
8.5 KiB
Python
200 lines
8.5 KiB
Python
"""Benachrichtigungen: E-Mail zu Freigaben, Webhook je Workspace, Profil, Zustellung mit Wiederholung."""
|
|
|
|
import json
|
|
import re
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.approvals.models import ApprovalRequest
|
|
from app.identity import Principal
|
|
from app.identity.models import User
|
|
from app.notify import service as notify
|
|
from app.notify.models import Notification
|
|
from app.tasks.worker import Worker
|
|
from app.tools import service as tools
|
|
from helpers import connect_repo, create_workspace
|
|
|
|
CHANGE = {"title": "README <ergänzen>", "text": "Neue Zeile."}
|
|
|
|
|
|
class FakeSMTP:
|
|
"""Ersetzt smtplib.SMTP: merkt sich verschickte Nachrichten."""
|
|
|
|
sent: list = []
|
|
fail = False
|
|
|
|
def __init__(self, host, port, timeout):
|
|
self.host, self.port = host, port
|
|
|
|
def __enter__(self):
|
|
if FakeSMTP.fail:
|
|
raise ConnectionRefusedError("Mailserver weg")
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def starttls(self):
|
|
pass
|
|
|
|
def login(self, user, password):
|
|
pass
|
|
|
|
def send_message(self, message):
|
|
FakeSMTP.sent.append(message)
|
|
|
|
|
|
@pytest.fixture
|
|
def smtp(monkeypatch):
|
|
FakeSMTP.sent, FakeSMTP.fail = [], False
|
|
monkeypatch.setattr(notify.smtplib, "SMTP", FakeSMTP)
|
|
return FakeSMTP
|
|
|
|
|
|
@pytest.fixture
|
|
def settings(settings):
|
|
return settings.model_copy(update={"smtp_host": "mail.intern", "smtp_from": "aiw@firma.de",
|
|
"public_url": "https://aiw.intern", "webhook_allow_local": True})
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(browser, people, db, git_repo, smtp):
|
|
with db.session() as session:
|
|
session.get(User, people["anna"]).is_admin = True
|
|
for name, address in (("anna", "anna@firma.de"), ("clara", "clara@firma.de"), ("ben", "ben@firma.de")):
|
|
session.get(User, people[name]).email = address
|
|
anna = browser("anna")
|
|
workspace_id = create_workspace(anna, people["team_demo"], "Plattform")
|
|
connect_repo(anna, workspace_id, git_repo)
|
|
anna.post(f"/workspaces/{workspace_id}/members", {"username": "ben", "role": "member"})
|
|
anna.post(f"/workspaces/{workspace_id}/members", {"username": "clara", "role": "admin"})
|
|
admin = Principal(user_id=people["anna"], username="anna", display_name="Anna", is_admin=True,
|
|
team_ids=frozenset({people["team_demo"]}))
|
|
with db.session() as session:
|
|
tools.classify_suggested(session, admin)
|
|
ben = browser("ben")
|
|
chat_id = int(ben.post(f"/workspaces/{workspace_id}/chats").headers["location"].rsplit("/", 1)[1])
|
|
return {"anna": anna, "ben": ben, "clara": browser("clara"), "id": workspace_id, "chat_id": chat_id}
|
|
|
|
|
|
def propose(setup, db, settings, fake_provider) -> ApprovalRequest:
|
|
fake_provider.script = [[("demo_write", {**CHANGE, "source": "Repo"})], "Wartet."]
|
|
setup["ben"].post(f"/chats/{setup['chat_id']}/messages", {"content": "Bitte"}, headers={"HX-Request": "true"})
|
|
Worker(db, settings).run_once()
|
|
with db.session() as session:
|
|
approval = session.scalar(select(ApprovalRequest))
|
|
session.expunge(approval)
|
|
return approval
|
|
|
|
|
|
def test_mail_an_entscheider_und_an_antragsteller(setup, db, settings, fake_provider, smtp):
|
|
approval = propose(setup, db, settings, fake_provider)
|
|
Worker(db, settings).run_once() # Zustellung
|
|
to = sorted(m["To"] for m in smtp.sent)
|
|
assert to == ["anna@firma.de", "clara@firma.de"] # beide Admins, nicht ben (Antragsteller)
|
|
message = smtp.sent[0]
|
|
assert message["Subject"] == "Freigabe nötig: README ergänzen (Plattform)"
|
|
assert message["From"] == "aiw@firma.de"
|
|
assert f"https://aiw.intern/approvals/{approval.id}" in message.get_content()
|
|
# Entscheidung → Mail an ben
|
|
smtp.sent.clear()
|
|
setup["clara"].post(f"/approvals/{approval.id}/decide", {"decision": "reject", "comment": "Bitte kleiner."})
|
|
Worker(db, settings).run_once()
|
|
assert [m["To"] for m in smtp.sent] == ["ben@firma.de"]
|
|
body = smtp.sent[0].get_content()
|
|
assert smtp.sent[0]["Subject"] == "Freigabe abgelehnt: README ergänzen"
|
|
assert "Kommentar: Bitte kleiner." in body and f"https://aiw.intern/chats/{setup['chat_id']}" in body
|
|
|
|
|
|
def test_abbestellt_und_ohne_mailserver(setup, db, settings, fake_provider, smtp, people):
|
|
setup["clara"].post("/profile", {"email": "clara@firma.de"}) # notify_me fehlt = abbestellt
|
|
propose(setup, db, settings, fake_provider)
|
|
Worker(db, settings).run_once()
|
|
assert [m["To"] for m in smtp.sent] == ["anna@firma.de"]
|
|
# Ohne Mailserver: übersprungen statt gestaut
|
|
with db.session() as session:
|
|
session.add(Notification(channel="email", user_id=people["anna"], event="test", subject="x", body="y"))
|
|
Worker(db, settings.model_copy(update={"smtp_host": ""})).run_once()
|
|
with db.session() as session:
|
|
assert session.scalar(select(Notification.status).order_by(Notification.id.desc())) == "skipped"
|
|
|
|
|
|
def test_wiederholung_bis_aufgegeben(setup, db, settings, people, smtp):
|
|
smtp.fail = True
|
|
with db.session() as session:
|
|
session.add(Notification(channel="email", user_id=people["anna"], event="test", subject="x", body="y"))
|
|
for _ in range(notify.MAX_ATTEMPTS):
|
|
notify.deliver(db, settings)
|
|
with db.session() as session:
|
|
item = session.scalar(select(Notification))
|
|
item.available_at = item.created # sofort wieder fällig
|
|
with db.session() as session:
|
|
item = session.scalar(select(Notification))
|
|
assert item.status == "failed" and item.attempts == notify.MAX_ATTEMPTS and "Mailserver weg" in item.error
|
|
assert "fehlgeschlagen" in setup["anna"].get("/admin/notifications").text
|
|
assert setup["ben"].get("/admin/notifications").status_code == 403
|
|
|
|
|
|
class Hook(BaseHTTPRequestHandler):
|
|
received: list = []
|
|
|
|
def do_POST(self):
|
|
Hook.received.append(json.loads(self.rfile.read(int(self.headers["Content-Length"]))))
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def hook():
|
|
Hook.received = []
|
|
server = HTTPServer(("127.0.0.1", 0), Hook)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
yield f"http://localhost:{server.server_port}/hook"
|
|
server.shutdown()
|
|
|
|
|
|
def test_webhook_des_workspace(setup, db, settings, fake_provider, hook):
|
|
base = f"/workspaces/{setup['id']}"
|
|
assert setup["ben"].post(f"{base}/webhook", {"url": hook}).status_code == 403
|
|
assert setup["anna"].post(f"{base}/webhook", {"url": "http://evil.example/x"}).status_code == 400
|
|
assert setup["anna"].post(f"{base}/webhook", {"url": "https://evil.example/x"}).status_code == 400
|
|
assert setup["anna"].post(f"{base}/webhook", {"url": hook}).status_code == 303
|
|
page = setup["anna"].get(f"{base}/settings").text
|
|
assert "Webhook eingerichtet: <code>localhost</code>" in page and hook not in page
|
|
setup["anna"].post(f"{base}/webhook/test")
|
|
propose(setup, db, settings, fake_provider)
|
|
Worker(db, settings).run_once()
|
|
texts = [r["text"] for r in Hook.received]
|
|
assert texts[0] == "Test-Nachricht aus Bifröst, Workspace „Plattform“."
|
|
assert re.match(r"Freigabe nötig im Workspace Plattform: README ergänzen \(demo_write\)\. "
|
|
r"https://aiw\.intern/approvals/\d+", texts[1])
|
|
setup["anna"].post(f"{base}/webhook", {"remove": "1"})
|
|
with db.session() as session:
|
|
assert session.scalar(select(__import__("app.workspaces.models", fromlist=["W"]).Workspace.webhook)) is None
|
|
|
|
|
|
def test_profil_und_admin_email(setup, db, people):
|
|
assert setup["ben"].post("/profile", {"email": "kein-mail"}).status_code == 400
|
|
page = setup["ben"].post("/profile", {"email": "Ben@Firma.de", "notify_me": "1"}).text
|
|
assert "Gespeichert." in page
|
|
with db.session() as session:
|
|
user = session.get(User, people["ben"])
|
|
assert user.email == "ben@firma.de" and user.notify
|
|
setup["anna"].post(f"/admin/users/{people['ben']}", {"display_name": "Ben", "active": "1",
|
|
"email": "b@firma.de"})
|
|
with db.session() as session:
|
|
assert session.get(User, people["ben"]).email == "b@firma.de"
|
|
assert 'href="/profile"' in setup["ben"].get("/workspaces").text
|
|
|
|
|
|
def test_benachrichtigungen_brauchen_csrf(setup):
|
|
for path in ["/profile", "/profile/test-email", f"/workspaces/{setup['id']}/webhook",
|
|
f"/workspaces/{setup['id']}/webhook/test"]:
|
|
assert setup["anna"].client.post(path, data={}, follow_redirects=False).status_code == 403, path
|