| # -*- coding: utf-8 -*- |
| # 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 pytest |
| import socket |
| import codecs |
| import unittest |
| import warnings |
| import platform |
| import os.path |
| from itertools import chain |
| |
| # In Python > 2.7 DeprecationWarnings are disabled by default |
| warnings.simplefilter("default") |
| |
| import libcloud.utils.files |
| |
| from libcloud.utils.misc import get_driver, set_driver |
| |
| from libcloud.utils.py3 import PY3 |
| from libcloud.utils.py3 import StringIO |
| from libcloud.utils.py3 import b |
| from libcloud.utils.py3 import bchr |
| from libcloud.utils.py3 import hexadigits |
| from libcloud.utils.py3 import urlquote |
| from libcloud.compute.types import Provider |
| from libcloud.compute.providers import DRIVERS |
| from libcloud.compute.drivers.dummy import DummyNodeDriver |
| from libcloud.utils.misc import get_secure_random_string |
| from libcloud.utils.networking import is_public_subnet |
| from libcloud.utils.networking import is_private_subnet |
| from libcloud.utils.networking import is_valid_ip_address |
| from libcloud.utils.networking import join_ipv4_segments |
| from libcloud.utils.networking import increment_ipv4_segments |
| from libcloud.utils.decorators import wrap_non_libcloud_exceptions |
| from libcloud.utils.publickey import ( |
| get_pubkey_openssh_fingerprint, |
| get_pubkey_ssh2_fingerprint, |
| ) |
| from libcloud.common.types import LibcloudError |
| from libcloud.storage.drivers.dummy import DummyIterator |
| |
| |
| WARNINGS_BUFFER = [] |
| |
| if PY3: |
| from io import FileIO as file |
| |
| |
| def show_warning(msg, cat, fname, lno, file=None, line=None): |
| WARNINGS_BUFFER.append((msg, cat, fname, lno)) |
| |
| |
| original_func = warnings.showwarning |
| |
| |
| class TestUtils(unittest.TestCase): |
| def setUp(self): |
| global WARNINGS_BUFFER |
| WARNINGS_BUFFER = [] |
| |
| def tearDown(self): |
| global WARNINGS_BUFFER |
| WARNINGS_BUFFER = [] |
| warnings.showwarning = original_func |
| |
| @unittest.skipIf(platform.system().lower() == "windows", "Unsupported on Windows") |
| def test_guess_file_mime_type(self): |
| file_path = os.path.abspath(__file__) |
| mimetype, encoding = libcloud.utils.files.guess_file_mime_type( |
| file_path=file_path |
| ) |
| |
| self.assertTrue(mimetype.find("python") != -1) |
| |
| def test_get_driver(self): |
| driver = get_driver(drivers=DRIVERS, provider=Provider.DUMMY) |
| self.assertTrue(driver is not None) |
| |
| try: |
| driver = get_driver(drivers=DRIVERS, provider="fooba") |
| except AttributeError: |
| pass |
| else: |
| self.fail("Invalid provider, but an exception was not thrown") |
| |
| def test_get_driver_string_and_enum_notation(self): |
| driver = get_driver(drivers=DRIVERS, provider=Provider.DUMMY) |
| self.assertEqual(driver, DummyNodeDriver) |
| |
| driver = get_driver(drivers=DRIVERS, provider="dummy") |
| self.assertEqual(driver, DummyNodeDriver) |
| |
| driver = get_driver(drivers=DRIVERS, provider="DUMMY") |
| self.assertEqual(driver, DummyNodeDriver) |
| |
| def test_set_driver(self): |
| # Set an existing driver |
| try: |
| driver = set_driver( |
| DRIVERS, |
| Provider.DUMMY, |
| "libcloud.storage.drivers.dummy", |
| "DummyStorageDriver", |
| ) |
| except AttributeError: |
| pass |
| |
| # Register a new driver |
| driver = set_driver( |
| DRIVERS, |
| "testingset", |
| "libcloud.storage.drivers.dummy", |
| "DummyStorageDriver", |
| ) |
| |
| self.assertTrue(driver is not None) |
| |
| # Register it again |
| try: |
| set_driver( |
| DRIVERS, |
| "testingset", |
| "libcloud.storage.drivers.dummy", |
| "DummyStorageDriver", |
| ) |
| except AttributeError: |
| pass |
| |
| # Register an invalid module |
| try: |
| set_driver( |
| DRIVERS, |
| "testingnew", |
| "libcloud.storage.drivers.dummy1", |
| "DummyStorageDriver", |
| ) |
| except ImportError: |
| pass |
| |
| # Register an invalid class |
| try: |
| set_driver( |
| DRIVERS, |
| "testingnew", |
| "libcloud.storage.drivers.dummy", |
| "DummyStorageDriver1", |
| ) |
| except AttributeError: |
| pass |
| |
| def test_deprecated_warning(self): |
| warnings.showwarning = show_warning |
| |
| libcloud.utils.SHOW_DEPRECATION_WARNING = False |
| self.assertEqual(len(WARNINGS_BUFFER), 0) |
| libcloud.utils.deprecated_warning("test_module") |
| self.assertEqual(len(WARNINGS_BUFFER), 0) |
| |
| libcloud.utils.SHOW_DEPRECATION_WARNING = True |
| self.assertEqual(len(WARNINGS_BUFFER), 0) |
| libcloud.utils.deprecated_warning("test_module") |
| self.assertEqual(len(WARNINGS_BUFFER), 1) |
| |
| def test_in_development_warning(self): |
| warnings.showwarning = show_warning |
| |
| libcloud.utils.SHOW_IN_DEVELOPMENT_WARNING = False |
| self.assertEqual(len(WARNINGS_BUFFER), 0) |
| libcloud.utils.in_development_warning("test_module") |
| self.assertEqual(len(WARNINGS_BUFFER), 0) |
| |
| libcloud.utils.SHOW_IN_DEVELOPMENT_WARNING = True |
| self.assertEqual(len(WARNINGS_BUFFER), 0) |
| libcloud.utils.in_development_warning("test_module") |
| self.assertEqual(len(WARNINGS_BUFFER), 1) |
| |
| def test_read_in_chunks_iterator_no_data(self): |
| iterator = DummyIterator() |
| generator1 = libcloud.utils.files.read_in_chunks( |
| iterator=iterator, yield_empty=False |
| ) |
| generator2 = libcloud.utils.files.read_in_chunks( |
| iterator=iterator, yield_empty=True |
| ) |
| |
| # yield_empty=False |
| count = 0 |
| for data in generator1: |
| count += 1 |
| self.assertEqual(data, b("")) |
| |
| self.assertEqual(count, 0) |
| |
| # yield_empty=True |
| count = 0 |
| for data in generator2: |
| count += 1 |
| self.assertEqual(data, b("")) |
| |
| self.assertEqual(count, 1) |
| |
| def test_read_in_chunks_iterator(self): |
| def iterator(): |
| for x in range(0, 1000): |
| yield "aa" |
| |
| for result in libcloud.utils.files.read_in_chunks( |
| iterator(), chunk_size=10, fill_size=False |
| ): |
| self.assertEqual(result, b("aa")) |
| |
| for result in libcloud.utils.files.read_in_chunks( |
| iterator(), chunk_size=10, fill_size=True |
| ): |
| self.assertEqual(result, b("aaaaaaaaaa")) |
| |
| def test_read_in_chunks_filelike(self): |
| class FakeFile(file): |
| def __init__(self): |
| self.remaining = 500 |
| |
| def read(self, size): |
| self.remaining -= 1 |
| if self.remaining == 0: |
| return "" |
| return "b" * (size + 1) |
| |
| for index, result in enumerate( |
| libcloud.utils.files.read_in_chunks( |
| FakeFile(), chunk_size=10, fill_size=False |
| ) |
| ): |
| self.assertEqual(result, b("b" * 11)) |
| |
| self.assertEqual(index, 498) |
| |
| for index, result in enumerate( |
| libcloud.utils.files.read_in_chunks( |
| FakeFile(), chunk_size=10, fill_size=True |
| ) |
| ): |
| if index != 548: |
| self.assertEqual(result, b("b" * 10)) |
| else: |
| self.assertEqual(result, b("b" * 9)) |
| |
| self.assertEqual(index, 548) |
| |
| def test_exhaust_iterator(self): |
| def iterator_func(): |
| for x in range(0, 1000): |
| yield "aa" |
| |
| data = b("aa" * 1000) |
| iterator = libcloud.utils.files.read_in_chunks(iterator=iterator_func()) |
| result = libcloud.utils.files.exhaust_iterator(iterator=iterator) |
| self.assertEqual(result, data) |
| |
| result = libcloud.utils.files.exhaust_iterator(iterator=iterator_func()) |
| self.assertEqual(result, data) |
| |
| data = "12345678990" |
| iterator = StringIO(data) |
| result = libcloud.utils.files.exhaust_iterator(iterator=iterator) |
| self.assertEqual(result, b(data)) |
| |
| def test_exhaust_iterator_empty_iterator(self): |
| data = "" |
| iterator = StringIO(data) |
| result = libcloud.utils.files.exhaust_iterator(iterator=iterator) |
| self.assertEqual(result, b(data)) |
| |
| def test_unicode_urlquote(self): |
| # Regression tests for LIBCLOUD-429 |
| if PY3: |
| # Note: this is a unicode literal |
| val = "\xe9" |
| else: |
| val = codecs.unicode_escape_decode("\xe9")[0] |
| |
| uri = urlquote(val) |
| self.assertEqual(b(uri), b("%C3%A9")) |
| |
| # Unicode without unicode characters |
| uri = urlquote("v=1") |
| self.assertEqual(b(uri), b("v%3D1")) |
| |
| # Already-encoded bytestring without unicode characters |
| uri = urlquote(b("v=1")) |
| self.assertEqual(b(uri), b("v%3D1")) |
| |
| def test_get_secure_random_string(self): |
| for i in range(1, 500): |
| value = get_secure_random_string(size=i) |
| self.assertEqual(len(value), i) |
| |
| def test_hexadigits(self): |
| self.assertEqual(hexadigits(b("")), []) |
| self.assertEqual(hexadigits(b("a")), ["61"]) |
| self.assertEqual( |
| hexadigits(b("AZaz09-")), ["41", "5a", "61", "7a", "30", "39", "2d"] |
| ) |
| |
| def test_bchr(self): |
| if PY3: |
| self.assertEqual(bchr(0), b"\x00") |
| self.assertEqual(bchr(97), b"a") |
| else: |
| self.assertEqual(bchr(0), "\x00") |
| self.assertEqual(bchr(97), "a") |
| |
| |
| class NetworkingUtilsTestCase(unittest.TestCase): |
| def test_is_public_and_is_private_subnet(self): |
| public_ips = ["213.151.0.8", "86.87.86.1", "8.8.8.8", "8.8.4.4"] |
| |
| private_ips = ["192.168.1.100", "10.0.0.1", "172.16.0.0"] |
| |
| for address in public_ips: |
| is_public = is_public_subnet(ip=address) |
| is_private = is_private_subnet(ip=address) |
| |
| self.assertTrue(is_public) |
| self.assertFalse(is_private) |
| |
| for address in private_ips: |
| is_public = is_public_subnet(ip=address) |
| is_private = is_private_subnet(ip=address) |
| |
| self.assertFalse(is_public) |
| self.assertTrue(is_private) |
| |
| def test_is_valid_ip_address(self): |
| valid_ipv4_addresses = [ |
| "192.168.1.100", |
| "10.0.0.1", |
| "213.151.0.8", |
| "77.77.77.77", |
| ] |
| |
| invalid_ipv4_addresses = [ |
| "10.1", |
| "256.256.256.256", |
| "0.567.567.567", |
| "192.168.0.257", |
| ] |
| |
| valid_ipv6_addresses = [ |
| "fe80::200:5aee:feaa:20a2", |
| "2607:f0d0:1002:51::4", |
| "2607:f0d0:1002:0051:0000:0000:0000:0004", |
| "::1", |
| ] |
| |
| invalid_ipv6_addresses = [ |
| "2607:f0d", |
| "2607:f0d0:0004", |
| ] |
| |
| for address in valid_ipv4_addresses: |
| status = is_valid_ip_address(address=address, family=socket.AF_INET) |
| self.assertTrue(status) |
| |
| for address in valid_ipv6_addresses: |
| status = is_valid_ip_address(address=address, family=socket.AF_INET6) |
| self.assertTrue(status) |
| |
| for address in chain(invalid_ipv4_addresses, invalid_ipv6_addresses): |
| status = is_valid_ip_address(address=address, family=socket.AF_INET) |
| self.assertFalse(status) |
| |
| for address in chain(invalid_ipv4_addresses, invalid_ipv6_addresses): |
| status = is_valid_ip_address(address=address, family=socket.AF_INET6) |
| self.assertFalse(status) |
| |
| def test_join_ipv4_segments(self): |
| values = [ |
| (("127", "0", "0", "1"), "127.0.0.1"), |
| (("255", "255", "255", "0"), "255.255.255.0"), |
| ] |
| |
| for segments, joined_ip in values: |
| result = join_ipv4_segments(segments=segments) |
| self.assertEqual(result, joined_ip) |
| |
| def test_increment_ipv4_segments(self): |
| values = [ |
| (("127", "0", "0", "1"), "127.0.0.2"), |
| (("255", "255", "255", "0"), "255.255.255.1"), |
| (("254", "255", "255", "255"), "255.0.0.0"), |
| (("100", "1", "0", "255"), "100.1.1.0"), |
| ] |
| |
| for segments, incremented_ip in values: |
| result = increment_ipv4_segments(segments=segments) |
| result = join_ipv4_segments(segments=result) |
| self.assertEqual(result, incremented_ip) |
| |
| |
| class TestPublicKeyUtils(unittest.TestCase): |
| |
| PUBKEY = ( |
| "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDOfbWSXOlqvYjZmRO84/lIoV4gvuX+" |
| "P1lLg50MMg6jZjLZIlYY081XPRmuom0xY0+BO++J2KgLl7gxJ6xMsKK2VQ+TakdfAH20" |
| "XfMcTohd/zVCeWsbqZQvEhVXBo4hPIktcfNz0u9Ez3EtInO+kb7raLcRhOVi9QmOkOrC" |
| "WtQU9mS71AWJuqI9H0YAnTiI8Hs5bn2tpMIqmTXT3g2bwywC25x1Nx9Hy0/FP+KUL6Ag" |
| "vDXv47l+TgSDfTBEkvq+IF1ITrnaOG+nRE02oZC6cwHYTifM/IOollkujxIQmi2Z+j66" |
| "OHSrjnEQugr0FqGJF2ygKfIh/i2u3fVLM60qE2NN user@example" |
| ) |
| |
| def test_pubkey_openssh_fingerprint(self): |
| fp = get_pubkey_openssh_fingerprint(self.PUBKEY) |
| self.assertEqual(fp, "35:22:13:5b:82:e2:5d:e1:90:8c:73:74:9f:ef:3b:d8") |
| |
| def test_pubkey_ssh2_fingerprint(self): |
| fp = get_pubkey_ssh2_fingerprint(self.PUBKEY) |
| self.assertEqual(fp, "11:ad:5d:4c:5b:99:c9:80:7e:81:03:76:5a:25:9d:8c") |
| |
| |
| def test_decorator(): |
| @wrap_non_libcloud_exceptions |
| def foo(): |
| raise Exception("bork") |
| |
| with pytest.raises(LibcloudError): |
| foo() |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
