1import collections
2
3import pytest
4
5from pytest_userver import s3api
6
7pytest_plugins = ['pytest_userver.plugins.core']
8
9
10def pytest_configure(config):
11 config.addinivalue_line('markers', 's3: store s3 files in mock')
12
13
14@pytest.fixture(name='s3_mock_storage')
15def _s3_mock_storage():
16 buckets = collections.defaultdict(s3api.S3MockBucketStorage)
17 return buckets
18
19
20@pytest.fixture(autouse=True)
21def s3_mock(mockserver, s3_mock_storage):
22 mock_base_url = '/mds-s3'
23 mock_impl = s3api.S3HandleMock(
24 mockserver=mockserver,
25 s3_mock_storage=s3_mock_storage,
26 mock_base_url=mock_base_url,
27 )
28
29 @mockserver.handler(mock_base_url, prefix=True)
30 def _mock_all(request):
31 if request.method == 'GET':
32 if 'prefix' in request.query:
33 return mock_impl.get_objects(request)
34 return mock_impl.get_object(request)
35
36 if request.method == 'PUT':
37 if 'uploadId' in request.query:
38 if 'partNumber' in request.query:
39 return mock_impl.upload_part(request)
40 return mockserver.make_response('Unknown or unsupported method', 404)
41 if request.headers.get('x-amz-copy-source', None):
42 return mock_impl.copy_object(request)
43 return mock_impl.put_object(request)
44
45 if request.method == 'POST':
46 if 'uploadId' in request.query:
47 return mock_impl.complete_multipart_upload(request)
48 if 'uploads' in request.query:
49 return mock_impl.create_multipart_upload(request)
50
51 if request.method == 'DELETE':
52 if 'uploadId' in request.query:
53 return mock_impl.abort_multipart_upload(request)
54 return mock_impl.delete_object(request)
55
56 if request.method == 'HEAD':
57 return mock_impl.get_object_head(request)
58
59 return mockserver.make_response('Unknown or unsupported method', 404)
60
61
62@pytest.fixture(autouse=True)
63def s3_apply(request, s3_mock_storage, load):
64 def _put_files(bucket, files):
65 bucket_storage = s3_mock_storage[bucket]
66 for s3_path, file_path in files.items():
67 bucket_storage.put_object(
68 key=s3_path,
69 data=load(file_path).encode('utf-8'),
70 )
71
72 for mark in request.node.iter_markers('s3'):
73 _put_files(*mark.args, **mark.kwargs)
