Use aioboto3 for asynchronous OCSP S3 uploads
This commit is contained in:
@@ -1222,10 +1222,16 @@ class OcspSnapshotTests(unittest.TestCase):
|
||||
from kca.models import OcspSnapshot
|
||||
|
||||
class FakeS3:
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc, traceback):
|
||||
return None
|
||||
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def put_object(self, **kwargs):
|
||||
async def put_object(self, **kwargs):
|
||||
self.calls.append(kwargs)
|
||||
|
||||
fake = FakeS3()
|
||||
@@ -1342,10 +1348,16 @@ class OcspSnapshotTests(unittest.TestCase):
|
||||
from kca.models import Certificate, OcspSnapshot
|
||||
|
||||
class FakeS3:
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc, traceback):
|
||||
return None
|
||||
|
||||
def __init__(self):
|
||||
self.objects = {}
|
||||
|
||||
def put_object(self, **kwargs):
|
||||
async def put_object(self, **kwargs):
|
||||
self.objects[kwargs["Key"]] = kwargs["Body"]
|
||||
|
||||
fake = FakeS3()
|
||||
|
||||
Reference in New Issue
Block a user