Add sources.detect to detect various source types

Change-Id: Ic1e325538f0975b04750e10233e877ffcfbf4263

metalsmith/_cmd.py

@@ -29,10 +29,6 @@ from metalsmith import sources
 LOG = logging.getLogger(__name__)
 
 
-def _is_http(smth):
-    return smth.startswith('http://') or smth.startswith('https://')
-
-
 class NICAction(argparse.Action):
     def __call__(self, parser, namespace, values, option_string=None):
         assert option_string in ('--port', '--network', '--ip')
@@ -64,41 +60,10 @@ def _do_deploy(api, args, formatter):
     if args.hostname and not _utils.is_hostname_safe(args.hostname):
         raise RuntimeError("%s cannot be used as a hostname" % args.hostname)
 
-    if _is_http(args.image):
+    source = sources.detect(args.image,
-        kwargs = {}
+                            kernel=args.image_kernel,
-        if not args.image_checksum:
+                            ramdisk=args.image_ramdisk,
-            raise RuntimeError("HTTP(s) images require --image-checksum")
+                            checksum=args.image_checksum)
-        elif _is_http(args.image_checksum):
-            kwargs['checksum_url'] = args.image_checksum
-        else:
-            kwargs['checksum'] = args.image_checksum
-
-        if args.image_kernel or args.image_ramdisk:
-            source = sources.HttpPartitionImage(args.image,
-                                                args.image_kernel,
-                                                args.image_ramdisk,
-                                                **kwargs)
-        else:
-            source = sources.HttpWholeDiskImage(args.image, **kwargs)
-    elif args.image.startswith('file://'):
-        if not args.image_checksum:
-            raise RuntimeError("File images require --image-checksum")
-
-        if args.image_kernel or args.image_ramdisk:
-            if not (args.image_kernel.startswith('file://') and
-                    args.image_ramdisk.startswith('file://')):
-                raise RuntimeError('Images with the file:// schema require '
-                                   'kernel and ramdisk images to also use '
-                                   'the file:// schema')
-            source = sources.FilePartitionImage(args.image,
-                                                args.image_kernel,
-                                                args.image_ramdisk,
-                                                args.image_checksum)
-        else:
-            source = sources.FileWholeDiskImage(args.image,
-                                                args.image_checksum)
-    else:
-        source = args.image
 
     config = _config.InstanceConfig(ssh_keys=ssh_keys)
     if args.user_name:
@@ -176,10 +141,8 @@ def _parse_args(args, config):
                         required=True)
     deploy.add_argument('--image-checksum',
                         help='image MD5 checksum or URL with checksums')
-    deploy.add_argument('--image-kernel', help='URL of the image\'s kernel',
+    deploy.add_argument('--image-kernel', help='URL of the image\'s kernel')
-                        default='')
+    deploy.add_argument('--image-ramdisk', help='URL of the image\'s ramdisk')
-    deploy.add_argument('--image-ramdisk', help='URL of the image\'s ramdisk',
-                        default='')
     deploy.add_argument('--network', help='network to use (name or UUID)',
                         dest='nics', action=NICAction)
     deploy.add_argument('--port', help='port to attach (name or UUID)',

metalsmith/sources.py

@@ -50,19 +50,19 @@ class GlanceImage(_Source):
 
         :param image: `Image` object, ID or name.
         """
-        self._image_id = image
+        self.image = image
         self._image_obj = None
 
     def _validate(self, connection):
         if self._image_obj is not None:
             return
         try:
-            self._image_obj = connection.image.find_image(self._image_id,
+            self._image_obj = connection.image.find_image(self.image,
                                                           ignore_missing=False)
         except openstack.exceptions.SDKException as exc:
             raise exceptions.InvalidImage(
                 'Cannot find image %(image)s: %(error)s' %
-                {'image': self._image_id, 'error': exc})
+                {'image': self.image, 'error': exc})
 
     def _node_updates(self, connection):
         self._validate(connection)
@@ -242,3 +242,82 @@ class FilePartitionImage(FileWholeDiskImage):
         updates['kernel'] = self.kernel_location
         updates['ramdisk'] = self.ramdisk_location
         return updates
+
+
+def detect(image, kernel=None, ramdisk=None, checksum=None):
+    """Try detecting the correct source type from the provided information.
+
+    .. note::
+        Images without a schema are assumed to be Glance images.
+
+    :param image: Location of the image: ``file://``, ``http://``, ``https://``
+        link or a Glance image name or UUID.
+    :param kernel: Location of the kernel (if present): ``file://``,
+        ``http://``, ``https://`` link or a Glance image name or UUID.
+    :param ramdisk: Location of the ramdisk (if present): ``file://``,
+        ``http://``, ``https://`` link or a Glance image name or UUID.
+    :param checksum: MD5 checksum of the image: ``http://`` or ``https://``
+        link or a string.
+    :return: A valid source object.
+    :raises: ValueError if the given parameters do not correspond to any
+        valid source.
+    """
+    image_type = _link_type(image)
+    checksum_type = _link_type(checksum)
+
+    if image_type == 'glance':
+        if kernel or ramdisk or checksum:
+            raise ValueError('kernel, image and checksum cannot be provided '
+                             'for Glance images')
+        else:
+            return GlanceImage(image)
+
+    kernel_type = _link_type(kernel)
+    ramdisk_type = _link_type(ramdisk)
+    if not checksum:
+        raise ValueError('checksum is required for HTTP and file images')
+
+    if image_type == 'file':
+        if (kernel_type not in (None, 'file')
+                or ramdisk_type not in (None, 'file')
+                or checksum_type == 'http'):
+            raise ValueError('kernal, ramdisk and checksum can only be files '
+                             'for file images')
+
+        if kernel or ramdisk:
+            return FilePartitionImage(image,
+                                      kernel_location=kernel,
+                                      ramdisk_location=ramdisk,
+                                      checksum=checksum)
+        else:
+            return FileWholeDiskImage(image, checksum=checksum)
+    else:
+        if (kernel_type not in (None, 'http')
+                or ramdisk_type not in (None, 'http')
+                or checksum_type == 'file'):
+            raise ValueError('kernal, ramdisk and checksum can only be HTTP '
+                             'links for HTTP images')
+
+        if checksum_type == 'http':
+            kwargs = {'checksum_url': checksum}
+        else:
+            kwargs = {'checksum': checksum}
+
+        if kernel or ramdisk:
+            return HttpPartitionImage(image,
+                                      kernel_url=kernel,
+                                      ramdisk_url=ramdisk,
+                                      **kwargs)
+        else:
+            return HttpWholeDiskImage(image, **kwargs)
+
+
+def _link_type(link):
+    if link is None:
+        return None
+    elif link.startswith('http://') or link.startswith('https://'):
+        return 'http'
+    elif link.startswith('file://'):
+        return 'file'
+    else:
+        return 'glance'

metalsmith/test/test_cmd.py

@@ -49,7 +49,7 @@ class TestDeploy(testtools.TestCase):
                                 candidates=None)
         reserve_defaults.update(reserve_args)
 
-        provision_defaults = dict(image='myimg',
+        provision_defaults = dict(image=mock.ANY,
                                   nics=[{'network': 'mynet'}],
                                   root_size_gb=None,
                                   swap_size_mb=None,
@@ -88,6 +88,10 @@ class TestDeploy(testtools.TestCase):
         self.assertEqual([], config.ssh_keys)
         mock_log.basicConfig.assert_called_once_with(level=mock_log.WARNING,
                                                      format=mock.ANY)
+
+        source = mock_pr.return_value.provision_node.call_args[1]['image']
+        self.assertIsInstance(source, sources.GlanceImage)
+        self.assertEqual("myimg", source.image)
         self.assertEqual(
             mock.call('metalsmith').setLevel(mock_log.WARNING).call_list() +
             mock.call(_cmd._URLLIB3_LOGGER).setLevel(

metalsmith/test/test_sources.py

@@ -0,0 +1,126 @@
+# Copyright 2019 Red Hat, Inc.
+#
+# Licensed 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 testtools
+
+from metalsmith import sources
+
+
+class TestDetect(testtools.TestCase):
+
+    def test_glance(self):
+        source = sources.detect('foobar')
+        self.assertIsInstance(source, sources.GlanceImage)
+        self.assertEqual(source.image, 'foobar')
+
+    def test_glance_invalid_arguments(self):
+        for kwargs in [{'kernel': 'foo'},
+                       {'ramdisk': 'foo'},
+                       {'checksum': 'foo'}]:
+            self.assertRaisesRegex(ValueError, 'cannot be provided',
+                                   sources.detect, 'foobar', **kwargs)
+
+    def test_checksum_required(self):
+        for tp in ('file', 'http', 'https'):
+            self.assertRaisesRegex(ValueError, 'checksum is required',
+                                   sources.detect, '%s://foo' % tp)
+
+    def test_file_whole_disk(self):
+        source = sources.detect('file:///image', checksum='abcd')
+        self.assertIs(source.__class__, sources.FileWholeDiskImage)
+        self.assertEqual(source.location, 'file:///image')
+        self.assertEqual(source.checksum, 'abcd')
+
+    def test_file_partition_disk(self):
+        source = sources.detect('file:///image', checksum='abcd',
+                                kernel='file:///kernel',
+                                ramdisk='file:///ramdisk')
+        self.assertIs(source.__class__, sources.FilePartitionImage)
+        self.assertEqual(source.location, 'file:///image')
+        self.assertEqual(source.checksum, 'abcd')
+        self.assertEqual(source.kernel_location, 'file:///kernel')
+        self.assertEqual(source.ramdisk_location, 'file:///ramdisk')
+
+    def test_file_partition_inconsistency(self):
+        for kwargs in [{'kernel': 'foo'},
+                       {'ramdisk': 'foo'},
+                       {'kernel': 'http://foo'},
+                       {'ramdisk': 'http://foo'},
+                       {'checksum': 'http://foo'}]:
+            kwargs.setdefault('checksum', 'abcd')
+            self.assertRaisesRegex(ValueError, 'can only be files',
+                                   sources.detect, 'file:///image', **kwargs)
+
+    def test_http_whole_disk(self):
+        source = sources.detect('http:///image', checksum='abcd')
+        self.assertIs(source.__class__, sources.HttpWholeDiskImage)
+        self.assertEqual(source.url, 'http:///image')
+        self.assertEqual(source.checksum, 'abcd')
+
+    def test_https_whole_disk(self):
+        source = sources.detect('https:///image', checksum='abcd')
+        self.assertIs(source.__class__, sources.HttpWholeDiskImage)
+        self.assertEqual(source.url, 'https:///image')
+        self.assertEqual(source.checksum, 'abcd')
+
+    def test_https_whole_disk_checksum(self):
+        source = sources.detect('https:///image',
+                                checksum='https://checksum')
+        self.assertIs(source.__class__, sources.HttpWholeDiskImage)
+        self.assertEqual(source.url, 'https:///image')
+        self.assertEqual(source.checksum_url, 'https://checksum')
+
+    def test_http_partition_disk(self):
+        source = sources.detect('http:///image', checksum='abcd',
+                                kernel='http:///kernel',
+                                ramdisk='http:///ramdisk')
+        self.assertIs(source.__class__, sources.HttpPartitionImage)
+        self.assertEqual(source.url, 'http:///image')
+        self.assertEqual(source.checksum, 'abcd')
+        self.assertEqual(source.kernel_url, 'http:///kernel')
+        self.assertEqual(source.ramdisk_url, 'http:///ramdisk')
+
+    def test_https_partition_disk(self):
+        source = sources.detect('https:///image', checksum='abcd',
+                                # Can mix HTTP and HTTPs
+                                kernel='http:///kernel',
+                                ramdisk='https:///ramdisk')
+        self.assertIs(source.__class__, sources.HttpPartitionImage)
+        self.assertEqual(source.url, 'https:///image')
+        self.assertEqual(source.checksum, 'abcd')
+        self.assertEqual(source.kernel_url, 'http:///kernel')
+        self.assertEqual(source.ramdisk_url, 'https:///ramdisk')
+
+    def test_https_partition_disk_checksum(self):
+        source = sources.detect('https:///image',
+                                # Can mix HTTP and HTTPs
+                                checksum='http://checksum',
+                                kernel='http:///kernel',
+                                ramdisk='https:///ramdisk')
+        self.assertIs(source.__class__, sources.HttpPartitionImage)
+        self.assertEqual(source.url, 'https:///image')
+        self.assertEqual(source.checksum_url, 'http://checksum')
+        self.assertEqual(source.kernel_url, 'http:///kernel')
+        self.assertEqual(source.ramdisk_url, 'https:///ramdisk')
+
+    def test_http_partition_inconsistency(self):
+        for kwargs in [{'kernel': 'foo'},
+                       {'ramdisk': 'foo'},
+                       {'kernel': 'file://foo'},
+                       {'ramdisk': 'file://foo'},
+                       {'checksum': 'file://foo'}]:
+            kwargs.setdefault('checksum', 'abcd')
+            self.assertRaisesRegex(ValueError, 'can only be HTTP',
+                                   sources.detect, 'http:///image', **kwargs)

releasenotes/notes/source-detect-673ad8c3e98c3df1.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    Adds new function ``metalsmith.sources.detect`` to automate detection of
+    various sources from their location, kernel, image and checksum.