| # 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 io |
| import os |
| import re |
| import sys |
| import gzip |
| import time |
| import atexit |
| import random |
| import socket |
| import string |
| import tempfile |
| import unittest |
| |
| import requests |
| |
| import libcloud.http |
| from libcloud.storage import types, providers |
| from libcloud.common.types import LibcloudError |
| |
| try: |
| import docker |
| except ImportError: |
| docker = None |
| |
| |
| libcloud.http.DEFAULT_REQUEST_TIMEOUT = 10 |
| |
| |
| MB = 1024 * 1024 |
| |
| |
| def get_content_iter_with_chunk_size(content, chunk_size=1024): |
| """ |
| Return iterator for the provided content which will return chunks of the specified size. |
| |
| For performance reasons, larger chunks should be used with very large content to speed up the |
| tests (since in some cases each iteration may result in an TCP send). |
| """ |
| # Ensure we still use multiple chunks and iterations |
| assert (len(content) / chunk_size) >= 10 |
| content = iter([content[i : i + chunk_size] for i in range(0, len(content), chunk_size)]) |
| return content |
| |
| |
| class Integration: |
| class TestBase(unittest.TestCase): |
| provider = None |
| account = None |
| secret = None |
| |
| container_name_prefix = "lcsit" |
| container_name_max_length = 63 |
| |
| def setUp(self): |
| for required in "provider", "account", "secret": |
| value = getattr(self, required, None) |
| if value is None: |
| raise unittest.SkipTest("config {} not set".format(required)) |
| |
| kwargs = {"key": self.account, "secret": self.secret} |
| |
| for optional in "host", "port", "secure": |
| value = getattr(self, optional, None) |
| if value is not None: |
| kwargs[optional] = value |
| |
| driver_class = providers.get_driver(self.provider) |
| self.driver = driver_class(**kwargs) |
| |
| def tearDown(self): |
| for container in self.driver.list_containers(): |
| if not container.name.startswith(self.container_name_prefix): |
| continue |
| |
| for obj in container.list_objects(): |
| try: |
| obj.delete() |
| except LibcloudError as ex: |
| print( |
| "Unable to delete object {} in container {}: {}." |
| "Delete it manually.".format(obj.name, container.name, ex), |
| file=sys.stderr, |
| ) |
| |
| try: |
| container.delete() |
| except LibcloudError as ex: |
| print( |
| "Unable to delete container {}: {}." |
| "Delete it manually.".format(container.name, ex), |
| file=sys.stderr, |
| ) |
| |
| def test_containers(self): |
| # make a new container |
| container_name = self._random_container_name() |
| container = self.driver.create_container(container_name) |
| self.assertEqual(container.name, container_name) |
| container = self.driver.get_container(container_name) |
| self.assertEqual(container.name, container_name) |
| |
| self.assert_existing_container_cannot_be_recreated(container) |
| |
| # check that the new container can be listed |
| containers = self.driver.list_containers() |
| self.assertIn(container_name, [c.name for c in containers]) |
| |
| # delete the container |
| self.driver.delete_container(container) |
| |
| # check that a deleted container can't be looked up |
| with self.assertRaises(types.ContainerDoesNotExistError): |
| self.driver.get_container(container_name) |
| |
| # check that the container is deleted |
| containers = self.driver.list_containers() |
| self.assertNotIn(container_name, [c.name for c in containers]) |
| |
| def _test_objects(self, do_upload, do_download, size=1 * MB): |
| content = os.urandom(size) |
| blob_name = "testblob" |
| container = self.driver.create_container(self._random_container_name()) |
| |
| # upload a file |
| obj = do_upload(container, blob_name, content) |
| self.assertEqual(obj.name, blob_name) |
| obj = self.driver.get_object(container.name, blob_name) |
| |
| # check that the file can be listed |
| blobs = self.driver.list_container_objects(container) |
| self.assertEqual([blob.name for blob in blobs], [blob_name]) |
| |
| # upload another file and check it's excluded in prefix listing |
| do_upload(container, blob_name[::-1], content[::-1]) |
| blobs = self.driver.list_container_objects(container, prefix=blob_name[0:3]) |
| self.assertEqual([blob.name for blob in blobs], [blob_name]) |
| |
| # check that the file can be read back |
| downloaded_content = do_download(obj) |
| self.assertEqual(len(downloaded_content), size) |
| self.assertEqual(downloaded_content, content) |
| |
| # delete the file |
| self.driver.delete_object(obj) |
| |
| # check that a missing file can't be deleted or looked up |
| self.assert_file_is_missing(container, obj) |
| |
| # check that the file is deleted |
| blobs = self.driver.list_container_objects(container) |
| self.assertEqual([blob.name for blob in blobs], [blob_name[::-1]]) |
| |
| def assert_existing_container_cannot_be_recreated(self, container): |
| with self.assertRaises(types.ContainerAlreadyExistsError): |
| self.driver.create_container(container.name) |
| |
| def assert_file_is_missing(self, container, obj): |
| with self.assertRaises(types.ObjectDoesNotExistError): |
| self.driver.delete_object(obj) |
| with self.assertRaises(types.ObjectDoesNotExistError): |
| self.driver.get_object(container.name, obj.name) |
| |
| def test_objects(self, size=1 * MB): |
| def do_upload(container, blob_name, content): |
| infile = self._create_tempfile(content=content) |
| return self.driver.upload_object(infile, container, blob_name) |
| |
| def do_download(obj): |
| outfile = self._create_tempfile() |
| self.driver.download_object(obj, outfile, overwrite_existing=True) |
| with open(outfile, "rb") as fobj: |
| return fobj.read() |
| |
| self._test_objects(do_upload, do_download, size) |
| |
| def test_objects_range_downloads(self): |
| blob_name = "testblob-range" |
| content = b"0123456789" |
| container = self.driver.create_container(self._random_container_name()) |
| |
| obj = self.driver.upload_object( |
| self._create_tempfile(content=content), container, blob_name |
| ) |
| self.assertEqual(obj.name, blob_name) |
| self.assertEqual(obj.size, len(content)) |
| |
| obj = self.driver.get_object(container.name, blob_name) |
| self.assertEqual(obj.name, blob_name) |
| self.assertEqual(obj.size, len(content)) |
| |
| values = [ |
| {"start_bytes": 0, "end_bytes": 1, "expected_content": b"0"}, |
| {"start_bytes": 1, "end_bytes": 5, "expected_content": b"1234"}, |
| {"start_bytes": 5, "end_bytes": None, "expected_content": b"56789"}, |
| {"start_bytes": 5, "end_bytes": len(content), "expected_content": b"56789"}, |
| {"start_bytes": 0, "end_bytes": None, "expected_content": b"0123456789"}, |
| {"start_bytes": 0, "end_bytes": len(content), "expected_content": b"0123456789"}, |
| ] |
| |
| for value in values: |
| # 1. download_object_range |
| start_bytes = value["start_bytes"] |
| end_bytes = value["end_bytes"] |
| outfile = self._create_tempfile() |
| |
| result = self.driver.download_object_range( |
| obj, |
| outfile, |
| start_bytes=start_bytes, |
| end_bytes=end_bytes, |
| overwrite_existing=True, |
| ) |
| self.assertTrue(result) |
| |
| with open(outfile, "rb") as fobj: |
| downloaded_content = fobj.read() |
| |
| if end_bytes is not None: |
| expected_content = content[start_bytes:end_bytes] |
| else: |
| expected_content = content[start_bytes:] |
| |
| msg = 'Expected "{}", got "{}" for values: {}'.format( |
| expected_content, |
| downloaded_content, |
| str(value), |
| ) |
| self.assertEqual(downloaded_content, expected_content, msg) |
| self.assertEqual(downloaded_content, value["expected_content"], msg) |
| |
| # 2. download_object_range_as_stream |
| downloaded_content = read_stream( |
| self.driver.download_object_range_as_stream( |
| obj, start_bytes=start_bytes, end_bytes=end_bytes |
| ) |
| ) |
| self.assertEqual(downloaded_content, expected_content) |
| |
| @unittest.skipUnless(os.getenv("LARGE_FILE_SIZE_MB"), "config not set") |
| def test_objects_large(self): |
| size = int(float(os.environ["LARGE_FILE_SIZE_MB"]) * MB) |
| self.test_objects(size) |
| |
| def test_objects_stream_io(self): |
| def do_upload(container, blob_name, content): |
| content = io.BytesIO(content) |
| return self.driver.upload_object_via_stream(content, container, blob_name) |
| |
| def do_download(obj): |
| return read_stream(self.driver.download_object_as_stream(obj)) |
| |
| self._test_objects(do_upload, do_download) |
| |
| def test_objects_stream_iterable(self): |
| def do_upload(container, blob_name, content): |
| # NOTE: We originally used a chunk size of 1 which resulted in many requests and as |
| # such, very slow tests. To speed things up, we use a longer chunk size. |
| assert (len(content) / 1024) >= 500 |
| content = get_content_iter_with_chunk_size(content, 1024) |
| return self.driver.upload_object_via_stream(content, container, blob_name) |
| |
| def do_download(obj): |
| return read_stream(self.driver.download_object_as_stream(obj)) |
| |
| self._test_objects(do_upload, do_download) |
| |
| def test_upload_via_stream_with_content_encoding(self): |
| object_name = "content_encoding.gz" |
| content = gzip.compress(os.urandom(MB // 100)) |
| container = self.driver.create_container(self._random_container_name()) |
| self.driver.upload_object_via_stream( |
| get_content_iter_with_chunk_size(content, 1000), |
| container, |
| object_name, |
| headers={"Content-Encoding": "gzip"}, |
| ) |
| |
| obj = self.driver.get_object(container.name, object_name) |
| |
| self.assertEqual(obj.extra.get("content_encoding"), "gzip") |
| |
| def test_cdn_url(self): |
| content = os.urandom(MB // 100) |
| container = self.driver.create_container(self._random_container_name()) |
| content_iter = get_content_iter_with_chunk_size(content, 1000) |
| obj = self.driver.upload_object_via_stream(content_iter, container, "cdn") |
| |
| response = requests.get(self.driver.get_object_cdn_url(obj)) |
| response.raise_for_status() |
| |
| self.assertEqual(response.content, content) |
| |
| def _create_tempfile(self, prefix="", content=b""): |
| fobj, path = tempfile.mkstemp(prefix=prefix, text=False) |
| |
| try: |
| os.write(fobj, content) |
| finally: |
| os.close(fobj) |
| |
| self.addCleanup(os.remove, path) |
| return path |
| |
| @classmethod |
| def _random_container_name(cls): |
| suffix = random_string(cls.container_name_max_length) |
| name = cls.container_name_prefix + suffix |
| name = re.sub("[^a-z0-9-]", "-", name) |
| name = re.sub("-+", "-", name) |
| name = name[: cls.container_name_max_length] |
| name = name.lower() |
| return name |
| |
| class ContainerTestBase(TestBase): |
| image = None |
| version = "latest" |
| environment = {} |
| command = None |
| ready_message = None |
| |
| host = "localhost" |
| port = None |
| secure = False |
| |
| client = None |
| container = None |
| verbose = False |
| |
| container_ready_timeout = 20 |
| |
| @classmethod |
| def setUpClass(cls): |
| if docker is None: |
| raise unittest.SkipTest("missing docker library") |
| |
| try: |
| cls.client = docker.from_env() |
| except docker.errors.DockerException: |
| raise unittest.SkipTest("unable to create docker client") |
| |
| for required in "image", "port": |
| value = getattr(cls, required, None) |
| if value is None: |
| raise unittest.SkipTest("config {} not set".format(required)) |
| |
| cls.container = cls.client.containers.run( |
| "{}:{}".format(cls.image, cls.version), |
| command=cls.command, |
| detach=True, |
| auto_remove=True, |
| ports={cls.port: cls.port}, |
| environment=cls.environment, |
| ) |
| |
| # We register atexit handler to ensure container is always killed, even if for some |
| # reason tearDownClass is sometimes not called (happened locally a couple of times) |
| atexit.register(cls._kill_container) |
| |
| wait_for(cls.port, cls.host) |
| |
| container_ready = cls.ready_message is None |
| |
| start_ts = int(time.time()) |
| timeout_ts = start_ts + cls.container_ready_timeout |
| |
| while not container_ready: |
| time.sleep(1) |
| |
| container_ready = any( |
| cls.ready_message in line for line in cls.container.logs().splitlines() |
| ) |
| |
| now_ts = int(time.time()) |
| |
| if now_ts >= timeout_ts: |
| raise ValueError( |
| "Container %s failed to start and become ready in %s seconds " |
| '(did not find "%s" message in container logs).\n\n' |
| "Container Logs:\n%s" |
| % ( |
| cls.container.short_id, |
| cls.container_ready_timeout, |
| cls.ready_message.decode("utf-8"), |
| cls.container.logs().decode("utf-8"), |
| ) |
| ) |
| |
| @classmethod |
| def tearDownClass(cls): |
| cls._kill_container() |
| atexit.unregister(cls._kill_container) |
| |
| @classmethod |
| def _kill_container(cls): |
| if cls.verbose: |
| print("Killing / stopping container with id %s..." % (cls.container.short_id)) |
| |
| if cls.verbose: |
| for line in cls.container.logs().splitlines(): |
| print(line) |
| |
| try: |
| cls.container.kill() |
| except docker.errors.DockerException as ex: |
| print( |
| "Unable to terminate docker container {}: {}." |
| "Stop it manually.".format(cls.container.short_id, ex), |
| file=sys.stderr, |
| ) |
| |
| |
| def wait_for(port, host="localhost", timeout=10): |
| start_time = time.perf_counter() |
| |
| while True: |
| try: |
| with socket.create_connection((host, port), timeout=timeout): |
| break |
| except OSError as ex: |
| if time.perf_counter() - start_time >= timeout: |
| raise TimeoutError( |
| "Waited too long for the port {} on host {} to start accepting " |
| "connections.".format(port, host) |
| ) from ex |
| |
| time.sleep(1) |
| |
| |
| def random_string(length, alphabet=string.ascii_lowercase + string.digits): |
| return "".join(random.choice(alphabet) for _ in range(length)) |
| |
| |
| def read_stream(stream): |
| buffer = io.BytesIO() |
| buffer.writelines(stream) |
| buffer.seek(0) |
| return buffer.read() |
