| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You under the Apache License, Version 2.0 |
| # (the "License"); you may not use this file except in compliance with |
| # the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import sys |
| import errno |
| import hashlib |
| from io import BytesIO |
| from unittest import mock |
| from unittest.mock import Mock |
| |
| from libcloud.test import MockHttp, BodyStream, unittest |
| from libcloud.utils.py3 import PY2, StringIO, b, httplib, assertRaisesRegex |
| from libcloud.storage.base import DEFAULT_CONTENT_TYPE, StorageDriver |
| from libcloud.common.exceptions import RateLimitReachedError |
| from libcloud.test.storage.base import BaseRangeDownloadMockHttp |
| |
| |
| class BaseMockRawResponse(MockHttp): |
| def _(self, method, url, body, headers): |
| body = "ab" |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def root(self, method, url, body, headers): |
| body = "ab" |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| |
| class BaseStorageTests(unittest.TestCase): |
| def setUp(self): |
| self.send_called = 0 |
| StorageDriver.connectionCls.conn_class = BaseMockRawResponse |
| |
| self.driver1 = StorageDriver("username", "key", host="localhost") |
| self.driver1.supports_chunked_encoding = True |
| |
| self.driver2 = StorageDriver("username", "key", host="localhost") |
| self.driver2.supports_chunked_encoding = False |
| |
| self.driver1.strict_mode = False |
| self.driver1.strict_mode = False |
| |
| def test__upload_object_iterator_must_have_next_method(self): |
| |
| valid_iterators = [BytesIO(b("134")), StringIO("bar")] |
| invalid_iterators = ["foobar", "", False, True, 1, object()] |
| |
| kwargs = { |
| "object_name": "foo", |
| "content_type": "foo/bar", |
| "request_path": "/", |
| "headers": {}, |
| } |
| |
| for value in valid_iterators: |
| kwargs["stream"] = value |
| self.driver1._upload_object(**kwargs) |
| |
| for value in invalid_iterators: |
| kwargs["stream"] = value |
| |
| try: |
| self.driver1._upload_object(**kwargs) |
| except AttributeError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test__upload_object_does_not_stream_response(self): |
| resp = self.driver1._upload_object( |
| object_name="foo", |
| content_type="foo/bar", |
| request_path="/", |
| stream=iter(b"foo"), |
| ) |
| mock_response = resp["response"].response._response |
| response_streamed = mock_response.request.stream |
| assert response_streamed is False |
| |
| def test__get_hash_function(self): |
| self.driver1.hash_type = "md5" |
| func = self.driver1._get_hash_function() |
| self.assertTrue(func) |
| |
| self.driver1.hash_type = "sha1" |
| func = self.driver1._get_hash_function() |
| self.assertTrue(func) |
| |
| try: |
| self.driver1.hash_type = "invalid-hash-function" |
| func = self.driver1._get_hash_function() |
| except RuntimeError: |
| pass |
| else: |
| self.fail("Invalid hash type but exception was not thrown") |
| |
| def test_upload_no_content_type_supplied_or_detected(self): |
| iterator = StringIO() |
| |
| # strict_mode is disabled, default content type should be used |
| self.driver1.connection = Mock() |
| |
| self.driver1._upload_object( |
| object_name="test", content_type=None, request_path="/", stream=iterator |
| ) |
| |
| headers = self.driver1.connection.request.call_args[-1]["headers"] |
| self.assertEqual(headers["Content-Type"], DEFAULT_CONTENT_TYPE) |
| |
| # strict_mode is enabled, exception should be thrown |
| |
| self.driver1.strict_mode = True |
| expected_msg = ( |
| 'File content-type could not be guessed for "test" ' |
| "and no content_type value is provided" |
| ) |
| assertRaisesRegex( |
| self, |
| AttributeError, |
| expected_msg, |
| self.driver1._upload_object, |
| object_name="test", |
| content_type=None, |
| request_path="/", |
| stream=iterator, |
| ) |
| |
| @mock.patch("libcloud.utils.files.exhaust_iterator") |
| @mock.patch("libcloud.utils.files.read_in_chunks") |
| def test_upload_object_hash_calculation_is_efficient( |
| self, mock_read_in_chunks, mock_exhaust_iterator |
| ): |
| # Verify that we don't buffer whole file in memory when calculating |
| # object has when iterator has __next__ method, but instead read and |
| # calculate hash in chunks |
| size = 100 |
| |
| self.driver1.connection = Mock() |
| |
| # stream has __next__ method and next() method |
| mock_read_in_chunks.return_value = "a" * size |
| |
| iterator = BodyStream("a" * size) |
| self.assertTrue(hasattr(iterator, "__next__")) |
| self.assertTrue(hasattr(iterator, "next")) |
| |
| self.assertEqual(mock_read_in_chunks.call_count, 0) |
| self.assertEqual(mock_exhaust_iterator.call_count, 0) |
| |
| result = self.driver1._upload_object( |
| object_name="test1", content_type=None, request_path="/", stream=iterator |
| ) |
| |
| hasher = hashlib.md5() |
| hasher.update(b("a") * size) |
| expected_hash = hasher.hexdigest() |
| |
| self.assertEqual(result["data_hash"], expected_hash) |
| self.assertEqual(result["bytes_transferred"], size) |
| |
| headers = self.driver1.connection.request.call_args[-1]["headers"] |
| self.assertEqual(headers["Content-Type"], DEFAULT_CONTENT_TYPE) |
| |
| self.assertEqual(mock_read_in_chunks.call_count, 1) |
| self.assertEqual(mock_exhaust_iterator.call_count, 0) |
| |
| # stream has only has next() method |
| mock_read_in_chunks.return_value = "b" * size |
| |
| iterator = iter([str(v) for v in ["b" * size]]) |
| |
| if PY2: |
| self.assertFalse(hasattr(iterator, "__next__")) |
| self.assertTrue(hasattr(iterator, "next")) |
| else: |
| self.assertTrue(hasattr(iterator, "__next__")) |
| self.assertFalse(hasattr(iterator, "next")) |
| |
| self.assertEqual(mock_read_in_chunks.call_count, 1) |
| self.assertEqual(mock_exhaust_iterator.call_count, 0) |
| |
| self.assertEqual(mock_read_in_chunks.call_count, 1) |
| self.assertEqual(mock_exhaust_iterator.call_count, 0) |
| |
| result = self.driver1._upload_object( |
| object_name="test2", content_type=None, request_path="/", stream=iterator |
| ) |
| |
| hasher = hashlib.md5() |
| hasher.update(b("b") * size) |
| expected_hash = hasher.hexdigest() |
| |
| self.assertEqual(result["data_hash"], expected_hash) |
| self.assertEqual(result["bytes_transferred"], size) |
| |
| headers = self.driver1.connection.request.call_args[-1]["headers"] |
| self.assertEqual(headers["Content-Type"], DEFAULT_CONTENT_TYPE) |
| |
| self.assertEqual(mock_read_in_chunks.call_count, 2) |
| self.assertEqual(mock_exhaust_iterator.call_count, 0) |
| |
| def test_upload_object_via_stream_illegal_seek_errors_are_ignored(self): |
| # Illegal seek errors should be ignored |
| size = 100 |
| |
| self.driver1.connection = Mock() |
| |
| seek_error = OSError("Illegal seek") |
| seek_error.errno = 29 |
| assert errno.ESPIPE == 29 |
| |
| iterator = BodyStream("a" * size) |
| iterator.seek = mock.Mock(side_effect=seek_error) |
| |
| result = self.driver1._upload_object( |
| object_name="test1", content_type=None, request_path="/", stream=iterator |
| ) |
| |
| hasher = hashlib.md5() |
| hasher.update(b("a") * size) |
| expected_hash = hasher.hexdigest() |
| |
| self.assertEqual(result["data_hash"], expected_hash) |
| self.assertEqual(result["bytes_transferred"], size) |
| |
| # But others shouldn't |
| self.driver1.connection = Mock() |
| |
| seek_error = OSError("Other error") |
| seek_error.errno = 21 |
| |
| iterator = BodyStream("b" * size) |
| iterator.seek = mock.Mock(side_effect=seek_error) |
| |
| self.assertRaisesRegex( |
| OSError, |
| "Other error", |
| self.driver1._upload_object, |
| object_name="test1", |
| content_type=None, |
| request_path="/", |
| stream=iterator, |
| ) |
| |
| def test_get_standard_range_str(self): |
| result = self.driver1._get_standard_range_str(0, 5) |
| self.assertEqual(result, "bytes=0-4") |
| |
| result = self.driver1._get_standard_range_str(0) |
| self.assertEqual(result, "bytes=0-") |
| result = self.driver1._get_standard_range_str(0, 1) |
| |
| self.assertEqual(result, "bytes=0-0") |
| |
| result = self.driver1._get_standard_range_str(200) |
| self.assertEqual(result, "bytes=200-") |
| |
| result = self.driver1._get_standard_range_str(10, 200) |
| self.assertEqual(result, "bytes=10-199") |
| |
| result = self.driver1._get_standard_range_str(10, 11) |
| self.assertEqual(result, "bytes=10-10") |
| |
| result = self.driver1._get_standard_range_str(10, 11, True) |
| self.assertEqual(result, "bytes=10-11") |
| |
| @mock.patch("os.environ", {"LIBCLOUD_RETRY_FAILED_HTTP_REQUESTS": True}) |
| def test_should_retry_rate_limited_errors(self): |
| class SecondException(Exception): |
| pass |
| |
| count = 0 |
| |
| def raise_on_second(*_, **__): |
| nonlocal count |
| count += 1 |
| if count > 1: |
| raise SecondException() |
| else: |
| raise RateLimitReachedError() |
| |
| send_mock = Mock() |
| self.driver1.connection.connection.session.send = send_mock |
| send_mock.side_effect = raise_on_second |
| with self.assertRaises(SecondException): |
| self.driver1._upload_object( |
| object_name="some name", |
| content_type="something", |
| request_path="/", |
| stream=iter([]), |
| ) |
| |
| @mock.patch("os.environ", {"LIBCLOUD_RETRY_FAILED_HTTP_REQUESTS": True}) |
| def test_should_retry_rate_limited_errors_until_success(self): |
| count = 0 |
| |
| def succeed_on_second(*_, **__) -> mock.MagicMock: |
| nonlocal count |
| count += 1 |
| if count > 1: |
| successful_response = mock.MagicMock() |
| successful_response.status_code = 200 |
| return successful_response |
| else: |
| raise RateLimitReachedError() |
| |
| self.driver1.connection.connection.session.send = Mock(side_effect=succeed_on_second) |
| uploaded_object = self.driver1._upload_object( |
| object_name="some name", |
| content_type="something", |
| request_path="/", |
| stream=iter([]), |
| ) |
| |
| self.assertEqual( |
| True, |
| uploaded_object["response"].success(), |
| "Expected to have successful response after retry", |
| ) |
| |
| |
| class BaseRangeDownloadMockHttpTestCase(unittest.TestCase): |
| def test_get_start_and_end_bytes_from_range_str(self): |
| mock_http = BaseRangeDownloadMockHttp(None, None) |
| body = "0123456789" |
| |
| range_str = "bytes=1-" |
| result = mock_http._get_start_and_end_bytes_from_range_str(range_str, body) |
| self.assertEqual(result, (1, len(body))) |
| range_str = "bytes=1-5" |
| |
| result = mock_http._get_start_and_end_bytes_from_range_str(range_str, body) |
| self.assertEqual(result, (1, 5)) |
| |
| range_str = "bytes=3-5" |
| result = mock_http._get_start_and_end_bytes_from_range_str(range_str, body) |
| self.assertEqual(result, (3, 5)) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
