darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-09-02 17:48:36 -04:00
Code Issues Packages Projects Releases Wiki Activity

Updated BE to allow for proper local to utc and back to local conversion

Browse Source
This commit is contained in:
alexjanousekGSA
2025-01-30 17:18:46 -05:00
parent d66a6bbd8c
commit 7374ac90a2
4 changed files with 200 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
app/dao/services_dao.py
View File

@@ -1,5 +1,7 @@
import uuid
from datetime import timedelta
import datetime
import pytz
from flask import current_app
from sqlalchemy import Float, cast, delete, select
@@ -734,31 +736,35 @@ def fetch_notification_stats_for_service_by_month_by_user(
return db.session.execute(stmt).all()
def get_specific_days_stats(data, start_date, days=None, end_date=None):
def get_specific_days_stats(data, start_date, days=None, end_date=None, timezone="UTC"):
user_timezone = pytz.timezone(timezone if timezone else "UTC")
if days is not None and end_date is not None:
raise ValueError("Only set days OR set end_date, not both.")
elif days is not None:
gen_range = generate_date_range(start_date, days=days)
date_range = list(generate_date_range(start_date, days=days))
elif end_date is not None:
gen_range = generate_date_range(start_date, end_date)
date_range = list(generate_date_range(start_date, end_date))
else:
raise ValueError("Either days or end_date must be set.")
for item in data:
print(
f"DEBUG12345 - Timestamp Check: {item.timestamp.isoformat()} {item.status.value})"
)
if not isinstance(item.timestamp, datetime.datetime):
print(f"ERROR: Found non-datetime value: {item.timestamp} ({type(item.timestamp)})")
grouped_data = {date: [] for date in gen_range} | {
timestamp.date(): [
row for row in data if row.timestamp.date() == timestamp.date()
]
for timestamp in {item.timestamp for item in data}
}
# Group data by date, ensuring ALL dates in range are included
grouped_data = {date: [] for date in date_range}
for item in data:
local_datetime = item.timestamp.replace(tzinfo=pytz.utc).astimezone(user_timezone)
local_date = local_datetime.date()
grouped_data[local_date].append(item)
# Ensure all dates exist in the final output, even if empty
stats = {
day.strftime("%Y-%m-%d"): statistics.format_statistics(rows)
for day, rows in grouped_data.items()
day.strftime("%Y-%m-%d"): statistics.format_statistics(grouped_data.get(day, []))
for day in date_range
}
return stats

49
app/service/rest.py
View File

@@ -221,50 +221,45 @@ def get_service_notification_statistics(service_id):
@service_blueprint.route("/<uuid:service_id>/statistics/<string:start>/<int:days>")
def get_service_notification_statistics_by_day(service_id, start, days):
# Allow timezone override, default to UTC
user_timezone = request.args.get("timezone", "UTC")
return jsonify(
data=get_service_statistics_for_specific_days(service_id, start, int(days))
data=get_service_statistics_for_specific_days(service_id, start, int(days), user_timezone)
)
def convert_local_to_utc(local_dt, user_timezone="America/New_York"):
local_tz = pytz.timezone(user_timezone)
localized_dt = local_tz.localize(local_dt)
local_timezone = pytz.timezone(user_timezone)
if local_dt.tzinfo is None:
localized_dt = local_timezone.localize(local_dt)
else:
localized_dt = local_dt.astimezone(local_timezone)
return localized_dt.astimezone(pytz.utc)
def get_service_statistics_for_specific_days(service_id, start, days=7):
print(f"DEBUG - Received start: {start}, days: {days}")
user_timezone = "America/New_York"
local_end_date = datetime.strptime(start, "%Y-%m-%d")
def get_service_statistics_for_specific_days(service_id, start, days=7, timezone="UTC"):
user_timezone = pytz.timezone(timezone)
local_end_date = datetime.strptime(start, "%Y-%m-%d").replace(tzinfo=user_timezone)
local_start_date = local_end_date - timedelta(days=days - 1)
utc_start_date = convert_local_to_utc(
local_start_date.replace(hour=0, minute=0, second=0), user_timezone
)
utc_end_date = convert_local_to_utc(
local_end_date.replace(hour=23, minute=59, second=59), user_timezone
)
utc_start_date = local_start_date.astimezone(pytz.utc).replace(hour=0, minute=0, second=0)
utc_end_date = local_end_date.astimezone(pytz.utc).replace(hour=23, minute=59, second=59)
now_local = (
datetime.utcnow()
.replace(tzinfo=pytz.utc)
.astimezone(pytz.timezone(user_timezone))
)
if now_local.hour >= 19:
now_utc = datetime.utcnow().replace(tzinfo=pytz.utc)
now_local = now_utc.astimezone(user_timezone)
# If the local day hasn't fully ended in UTC yet, extend the end date to the next day
if now_local.date() == local_end_date.date() and now_utc.date() > now_local.date():
utc_end_date += timedelta(days=1)
print(f"DEBUG - Querying db from {utc_start_date} UTC to {utc_end_date} UTC")
results = dao_fetch_stats_for_service_from_days(
service_id, utc_start_date, utc_end_date
)
stats = get_specific_days_stats(results, utc_start_date, days=days)
results = dao_fetch_stats_for_service_from_days(service_id, utc_start_date, utc_end_date)
stats = get_specific_days_stats(results, utc_start_date, days=days, timezone=timezone)
return stats
@service_blueprint.route(
"/<uuid:service_id>/statistics/user/<uuid:user_id>/<string:start>/<int:days>"
)

183
tests/app/dao/test_services_dao.py
View File

@@ -1,7 +1,7 @@
import uuid
from datetime import datetime, timedelta
from unittest import mock
# from unittest.mock import Mock
from unittest.mock import Mock
import pytest
import sqlalchemy
@@ -40,7 +40,7 @@ from app.dao.services_dao import (
delete_service_and_all_associated_db_objects,
get_live_services_with_organization,
get_services_by_partial_name,
# get_specific_days_stats,
get_specific_days_stats,
)
from app.dao.users_dao import create_user_code, save_model_user
from app.enums import (
@@ -51,7 +51,7 @@ from app.enums import (
OrganizationType,
PermissionType,
ServicePermissionType,
# StatisticsType,
StatisticsType,
TemplateType,
)
from app.models import (
@@ -1625,180 +1625,3 @@ def test_get_live_services_with_organization(sample_organization):
(live_service.name, sample_organization.name),
(service_without_org.name, None),
]
_this_date = utc_now() - timedelta(days=4)
# @pytest.mark.parametrize(
# ["data", "start_date", "days", "end_date", "expected", "is_error"],
# [
# [None, _this_date, None, None, None, True],
# [None, _this_date, 4, _this_date - timedelta(4), None, True],
# [
# [
# {"day": _this_date, "something": "else"},
# {"day": _this_date, "something": "new"},
# {"day": _this_date + timedelta(days=1), "something": "borrowed"},
# {"day": _this_date + timedelta(days=2), "something": "old"},
# {"day": _this_date + timedelta(days=4), "something": "blue"},
# ],
# _this_date,
# 4,
# None,
# {
# _this_date.date().strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 2,
# },
# },
# (_this_date.date() + timedelta(days=1)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 1,
# },
# },
# (_this_date.date() + timedelta(days=2)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 1,
# },
# },
# (_this_date.date() + timedelta(days=3)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# },
# (_this_date.date() + timedelta(days=4)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 1,
# },
# },
# },
# False,
# ],
# [
# [
# {"day": _this_date, "something": "else"},
# {"day": _this_date, "something": "new"},
# {"day": _this_date + timedelta(days=1), "something": "borrowed"},
# {"day": _this_date + timedelta(days=2), "something": "old"},
# {"day": _this_date + timedelta(days=4), "something": "blue"},
# ],
# _this_date,
# None,
# _this_date + timedelta(4),
# {
# _this_date.date().strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 2,
# },
# },
# (_this_date.date() + timedelta(days=1)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 1,
# },
# },
# (_this_date.date() + timedelta(days=2)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 1,
# },
# },
# (_this_date.date() + timedelta(days=3)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# },
# (_this_date.date() + timedelta(days=4)).strftime("%Y-%m-%d"): {
# TemplateType.EMAIL: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 0,
# },
# TemplateType.SMS: {
# StatisticsType.DELIVERED: 0,
# StatisticsType.FAILURE: 0,
# StatisticsType.REQUESTED: 1,
# },
# },
# },
# False,
# ],
# ],
# )
# def test_get_specific_days(data, start_date, days, end_date, expected, is_error):
# if is_error:
# with pytest.raises(ValueError):
# get_specific_days_stats(data, start_date, days, end_date)
# else:
# new_data = []
# for line in data:
# new_line = Mock()
# new_line.day = line["day"]
# new_line.notification_type = NotificationType.SMS
# new_line.count = 1
# new_line.something = line["something"]
# new_data.append(new_line)
# results = get_specific_days_stats(new_data, start_date, days, end_date)
# assert results == expected

155
tests/app/dao/test_services_get_specific_days.py Normal file
View File

@@ -0,0 +1,155 @@
import pytest
from datetime import datetime
import pytz
from unittest.mock import Mock
from app.dao.services_dao import get_specific_days_stats
from app.enums import StatisticsType
from app.models import TemplateType
def generate_expected_output(requested_days, requested_sms_days):
output = {}
for day in requested_days:
output[day] = {
TemplateType.SMS: {
StatisticsType.REQUESTED: 1 if day in requested_sms_days else 0,
StatisticsType.DELIVERED: 0,
StatisticsType.FAILURE: 0
},
TemplateType.EMAIL: {
StatisticsType.REQUESTED: 0,
StatisticsType.DELIVERED: 0,
StatisticsType.FAILURE: 0
}
}
return output
def create_mock_notification(notification_type, status, timestamp, count=1):
return Mock(
notification_type=notification_type,
status=status,
timestamp=timestamp,
count=count
)
test_cases = [
# Case with normal dates that don't carry over to next day when converted to UTC
(
[create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 1, 29, 1, 20, 18, tzinfo=pytz.utc))],
datetime(2025, 1, 28, tzinfo=pytz.utc),
2,
"America/New_York",
generate_expected_output(["2025-01-28", "2025-01-29"], ["2025-01-28"])
),
# Case where EST is saved as next day in UTC, it needs to be converted back to EST after retrieval
(
[create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 1, 30, 4, 30, 0, tzinfo=pytz.utc))],
datetime(2025, 1, 29, tzinfo=pytz.utc),
2,
"America/New_York",
generate_expected_output(["2025-01-29", "2025-01-30"], ["2025-01-29"])
),
# Case where UTC is queryed
(
[create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 1, 29, 10, 15, 0, tzinfo=pytz.utc))],
datetime(2025, 1, 28, tzinfo=pytz.utc),
2,
"UTC",
generate_expected_output(["2025-01-28", "2025-01-29"], ["2025-01-29"])
),
# Central time test
(
[create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 1, 29, 3, 0, 0, tzinfo=pytz.utc))],
datetime(2025, 1, 28, tzinfo=pytz.utc),
2,
"America/Chicago",
generate_expected_output(["2025-01-28", "2025-01-29"], ["2025-01-28"])
),
# Mountain time test
(
[create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 1, 29, 5, 0, 0, tzinfo=pytz.utc))],
datetime(2025, 1, 28, tzinfo=pytz.utc),
2,
"America/Denver",
generate_expected_output(["2025-01-28", "2025-01-29"], ["2025-01-28"])
),
# Pacific time test
(
[create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 1, 29, 7, 30, 0, tzinfo=pytz.utc))],
datetime(2025, 1, 28, tzinfo=pytz.utc),
2,
"America/Los_Angeles",
generate_expected_output(["2025-01-28", "2025-01-29"], ["2025-01-28"])
),
# Case where no timezone is provided, ensuring it defaults to UTC
(
[create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 1, 29, 10, 15, 0, tzinfo=pytz.utc))],
datetime(2025, 1, 28, tzinfo=pytz.utc),
2,
None, # No timezone passed
generate_expected_output(["2025-01-28", "2025-01-29"], ["2025-01-29"])
),
# Daylights savings time Spring Forward: March 10, 2024 at 2 AM EST => jumps to 3 AM EST
(
[create_mock_notification(
TemplateType.SMS,
StatisticsType.REQUESTED,
datetime(2024, 3, 10, 6, 30, 0, tzinfo=pytz.utc) # UTC 6:30 AM => 1:30 AM EST (before DST kicks in)
)],
datetime(2024, 3, 9, tzinfo=pytz.utc),
2,
"America/New_York",
generate_expected_output(["2024-03-09", "2024-03-10"], ["2024-03-10"]) # Should map to March 10 EST
),
# Fall Back: November 3, 2024 at 2 AM EDT => goes back to 1 AM EST
(
[create_mock_notification(
TemplateType.SMS,
StatisticsType.REQUESTED,
datetime(2024, 11, 3, 5, 30, 0, tzinfo=pytz.utc) # UTC 5:30 AM => 1:30 AM EST after DST ends
)],
datetime(2024, 11, 2, tzinfo=pytz.utc),
2,
"America/New_York",
generate_expected_output(["2024-11-02", "2024-11-03"], ["2024-11-03"])
),
# There are no notifications
(
[],
datetime(2025, 1, 29, tzinfo=pytz.utc),
2,
"UTC",
generate_expected_output(["2025-01-29", "2025-01-30"], [])
),
# Midnight edge case
(
[create_mock_notification(
TemplateType.SMS,
StatisticsType.REQUESTED,
datetime(2025, 1, 10, 0, 0, 0, tzinfo=pytz.utc) # 12:00 AM UTC => 7:00 PM Jan 9 EST
)],
datetime(2025, 1, 9, tzinfo=pytz.utc),
2,
"America/New_York",
generate_expected_output(["2025-01-09", "2025-01-10"], ["2025-01-09"]) # Goes back to Jan 9 EST
),
# Large query testing large amounts of data
(
[create_mock_notification(
TemplateType.SMS,
StatisticsType.REQUESTED,
datetime(2025, 1, 15, 12, 0, 0, tzinfo=pytz.utc)
)],
datetime(2025, 1, 1, tzinfo=pytz.utc),
30,
"America/New_York",
generate_expected_output(
[f"2025-01-{str(day).zfill(2)}" for day in range(1, 31)],
["2025-01-15"]
)
)
]
@pytest.mark.parametrize("mocked_notifications, start_date, days, timezone, expected_output", test_cases)
def test_get_specific_days(mocked_notifications, start_date, days, timezone, expected_output):
results = get_specific_days_stats(mocked_notifications, start_date, days, timezone=timezone)
assert results == expected_output