From 76e4666b187b72bd2921b3c00d7ab8f0dc344741 Mon Sep 17 00:00:00 2001 From: joybytes <171517790+joybytes@users.noreply.github.com> Date: Thu, 19 Dec 2024 16:06:14 +0000 Subject: [PATCH] feat: Get most recent usage date for inbound number associated to service - Look through the following tables to get most recent date for inbound sms use: notification, inbound sms and inbound sms history - Endpoint created to showcase information on the front-end in regards to number activity to prevent us from going into the DB to check the latest use for the number when deciding to archive or release this number --- app/dao/inbound_sms_dao.py | 48 ++++++++++++++++++++++ app/inbound_sms/rest.py | 14 +++++++ tests/app/dao/test_inbound_sms_dao.py | 57 +++++++++++++++++++++++++++ tests/app/inbound_sms/test_rest.py | 36 +++++++++++++++++ 4 files changed, 155 insertions(+) diff --git a/app/dao/inbound_sms_dao.py b/app/dao/inbound_sms_dao.py index 8a7a1387c0..f851113d83 100644 --- a/app/dao/inbound_sms_dao.py +++ b/app/dao/inbound_sms_dao.py @@ -1,3 +1,6 @@ +from datetime import datetime +from uuid import UUID + from flask import current_app from sqlalchemy import and_, desc from sqlalchemy.dialects.postgresql import insert @@ -7,8 +10,10 @@ from app.constants import SMS_TYPE from app.dao.dao_utils import autocommit from app.models import ( + InboundNumber, InboundSms, InboundSmsHistory, + Notification, Service, ServiceDataRetention, ) @@ -179,3 +184,46 @@ def dao_get_paginated_most_recent_inbound_sms_by_user_number_for_service(service ) return q.paginate(page=page, per_page=current_app.config["PAGE_SIZE"]) + + +def dao_get_most_recent_inbound_usage_date(service_id: UUID, inbound: InboundNumber) -> datetime | None: + last_notification = ( + Notification.query.filter( + Notification.reply_to_text == inbound.number, + Notification.service_id == service_id, + ) + .order_by(Notification.created_at.desc()) + .first() + ) + + last_inbound_sms = ( + InboundSms.query.filter( + InboundSms.notify_number == inbound.number, + InboundSms.service_id == service_id, + ) + .order_by(InboundSms.created_at.desc()) + .first() + ) + + last_inbound_sms_history = ( + InboundSmsHistory.query.filter( + InboundSmsHistory.notify_number == inbound.number, + InboundSmsHistory.service_id == service_id, + ) + .order_by(InboundSmsHistory.created_at.desc()) + .first() + ) + + most_recent_usage = max( + filter( + None, + [ + last_notification.created_at if last_notification else None, + last_inbound_sms.created_at if last_inbound_sms else None, + last_inbound_sms_history.created_at if last_inbound_sms_history else None, + ], + ), + default=None, + ) + + return most_recent_usage diff --git a/app/inbound_sms/rest.py b/app/inbound_sms/rest.py index 2d439f5dc0..d8dffaddf8 100644 --- a/app/inbound_sms/rest.py +++ b/app/inbound_sms/rest.py @@ -2,12 +2,14 @@ from notifications_utils.recipient_validation.phone_number import try_validate_and_format_phone_number from app.dao.inbound_numbers_dao import ( + dao_get_inbound_number_for_service, dao_remove_inbound_sms_for_service, ) from app.dao.inbound_sms_dao import ( dao_count_inbound_sms_for_service, dao_get_inbound_sms_by_id, dao_get_inbound_sms_for_service, + dao_get_most_recent_inbound_usage_date, dao_get_paginated_most_recent_inbound_sms_by_user_number_for_service, ) from app.dao.service_data_retention_dao import fetch_service_data_retention_by_notification_type @@ -82,3 +84,15 @@ def remove_inbound_sms_for_service(service_id): except Exception as e: current_app.logger.error("error removing inbound SMS for service %s: %s", service_id, e) return jsonify({"message": str(e)}), 500 + + +@inbound_sms.route("/most-recent-usage", methods=["GET"]) +def get_most_recent_inbound_usage_date(service_id): + inbound = dao_get_inbound_number_for_service(service_id) + + if not inbound: + return jsonify(message="inbound not found"), 404 + + most_recent_date = dao_get_most_recent_inbound_usage_date(service_id, inbound) + + return jsonify(most_recent_date=most_recent_date), 200 diff --git a/tests/app/dao/test_inbound_sms_dao.py b/tests/app/dao/test_inbound_sms_dao.py index 35f1ecc565..80cc66367a 100644 --- a/tests/app/dao/test_inbound_sms_dao.py +++ b/tests/app/dao/test_inbound_sms_dao.py @@ -4,10 +4,13 @@ from freezegun import freeze_time from app import db +from app.constants import KEY_TYPE_NORMAL +from app.dao.inbound_numbers_dao import dao_get_inbound_number_for_service from app.dao.inbound_sms_dao import ( dao_count_inbound_sms_for_service, dao_get_inbound_sms_by_id, dao_get_inbound_sms_for_service, + dao_get_most_recent_inbound_usage_date, dao_get_paginated_inbound_sms_for_service_for_public_api, dao_get_paginated_most_recent_inbound_sms_by_user_number_for_service, delete_inbound_sms_older_than_retention, @@ -15,8 +18,11 @@ from app.models import InboundSmsHistory from tests.app.db import ( create_inbound_sms, + create_job, + create_notification, create_service, create_service_data_retention, + create_template, ) from tests.conftest import set_config @@ -353,3 +359,54 @@ def test_most_recent_inbound_sms_only_returns_values_within_7_days(sample_servic assert len(res.items) == 1 assert res.items[0].content == "new" + + +def test_dao_get_most_recent_inbound_usage_date_notifications_table(sample_service, sample_inbound_numbers): + template = create_template(service=sample_service) + job = create_job(template=template) + inbound = dao_get_inbound_number_for_service(sample_service.id) + + notification_time = datetime.utcnow() + + create_notification( + template=template, + job=job, + job_row_number=None, + to_field=None, + status="created", + reference=None, + created_at=notification_time, + sent_at=None, + billable_units=1, + personalisation=None, + api_key=None, + key_type=KEY_TYPE_NORMAL, + reply_to_text=inbound.number, + ) + + most_recent_date = dao_get_most_recent_inbound_usage_date(sample_service.id, inbound) + assert most_recent_date == notification_time + + +def test_dao_get_most_recent_inbound_usage_date_inbound_sms_table(sample_service, sample_inbound_numbers): + sms_time = datetime.utcnow() + create_inbound_sms(sample_service, created_at=sms_time) + inbound = dao_get_inbound_number_for_service(sample_service.id) + + most_recent_date = dao_get_most_recent_inbound_usage_date(sample_service.id, inbound) + assert most_recent_date == sms_time + + +def test_dao_get_most_recent_inbound_usage_date_inbound_sms_history_table( + sample_service, sample_inbound_numbers, sample_inbound_sms_history +): + inbound = next((x for x in sample_inbound_numbers if x.service_id == sample_service.id), None) + most_recent_date = dao_get_most_recent_inbound_usage_date(sample_service.id, inbound) + + assert most_recent_date is not None + assert most_recent_date.date() == sample_inbound_sms_history.created_at.date() + + +def test_dao_get_most_recent_inbound_usage_date_no_recent_notifications(sample_service, sample_inbound_numbers): + inbound = next((x for x in sample_inbound_numbers if x.service_id == sample_service.id), None) + assert dao_get_most_recent_inbound_usage_date(sample_service.id, inbound) is None diff --git a/tests/app/inbound_sms/test_rest.py b/tests/app/inbound_sms/test_rest.py index 57aa67ec6f..6bddd5bfe2 100644 --- a/tests/app/inbound_sms/test_rest.py +++ b/tests/app/inbound_sms/test_rest.py @@ -2,6 +2,7 @@ from datetime import datetime, timedelta import pytest +from dateutil.parser import parse from freezegun import freeze_time from app.constants import INBOUND_SMS_TYPE @@ -314,3 +315,38 @@ def test_remove_inbound_sms_for_service_success_without_inbound_number(admin_req _data={"archive": True}, _expected_status=200, ) + + +def test_get_most_recent_inbound_usage_date_success( + admin_request, sample_service, sample_inbound_numbers, sample_inbound_sms_history +): + response = admin_request.get( + "inbound_sms.get_most_recent_inbound_usage_date", service_id=sample_service.id, _expected_status=200 + ) + + assert response is not None + response_date = parse(response["most_recent_date"]) + + assert response_date.date() == datetime.utcnow().date() + + +def test_get_most_recent_inbound_usage_date_success_no_usage_found( + admin_request, sample_service, sample_inbound_numbers +): + response = admin_request.get( + "inbound_sms.get_most_recent_inbound_usage_date", service_id=sample_service.id, _expected_status=200 + ) + + assert response is not None + assert response["most_recent_date"] is None + + +def test_get_most_recent_inbound_usage_date_404_no_inbound( + admin_request, + sample_service, +): + response = admin_request.get( + "inbound_sms.get_most_recent_inbound_usage_date", service_id=sample_service.id, _expected_status=404 + ) + + assert response["message"] == "inbound not found"