|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
sys.path.insert(0, ".")
|
|
import requests
|
|
from shared.otp_mailbox import FrappeOtpMailbox, OtpMailboxError, OtpMailboxTransientError, OtpRequest, MailboxBaseline
|
|
|
|
UTC = timezone.utc
|
|
IST = timezone(timedelta(hours=5, minutes=30))
|
|
fails = []
|
|
def check(name, cond, detail=""):
|
|
print(("PASS " if cond else "FAIL ") + name + (f" [{detail}]" if not cond and detail else ""))
|
|
if not cond: fails.append(name)
|
|
|
|
class Resp:
|
|
def __init__(self, data, status=200): self._d, self.status_code, self.ok, self.text, self.headers = data, status, status < 400, "x", {}
|
|
def json(self): return self._d
|
|
|
|
class Session:
|
|
"""request(): scripted behaviour per call; post/get unused."""
|
|
def __init__(self, script, docs): self.headers, self.script, self.docs, self.calls = {}, list(script), docs, 0
|
|
def request(self, method, url, **kw):
|
|
self.calls += 1
|
|
step = self.script.pop(0) if self.script else "ok"
|
|
if step == "timeout": raise requests.ReadTimeout("Read timed out. (read timeout=30)")
|
|
if step == "403": return Resp({}, 403)
|
|
return Resp({"data": list(self.docs)})
|
|
def post(self, *a, **k): return Resp({})
|
|
def get(self, *a, **k): return Resp({"message": {"time_zone": "Asia/Kolkata"}})
|
|
|
|
T0 = datetime(2026, 10, 14, 5, 50, 36, tzinfo=UTC)
|
|
now = [T0]
|
|
def doc(name, otp, sent):
|
|
body = f"{otp} is your One Time Password (OTP) of User ID user01 for verification."
|
|
return {"name": name, "subject": "FW: SEZOnline Authentication", "creation": sent.astimezone(IST).strftime("%Y-%m-%d %H:%M:%S.%f"),
|
|
"communication_date": sent.astimezone(IST).strftime("%Y-%m-%d %H:%M:%S"), "message_id": f"<{name}>", "sender": "m", "content": body, "text_content": body}
|
|
def box(session):
|
|
return FrappeOtpMailbox(base_url="https://x", api_key="k", api_secret="s", email_account="A", expected_user_id="user01",
|
|
frappe_time_zone="Asia/Kolkata", session=session, sleep=lambda s: None, clock=lambda: 0.0, utc_now=lambda: now[0], trigger_pull=False)
|
|
|
|
req = OtpRequest(sequence=1, baseline=MailboxBaseline()); req.stamp(T0)
|
|
good = doc("GOOD", "123456", T0 + timedelta(seconds=40))
|
|
now[0] = T0 + timedelta(seconds=60)
|
|
|
|
# 1. Two timeouts while waiting, then the mail is returned (before: the first timeout aborted the login)
|
|
s = Session(["timeout", "timeout"], [good])
|
|
b = box(s)
|
|
# the fake clock never advances, so give the loop one pass per call by ending after the mail is found
|
|
msg = None
|
|
for _ in range(5):
|
|
msg = b.wait_for_otp(req, timeout_seconds=10**6) if False else None
|
|
# drive the loop properly: deadline in the past after N polls
|
|
ticks = iter(range(0, 1000))
|
|
b2 = FrappeOtpMailbox(base_url="https://x", api_key="k", api_secret="s", email_account="A", expected_user_id="user01",
|
|
frappe_time_zone="Asia/Kolkata", session=Session(["timeout", "timeout"], [good]), sleep=lambda s: None,
|
|
clock=lambda: float(next(ticks)), utc_now=lambda: now[0], trigger_pull=False)
|
|
msg = b2.wait_for_otp(req, timeout_seconds=50)
|
|
check("wait survives 2 timeouts and then returns the mail", msg is not None and msg.otp == "123456")
|
|
|
|
# 2. Timeouts for the whole wait -> returns None (a clean 'no OTP mail'), not an exception
|
|
ticks2 = iter(range(0, 1000))
|
|
b3 = FrappeOtpMailbox(base_url="https://x", api_key="k", api_secret="s", email_account="A", expected_user_id="user01",
|
|
frappe_time_zone="Asia/Kolkata", session=Session(["timeout"] * 100, [good]), sleep=lambda s: None,
|
|
clock=lambda: float(next(ticks2)), utc_now=lambda: now[0], trigger_pull=False)
|
|
check("timeouts for the whole wait end as 'no mail' (None), not a crash", b3.wait_for_otp(req, timeout_seconds=20) is None)
|
|
|
|
# 3. A real error (403) is still fatal immediately
|
|
ticks3 = iter(range(0, 1000))
|
|
b4 = FrappeOtpMailbox(base_url="https://x", api_key="k", api_secret="s", email_account="A", expected_user_id="user01",
|
|
frappe_time_zone="Asia/Kolkata", session=Session(["403"], [good]), sleep=lambda s: None,
|
|
clock=lambda: float(next(ticks3)), utc_now=lambda: now[0], trigger_pull=False)
|
|
try:
|
|
b4.wait_for_otp(req, timeout_seconds=20); check("403 stays fatal", False)
|
|
except OtpMailboxError as e:
|
|
check("403 stays fatal (not retried)", not isinstance(e, OtpMailboxTransientError) and "403" in str(e))
|
|
|
|
# 4. Baseline snapshot retries a timeout then succeeds; gives up after 3 attempts
|
|
sb = Session(["timeout", "timeout"], [good])
|
|
check("baseline snapshot survives 2 timeouts", len(box(sb).prepare_request(1).baseline.names) == 1 and sb.calls == 3)
|
|
sb2 = Session(["timeout"] * 5, [good])
|
|
try:
|
|
box(sb2).prepare_request(1); check("baseline gives up after 3 timeouts", False)
|
|
except OtpMailboxTransientError:
|
|
check("baseline gives up after 3 timeouts", sb2.calls == 3)
|
|
|
|
# 5. configurable timeout is actually passed to requests
|
|
seen = {}
|
|
class S5(Session):
|
|
def request(self, method, url, **kw): seen["t"] = kw.get("timeout"); return super().request(method, url, **kw)
|
|
FrappeOtpMailbox(base_url="https://x", api_key="k", api_secret="s", email_account="A", frappe_time_zone="Asia/Kolkata",
|
|
session=S5([], [good]), sleep=lambda s: None, trigger_pull=False, request_timeout_seconds=90)._list_matching(["name"])
|
|
check("request timeout is configurable (90s used)", seen.get("t") == 90, str(seen))
|
|
|
|
print("ALL PASSED" if not fails else f"FAILED: {fails}"); sys.exit(1 if fails else 0)
|