RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

testing public method store instead of private method _store_in_thread 

need to mock deferToThread function
This commit is contained in:
Victor Torres 2019-02-08 15:19:57 -02:00
parent cfd183a9d1
commit f824f5b2d1
1 changed files with 28 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
tests/test_feedexport.py
View File

@ -237,7 +237,7 @@ class S3FeedStorageTest(unittest.TestCase):
self.assertEqual(storage.secret_key, 'secret_key')
self.assertEqual(storage.acl, 'custom-acl')
def test_store_in_thread_botocore_without_acl(self):
def test_store_botocore_without_acl(self):
storage = S3FeedStorage(
's3://mybucket/export.csv',
'access_key',
@ -247,13 +247,19 @@ class S3FeedStorageTest(unittest.TestCase):
self.assertEqual(storage.secret_key, 'secret_key')
self.assertEqual(storage.acl, None)
def _defer(f, *args, **kwargs):
return f(*args, **kwargs)
with mock.patch('botocore.client.BaseClient._make_api_call') as m:
storage._store_in_thread(BytesIO(b'test file'))
with mock.patch('twisted.internet.threads.deferToThread',
new=_defer):
storage.store(BytesIO(b'test file'))
operation_name, api_params = m.call_args[0]
self.assertEqual(operation_name, 'PutObject')
self.assertNotIn('ACL', api_params)
def test_store_in_thread_botocore_with_acl(self):
def test_store_botocore_with_acl(self):
storage = S3FeedStorage(
's3://mybucket/export.csv',
'access_key',
@ -264,13 +270,19 @@ class S3FeedStorageTest(unittest.TestCase):
self.assertEqual(storage.secret_key, 'secret_key')
self.assertEqual(storage.acl, 'custom-acl')
def _defer(f, *args, **kwargs):
return f(*args, **kwargs)
with mock.patch('botocore.client.BaseClient._make_api_call') as m:
storage._store_in_thread(BytesIO(b'test file'))
with mock.patch('twisted.internet.threads.deferToThread',
new=_defer):
storage.store(BytesIO(b'test file'))
operation_name, api_params = m.call_args[0]
self.assertEqual(operation_name, 'PutObject')
self.assertEqual(api_params.get('ACL'), 'custom-acl')
def test_store_in_thread_not_botocore_without_acl(self):
def test_store_not_botocore_without_acl(self):
storage = S3FeedStorage(
's3://mybucket/export.csv',
'access_key',
@ -284,7 +296,11 @@ class S3FeedStorageTest(unittest.TestCase):
storage.connect_s3 = mock.MagicMock()
self.assertFalse(storage.is_botocore)
storage._store_in_thread(BytesIO(b'test file'))
def _defer(f, *args, **kwargs):
return f(*args, **kwargs)
with mock.patch('twisted.internet.threads.deferToThread', new=_defer):
storage.store(BytesIO(b'test file'))
conn = storage.connect_s3(*storage.connect_s3.call_args)
bucket = conn.get_bucket(*conn.get_bucket.call_args)
@ -294,7 +310,7 @@ class S3FeedStorageTest(unittest.TestCase):
key.set_contents_from_file.call_args
)
def test_store_in_thread_not_botocore_with_acl(self):
def test_store_not_botocore_with_acl(self):
storage = S3FeedStorage(
's3://mybucket/export.csv',
'access_key',
@ -309,7 +325,11 @@ class S3FeedStorageTest(unittest.TestCase):
storage.connect_s3 = mock.MagicMock()
self.assertFalse(storage.is_botocore)
storage._store_in_thread(BytesIO(b'test file'))
def _defer(f, *args, **kwargs):
return f(*args, **kwargs)
with mock.patch('twisted.internet.threads.deferToThread', new=_defer):
storage.store(BytesIO(b'test file'))
conn = storage.connect_s3(*storage.connect_s3.call_args)
bucket = conn.get_bucket(*conn.get_bucket.call_args)