Fixed bug with deconstruction

This commit is contained in:
alexjanousekGSA
2025-02-24 17:20:17 -05:00
parent aad562bb88
commit 896d26343c
2 changed files with 24 additions and 51 deletions

View File

@@ -905,25 +905,22 @@ def get_specific_hours_stats(data, start_date, hours=None, end_date=None, total_
else: else:
raise ValueError("Either hours or end_date must be set.") raise ValueError("Either hours or end_date must be set.")
# Ensure all hours exist in the output (even if empty)
grouped_data = {hour: [] for hour in gen_range} grouped_data = {hour: [] for hour in gen_range}
# Normalize timestamps and group notifications
for row in data: for row in data:
row_hour = row.timestamp.replace(minute=0, second=0, microsecond=0) # Normalize to full hour timestamp = row
row_hour = timestamp.replace(minute=0, second=0, microsecond=0)
if row_hour in grouped_data: if row_hour in grouped_data:
grouped_data[row_hour].append(row) grouped_data[row_hour].append(row)
# Ensure `total_notifications` is a dictionary
total_notifications = total_notifications or {}
# Format statistics, returning only hours with results # Format statistics, returning only hours with results
stats = { stats = {
hour.strftime("%Y-%m-%dT%H:00:00Z"): statistics.format_statistics( hour.strftime("%Y-%m-%dT%H:00:00Z"): statistics.format_statistics(
rows, rows,
total_notifications.get(hour, 0) total_notifications.get(hour, 0) if total_notifications else None
) )
for hour, rows in grouped_data.items() if rows # Only include hours with notifications for hour, rows in grouped_data.items() if rows
} }
return stats return stats

View File

@@ -1,5 +1,5 @@
from collections import namedtuple
from datetime import datetime from datetime import datetime
from unittest.mock import Mock
import pytest import pytest
@@ -7,15 +7,12 @@ from app.dao.services_dao import get_specific_hours_stats
from app.enums import StatisticsType from app.enums import StatisticsType
from app.models import TemplateType from app.models import TemplateType
NotificationRow = namedtuple("NotificationRow", ["notification_type", "status", "timestamp", "count"])
def generate_expected_hourly_output(requested_sms_hours): def generate_expected_hourly_output(requested_sms_hours):
""" return {
Generates expected output only for hours where notifications exist. hour: {
Removes empty hours from the output to match function behavior.
"""
output = {}
for hour in requested_sms_hours:
output[hour] = {
TemplateType.SMS: { TemplateType.SMS: {
StatisticsType.REQUESTED: 1, StatisticsType.REQUESTED: 1,
StatisticsType.DELIVERED: 0, StatisticsType.DELIVERED: 0,
@@ -29,23 +26,23 @@ def generate_expected_hourly_output(requested_sms_hours):
StatisticsType.PENDING: 0, StatisticsType.PENDING: 0,
}, },
} }
return output for hour in requested_sms_hours
}
def create_mock_notification(notification_type, status, timestamp, count=1): def create_mock_notification(notification_type, status, timestamp, count=1):
""" """
Creates a mock notification object with the required attributes. Creates a named tuple with the attributes required by format_statistics.
""" """
mock = Mock() return NotificationRow(
mock.notification_type = notification_type notification_type=notification_type,
mock.status = status status=status,
mock.timestamp = timestamp.replace(minute=0, second=0, microsecond=0) timestamp=timestamp.replace(minute=0, second=0, microsecond=0),
mock.count = count count=count
return mock )
test_cases = [ test_cases = [
# Single notification at 14:00 (Only 14:00 is expected in output)
( (
[create_mock_notification( [create_mock_notification(
TemplateType.SMS, TemplateType.SMS,
@@ -54,11 +51,8 @@ test_cases = [
)], )],
datetime(2025, 2, 18, 12, 0), datetime(2025, 2, 18, 12, 0),
6, 6,
generate_expected_hourly_output( generate_expected_hourly_output(["2025-02-18T14:00:00Z"]),
["2025-02-18T14:00:00Z"]
),
), ),
# Notification at 17:59 (Only 17:00 is expected in output)
( (
[create_mock_notification( [create_mock_notification(
TemplateType.SMS, TemplateType.SMS,
@@ -67,18 +61,9 @@ test_cases = [
)], )],
datetime(2025, 2, 18, 15, 0), datetime(2025, 2, 18, 15, 0),
3, 3,
generate_expected_hourly_output( generate_expected_hourly_output(["2025-02-18T17:00:00Z"]),
["2025-02-18T17:00:00Z"]
),
), ),
# No notifications at all (Expect empty `{}`) ([], datetime(2025, 2, 18, 10, 0), 4, {}),
(
[],
datetime(2025, 2, 18, 10, 0),
4,
{},
),
# Two notifications at 09:00 and 11:00 (Only those hours expected)
( (
[ [
create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 2, 18, 9, 30, 0)), create_mock_notification(TemplateType.SMS, StatisticsType.REQUESTED, datetime(2025, 2, 18, 9, 30, 0)),
@@ -86,25 +71,16 @@ test_cases = [
], ],
datetime(2025, 2, 18, 8, 0), datetime(2025, 2, 18, 8, 0),
5, 5,
generate_expected_hourly_output( generate_expected_hourly_output(["2025-02-18T09:00:00Z", "2025-02-18T11:00:00Z"]),
["2025-02-18T09:00:00Z", "2025-02-18T11:00:00Z"]
),
), ),
] ]
@pytest.mark.parametrize( @pytest.mark.parametrize("mocked_notifications, start_date, hours, expected_output", test_cases)
"mocked_notifications, start_date, hours, expected_output",
test_cases,
)
def test_get_specific_hours(mocked_notifications, start_date, hours, expected_output): def test_get_specific_hours(mocked_notifications, start_date, hours, expected_output):
"""
Tests get_specific_hours_stats to ensure it correctly aggregates hourly statistics.
"""
results = get_specific_hours_stats( results = get_specific_hours_stats(
mocked_notifications, mocked_notifications,
start_date, start_date,
hours=hours hours=hours
) )
assert results == expected_output, f"Expected {expected_output}, but got {results}" assert results == expected_output, f"Expected {expected_output}, but got {results}"