Expose object modification evidence
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
@@ -26,10 +27,12 @@ class _S3Error(RuntimeError):
|
||||
class _FakeS3Client:
|
||||
def __init__(self) -> None:
|
||||
self.objects: dict[str, bytes] = {}
|
||||
self.modified_at: dict[str, datetime] = {}
|
||||
self.head_error: Exception | None = None
|
||||
|
||||
def put_object(self, *, Key: str, Body: bytes, **_kwargs) -> None:
|
||||
self.objects[Key] = Body
|
||||
self.modified_at[Key] = datetime.now(timezone.utc)
|
||||
|
||||
def get_object(self, *, Key: str, **_kwargs):
|
||||
try:
|
||||
@@ -45,10 +48,40 @@ class _FakeS3Client:
|
||||
payload = self.objects[Key]
|
||||
except KeyError as exc:
|
||||
raise _S3Error("NotFound", 404) from exc
|
||||
return {"ContentLength": len(payload)}
|
||||
return {
|
||||
"ContentLength": len(payload),
|
||||
"LastModified": self.modified_at[Key],
|
||||
}
|
||||
|
||||
def delete_object(self, *, Key: str, **_kwargs) -> None:
|
||||
self.objects.pop(Key, None)
|
||||
self.modified_at.pop(Key, None)
|
||||
|
||||
def list_objects_v2(
|
||||
self,
|
||||
*,
|
||||
Prefix: str,
|
||||
MaxKeys: int,
|
||||
StartAfter: str | None = None,
|
||||
**_kwargs,
|
||||
):
|
||||
keys = [
|
||||
key
|
||||
for key in sorted(self.objects)
|
||||
if key.startswith(Prefix) and (StartAfter is None or key > StartAfter)
|
||||
]
|
||||
selected = keys[:MaxKeys]
|
||||
return {
|
||||
"Contents": [
|
||||
{
|
||||
"Key": key,
|
||||
"Size": len(self.objects[key]),
|
||||
"LastModified": self.modified_at[key],
|
||||
}
|
||||
for key in selected
|
||||
],
|
||||
"IsTruncated": len(keys) > len(selected),
|
||||
}
|
||||
|
||||
|
||||
class ObjectStorageTests(unittest.TestCase):
|
||||
@@ -67,6 +100,7 @@ class ObjectStorageTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(b"a", backend.get_bytes("campaign/a.eml"))
|
||||
self.assertEqual(1, backend.stat("campaign/a.eml").size_bytes)
|
||||
self.assertIsNotNone(backend.stat("campaign/a.eml").modified_at)
|
||||
self.assertEqual(
|
||||
("campaign/a.eml",), tuple(item.key for item in first.objects)
|
||||
)
|
||||
@@ -134,6 +168,10 @@ class ObjectStorageTests(unittest.TestCase):
|
||||
backend.put_bytes("campaign/message.eml", b"message/rfc822")
|
||||
self.assertTrue(backend.exists("campaign/message.eml"))
|
||||
self.assertEqual(b"message/rfc822", backend.get_bytes("campaign/message.eml"))
|
||||
self.assertIsNotNone(backend.stat("campaign/message.eml").modified_at)
|
||||
self.assertIsNotNone(
|
||||
backend.list_objects(prefix="campaign/").objects[0].modified_at
|
||||
)
|
||||
self.assertFalse(backend.exists("campaign/missing.eml"))
|
||||
|
||||
client.head_error = _S3Error("AccessDenied", 403)
|
||||
|
||||
Reference in New Issue
Block a user