vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Change blobuploadcleanupworker to use a data interface

Browse source
This commit is contained in:
Joseph Schorr 2017-07-11 16:58:09 +03:00
parent b2053829f9
commit bdab367285
5 changed files with 93 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

17
workers/blobuploadcleanupworker/test/test_blobuploadcleanupworker.py
View file

@ -1,21 +1,13 @@
import pytest
from contextlib import contextmanager
from datetime import datetime, timedelta
from mock import patch, Mock
from test.fixtures import *
from data import model
from workers.blobuploadcleanupworker.blobuploadcleanupworker import (BlobUploadCleanupWorker,
DELETION_DATE_THRESHOLD)
from workers.blobuploadcleanupworker.blobuploadcleanupworker import BlobUploadCleanupWorker
from workers.blobuploadcleanupworker.models_pre_oci import pre_oci_model as model
def test_blobuploadcleanupworker(initialized_db):
# Create a blob upload older than the threshold.
blob_upload = model.blob.initiate_upload('devtable', 'simple', 'foobarbaz', 'local_us', {})
blob_upload.created = datetime.now() - (DELETION_DATE_THRESHOLD + timedelta(days=100))
blob_upload.save()
assert model.blob.get_stale_blob_upload(DELETION_DATE_THRESHOLD) is not None
blob_upload = model.create_stale_upload_for_testing()
# Note: We need to override UseThenDisconnect to ensure to remains connected to the test DB.
@contextmanager
@ -32,5 +24,4 @@ def test_blobuploadcleanupworker(initialized_db):
storage_mock.cancel_chunked_upload.assert_called_once()
# Ensure the blob no longer exists.
with pytest.raises(model.InvalidBlobUpload):
model.blob.get_blob_upload('devtable', 'simple', blob_upload.uuid)
model.blob_upload_exists(blob_upload.uuid)