ManikandanandClaude Opus 5.5
d500f750da
Snapshot of D:\Ecw Process a11c9ff (2026-10-30). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
131 lines
3.8 KiB
Python
131 lines
3.8 KiB
Python
r"""
|
|
core/notify.py — the agent's email to the operator (operator, 2026-10-30):
|
|
what happened, what the agent did about it, and whether it recovered.
|
|
|
|
Rules the operator set:
|
|
* sent from the bot's own mailbox (SMTP, the 2FA mailbox's account);
|
|
* screenshots of login and error pages only — never a claim screen, never
|
|
page HTML: those carry PHI;
|
|
* no claim numbers in the mail.
|
|
|
|
python tests\test_notify.py
|
|
|
|
No mail is sent: the SMTP transport is a stand-in.
|
|
"""
|
|
|
|
import sys
|
|
import pathlib
|
|
import tempfile
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "src"))
|
|
|
|
from ecw_bot import settings as cfg # noqa: E402
|
|
from ecw_bot.core import notify # noqa: E402
|
|
|
|
CASES = []
|
|
|
|
|
|
def case(fn):
|
|
CASES.append(fn)
|
|
return fn
|
|
|
|
|
|
def _capture():
|
|
sent = []
|
|
notify._transport = lambda msg: sent.append(msg)
|
|
return sent
|
|
|
|
|
|
def _files(tmp, *names):
|
|
out = []
|
|
for n in names:
|
|
p = pathlib.Path(tmp) / n
|
|
p.write_bytes(b"x")
|
|
out.append(p)
|
|
return out
|
|
|
|
|
|
@case
|
|
def an_incident_says_what_happened_and_what_the_agent_did(tmp):
|
|
sent = _capture()
|
|
assert notify.incident("Login failed", "2FA mail came 7 minutes late",
|
|
"Resent the OTP, will retry in 10 minutes",
|
|
recovered=False) is True
|
|
msg = sent[0]
|
|
assert msg["To"] == cfg.NOTIFY_EMAIL_TO
|
|
assert "Login failed" in msg["Subject"]
|
|
body = msg.get_body(("plain",)).get_content()
|
|
assert "2FA mail came 7 minutes late" in body
|
|
assert "Resent the OTP" in body
|
|
assert "Not recovered" in body
|
|
|
|
|
|
@case
|
|
def claim_numbers_never_reach_the_mail(tmp):
|
|
sent = _capture()
|
|
notify.incident("Stuck dialog on claim 9107000",
|
|
"Claim 9107000 form would not close", "Reloaded eCW",
|
|
recovered=True)
|
|
msg = sent[0]
|
|
text = msg["Subject"] + msg.get_body(("plain",)).get_content()
|
|
assert "9107000" not in text, text
|
|
|
|
|
|
@case
|
|
def only_login_and_error_screenshots_are_attached(tmp):
|
|
sent = _capture()
|
|
shots = _files(tmp, "211944_login_unconfirmed.png",
|
|
"9107000_claim_form_leftover_204500.png",
|
|
"211944_login_unconfirmed.html",
|
|
"101500_security_image_page.png")
|
|
notify.incident("Login failed", "x", "y", recovered=False, shots=shots)
|
|
names = sorted(a.get_filename() for a in sent[0].iter_attachments())
|
|
assert names == ["101500_security_image_page.png",
|
|
"211944_login_unconfirmed.png"], names
|
|
|
|
|
|
@case
|
|
def nothing_is_sent_when_no_recipient_is_set(tmp):
|
|
sent = _capture()
|
|
old = cfg.NOTIFY_EMAIL_TO
|
|
cfg.NOTIFY_EMAIL_TO = ""
|
|
try:
|
|
assert notify.incident("x", "y", "z", recovered=True) is False
|
|
finally:
|
|
cfg.NOTIFY_EMAIL_TO = old
|
|
assert sent == []
|
|
|
|
|
|
@case
|
|
def a_mail_server_error_never_stops_the_run(tmp):
|
|
def boom(msg):
|
|
raise OSError("connection refused")
|
|
notify._transport = boom
|
|
assert notify.incident("x", "y", "z", recovered=True) is False
|
|
|
|
|
|
def main() -> int:
|
|
failures = 0
|
|
real = notify._transport
|
|
cfg.NOTIFY_EMAIL_TO = "operator@example.test"
|
|
for fn in CASES:
|
|
name = fn.__name__.replace("_", " ")
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
try:
|
|
fn(tmp)
|
|
print(f" ok {name}")
|
|
except AssertionError as exc:
|
|
print(f" FAIL {name}: {exc}")
|
|
failures += 1
|
|
except Exception as exc:
|
|
print(f" ERROR {name}: {type(exc).__name__}: {exc}")
|
|
failures += 1
|
|
finally:
|
|
notify._transport = real
|
|
print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|