100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify SMS and OTP logging pipelines write to the database."""
|
|
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.database import SessionLocal
|
|
from app.log_service import record_otp_log, record_sms_logs
|
|
from app.message_send_service import RecipientSendResult, SendMessageError, send_template_messages
|
|
from app.models import OtpMessageLog, SenderConfig, SmsMessageLog, TemplateConfig, UrmiOtpProject, UserAccount
|
|
|
|
|
|
def main() -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
user = db.scalar(select(UserAccount).limit(1))
|
|
if not user:
|
|
print("SKIP: no user account")
|
|
return
|
|
|
|
sms_before = db.scalar(select(func.count()).select_from(SmsMessageLog)) or 0
|
|
otp_before = db.scalar(select(func.count()).select_from(OtpMessageLog)) or 0
|
|
|
|
sender = db.scalar(
|
|
select(SenderConfig).where(
|
|
SenderConfig.user_id == user.id,
|
|
SenderConfig.status == "approved",
|
|
)
|
|
)
|
|
template = db.scalar(
|
|
select(TemplateConfig).where(
|
|
TemplateConfig.user_id == user.id,
|
|
TemplateConfig.status == "approved",
|
|
)
|
|
)
|
|
if sender and template:
|
|
try:
|
|
send_template_messages(
|
|
db,
|
|
user,
|
|
sender_id=sender.sender_id,
|
|
template_id=template.id,
|
|
message=template.template_text.replace("{#var#}", "123456", 1).replace("{#var#}", "10", 1)
|
|
if "{#var#}" in template.template_text
|
|
else template.template_text,
|
|
recipients=["919999999999"],
|
|
)
|
|
except SendMessageError as exc:
|
|
print(f"SMS send path exercised (gateway may reject): {exc.detail}")
|
|
else:
|
|
record_sms_logs(
|
|
db,
|
|
user,
|
|
sender_id="TESTDLT",
|
|
template_name="TEST_TEMPLATE",
|
|
template_id=uuid4(),
|
|
results=[
|
|
RecipientSendResult(
|
|
to="919999999999",
|
|
success=True,
|
|
job_id="test-job",
|
|
message_id="test-msg",
|
|
)
|
|
],
|
|
)
|
|
print("SMS log inserted via record_sms_logs fallback")
|
|
|
|
project = db.scalar(
|
|
select(UrmiOtpProject).where(
|
|
UrmiOtpProject.user_id == user.id,
|
|
UrmiOtpProject.status == "approved",
|
|
)
|
|
)
|
|
record_otp_log(
|
|
db,
|
|
user,
|
|
project_name=project.project_name if project else "test-project",
|
|
template_type="login_authentication",
|
|
to_number="919999999999",
|
|
status="accepted",
|
|
)
|
|
print("OTP log inserted")
|
|
|
|
sms_after = db.scalar(select(func.count()).select_from(SmsMessageLog)) or 0
|
|
otp_after = db.scalar(select(func.count()).select_from(OtpMessageLog)) or 0
|
|
|
|
assert sms_after > sms_before, "SMS log count did not increase"
|
|
assert otp_after > otp_before, "OTP log count did not increase"
|
|
|
|
latest_sms = db.scalar(select(SmsMessageLog).order_by(SmsMessageLog.created_at.desc()).limit(1))
|
|
latest_otp = db.scalar(select(OtpMessageLog).order_by(OtpMessageLog.created_at.desc()).limit(1))
|
|
print(f"OK sms_logs={sms_after} latest={latest_sms.sender_id}->{latest_sms.to_number} status={latest_sms.status}")
|
|
print(f"OK otp_logs={otp_after} latest={latest_otp.project_name}->{latest_otp.to_number} type={latest_otp.template_type}")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|