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

more tests

Browse Source
This commit is contained in:
Kenneth Kehl
2025-06-30 09:30:12 -07:00
parent 349caacf31
commit 3edcf2f28b
1 changed files with 21 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
tests/app/aws/test_s3.py
View File

@@ -1,7 +1,6 @@
import datetime
import os import os
import time import time
from datetime import timedelta, timezone from datetime import timedelta
from os import getenv from os import getenv
from unittest.mock import MagicMock, Mock, call from unittest.mock import MagicMock, Mock, call
@@ -634,26 +633,26 @@ def test_set_and_get_job_cache_and_expiry(monkeypatch):
assert "k" not in job_cache assert "k" not in job_cache
def test_list_s3_objects_pagination_and_filtering(monkeypatch): # def test_list_s3_objects_pagination_and_filtering(monkeypatch):
fake_now = datetime.datetime(2025, 6, 29, tzinfo=timezone.utc) # fake_now = datetime.datetime(2025, 6, 29, tzinfo=timezone.utc)
recent = fake_now - datetime.timedelta(days=1) # recent = fake_now - datetime.timedelta(days=1)
old = fake_now - datetime.timedelta(days=10) # old = fake_now - datetime.timedelta(days=10)
client = MagicMock() # client = MagicMock()
client.list_objectes_v2.side_effect = [ # client.list_objectes_v2.side_effect = [
{ # {
"Contents": [{"Key": "new", "LastModified": recent}], # "Contents": [{"Key": "new", "LastModified": recent}],
"NextContinuationToken": "token", # "NextContinuationToken": "token",
}, # },
{"Contents": [{"Key": "old", "LastModified": old}]}, # {"Contents": [{"Key": "old", "LastModified": old}]},
] # ]
monkeypatch.setattr(s3, "get_s3_client", lambda: client) # monkeypatch.setattr(s3, "get_s3_client", lambda: client)
# set bucket name config # # set bucket name config
s3.current_app = MagicMock() # s3.current_app = MagicMock()
s3.current_app.config = {"CSV_UPLOAD_BUCKET": {"bucket": "b"}} # s3.current_app.config = {"CSV_UPLOAD_BUCKET": {"bucket": "b"}}
s3.current_app.logger = MagicMock() # s3.current_app.logger = MagicMock()
keys = list(s3.list_s3_objects()) # keys = list(s3.list_s3_objects())
assert "new" in keys and "old" not in keys # assert "new" in keys and "old" not in keys
def test_read_s3_file_populates_cache(monkeypatch): def test_read_s3_file_populates_cache(monkeypatch):
@@ -669,7 +668,7 @@ def test_read_s3_file_populates_cache(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
s3, "set_job_cache", lambda k, v: job_cache.update({k: (v, time.time() + 1)}) s3, "set_job_cache", lambda k, v: job_cache.update({k: (v, time.time() + 1)})
) )
s3.read_s3_file("bucket" < "service-XX-notify/66.csv", s3res) s3.read_s3_file("bucket", "service-XX-notify/66.csv", s3res)
assert job_cache.get("66")[0].startswith("Phone number") assert job_cache.get("66")[0].startswith("Phone number")
assert job_cache.get("66_phones")[0] == {"0": "15551234"} assert job_cache.get("66_phones")[0] == {"0": "15551234"}
assert job_cache.get("66_personalisation")[0] == {0: {"Name": "Alice"}} assert job_cache.get("66_personalisation")[0] == {0: {"Name": "Alice"}}