darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-12 16:22:17 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fixed test_get_notification.py tests

Browse Source
This commit is contained in:
alexjanousekGSA
2025-05-14 15:01:01 -04:00
parent b7d87b6db8
commit a100fa6eb8
5 changed files with 66 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
app/notifications/rest.py
View File

@@ -15,9 +15,9 @@ from app.notifications.validators import (
service_has_permission,
validate_template,
)
from app.public_schemas.public import PublicNotificationResponseSchema
from app.schemas import (
email_notification_schema,
notification_with_personalisation_schema,
notifications_filter_schema,
sms_template_notification_schema,
)
@@ -50,7 +50,7 @@ def get_notification_by_id(notification_id):
notification.to = recipient
notification.normalised_to = recipient
serialized = notification_with_personalisation_schema.dump(notification)
serialized = PublicNotificationResponseSchema().dump(notification)
return jsonify(data={"notification": serialized}), 200
@@ -93,7 +93,7 @@ def get_all_notifications():
notification.normalised_to = recipient
result = jsonify(
notifications=notification_with_personalisation_schema.dump(
notifications=PublicNotificationResponseSchema().dump(
pagination.items, many=True
),
page_size=page_size,

86
app/public_schemas/public.py
View File

@@ -1,7 +1,10 @@
from datetime import timezone
from datetime import timezone, datetime
from uuid import UUID
from marshmallow import Schema, fields, pre_dump
from marshmallow import EXCLUDE, Schema, fields, post_dump, pre_dump
from app.schemas import FlexibleDateTime, JobSchema, TemplateSchema
from app import ma
class PublicTemplateSchema(Schema):
id = fields.UUID(required=True)
@@ -35,36 +38,63 @@ class PublicNotificationSchema(Schema):
body = fields.String(required=True)
content_char_count = fields.Integer(required=True)
@pre_dump
def transform(self, notification, **kwargs):
class PublicNotificationResponseSchema(PublicNotificationSchema):
class Meta:
unknown = EXCLUDE
@post_dump
def transform(self, data, **kwargs):
def to_rfc3339(dt):
if dt is None:
return None
if isinstance(dt, str):
try:
dt = datetime.fromisoformat(dt)
except ValueError:
return dt # fallback, might already be valid
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.isoformat().replace("+00:00", "Z")
return {
**notification.__dict__,
"created_at": to_rfc3339(getattr(notification, "created_at", None)),
"sent_at": to_rfc3339(getattr(notification, "sent_at", None)),
"updated_at": to_rfc3339(getattr(notification, "updated_at", None)),
"sent_by": getattr(notification, "sent_by", None),
"reference": getattr(notification, "reference", None),
"service": str(notification.service.id) if notification.service else None,
"api_key": str(notification.api_key.id) if notification.api_key else None,
"body": getattr(notification, "body", None)
or (notification.template.content if notification.template else ""),
"content_char_count": len(
getattr(notification, "body", "")
or (notification.template.content if notification.template else "")
),
"job": (
{
"id": str(notification.job.id),
"original_file_name": notification.job.original_file_name,
}
if hasattr(notification, "job") and notification.job
else None
),
}
data["created_at"] = to_rfc3339(data.get("created_at"))
data["sent_at"] = to_rfc3339(data.get("sent_at"))
data["updated_at"] = to_rfc3339(data.get("updated_at"))
# Fallback content
template = data.get("template", {})
body = data.get("body") or (template.get("content") if isinstance(template, dict) else "")
data["body"] = body or ""
data["content_char_count"] = len(data["body"])
# Extract UUID string for service
service = data.get("service")
if hasattr(service, "id"):
data["service"] = str(service.id)
elif isinstance(service, UUID):
data["service"] = str(service)
elif isinstance(service, str) and service.startswith("<Service "):
# fallback if __str__ was called on the SQLAlchemy object
data["service"] = service.split("<Service ")[1].rstrip(">")
else:
data["service"] = str(service) # best effort fallback
# Extract UUID string for api_key
api_key = data.get("api_key")
if hasattr(api_key, "id"):
data["api_key"] = str(api_key.id)
elif isinstance(api_key, UUID):
data["api_key"] = str(api_key)
elif isinstance(api_key, str) and api_key.startswith("<ApiKey "):
data["api_key"] = api_key.split("<ApiKey ")[1].rstrip(">")
else:
data["api_key"] = str(api_key) if api_key else None
# Fix job dict
job = data.get("job")
if isinstance(job, dict) and "id" in job:
job_id = job.get("id")
job["id"] = str(job_id) if job_id else None
data["job"] = job
return data

59
app/schemas.py
View File

@@ -603,64 +603,6 @@ class NotificationWithTemplateSchema(BaseSchema):
return in_data
class NotificationWithPersonalisationSchema(NotificationWithTemplateSchema):
template_history = fields.Nested(
TemplateHistorySchema,
attribute="template",
only=["id", "name", "template_type", "content", "subject", "version"],
dump_only=True,
)
# Mark as many fields as possible as required since this is a public api.
# WARNING: Does _not_ reference fields computed in handle_template_merge, such as
# 'body', 'subject' [for emails], and 'content_char_count'
# db rows
billable_units = auto_field()
created_at = auto_field()
id = auto_field()
job_row_number = auto_field()
notification_type = auto_field(by_value=True)
reference = auto_field()
sent_at = auto_field()
sent_by = auto_field()
status = auto_field(by_value=True)
template_version = auto_field()
to = auto_field()
updated_at = auto_field()
class Meta(NotificationWithTemplateSchema.Meta):
# Overwrite the `NotificationWithTemplateSchema` base class to not exclude `_personalisation`, which
# isn't a defined field for this class
exclude = ()
@pre_dump
def handle_personalisation_property(self, in_data, **kwargs):
in_data._merged_personalisation = in_data.personalisation
return in_data
@post_dump(pass_original=True)
def handle_template_merge(self, in_data, original_obj, **kwargs):
personalisation = getattr(original_obj, "_merged_personalisation", None)
in_data["template"] = in_data.pop("template_history")
template = get_template_instance(in_data["template"], personalisation)
in_data["body"] = template.content_with_placeholders_filled_in
if in_data["template"]["template_type"] != TemplateType.SMS:
in_data["subject"] = template.subject
in_data["content_char_count"] = None
else:
in_data["content_char_count"] = template.content_count
in_data["template"].pop("content", None)
in_data["template"].pop("subject", None)
in_data.pop("personalisation", None)
return in_data
class InvitedUserSchema(BaseSchema):
auth_type = auto_field(by_value=True)
created_at = FlexibleDateTime()
@@ -812,7 +754,6 @@ sms_template_notification_schema = SmsTemplateNotificationSchema()
email_notification_schema = EmailNotificationSchema()
notification_schema = NotificationModelSchema()
notification_with_template_schema = NotificationWithTemplateSchema()
notification_with_personalisation_schema = NotificationWithPersonalisationSchema()
invited_user_schema = InvitedUserSchema()
email_data_request_schema = EmailDataSchema()
partial_email_data_request_schema = EmailDataSchema(partial_email=True)

32
tests/app/public_contracts/test_GET_notification.py
View File

@@ -24,8 +24,6 @@ def _get_notification(client, notification, url):
# v0
def test_get_api_sms_contract(client, sample_notification):
response_json = return_json_from_response(
_get_notification(
@@ -37,18 +35,6 @@ def test_get_api_sms_contract(client, sample_notification):
validate_v0(response_json, "GET_notification_return_sms.json")
@pytest.mark.skip(reason="Update to fetch email from s3")
def test_get_api_email_contract(client, sample_email_notification):
response_json = return_json_from_response(
_get_notification(
client,
sample_email_notification,
"/notifications/{}".format(sample_email_notification.id),
)
)
validate_v0(response_json, "GET_notification_return_email.json")
def test_get_job_sms_contract(client, sample_notification):
response_json = return_json_from_response(
_get_notification(
@@ -60,22 +46,12 @@ def test_get_job_sms_contract(client, sample_notification):
validate_v0(response_json, "GET_notification_return_sms.json")
@pytest.mark.skip(reason="Update to fetch email from s3")
def test_get_job_email_contract(client, sample_email_notification):
response_json = return_json_from_response(
_get_notification(
client,
sample_email_notification,
"/notifications/{}".format(sample_email_notification.id),
)
)
validate_v0(response_json, "GET_notification_return_email.json")
def test_get_notifications_contract(
client, sample_notification, sample_email_notification
):
def test_get_notifications_contract(client, sample_notification, sample_email_notification):
response_json = return_json_from_response(
_get_notification(client, sample_notification, "/notifications")
)
notifications = response_json["notifications"]
assert notifications, "No notifications returned"
assert notifications[0]["template"]["template_type"] == "sms"
validate_v0(response_json, "GET_notifications_return.json")

2
tests/app/test_schemas.py
View File

@@ -48,7 +48,7 @@ def test_notification_schema_adds_api_key_name(sample_notification):
"notification_with_template_schema",
"notification_schema",
"notification_with_template_schema",
"notification_with_personalisation_schema",
"public_notification_response_schema",
],
)
def test_notification_schema_has_correct_status(sample_notification, schema_name):