Merge bbce761 into 3088aa7

cloudshell/devices/networking_utils.py

@@ -1,12 +1,8 @@
 #!/usr/bin/python
 # -*- coding: utf-8 -*-
 from functools import wraps
-
 import jsonpickle
-import os
-import re
-
-from urlparse import urlsplit, SplitResult
+from urlparse import urlsplit, urlunsplit
 
 
 def validate_vlan_number(number):
@@ -66,46 +62,53 @@ def parse_url(url):
                 if hasattr(parsed, attr_value):
                     value = getattr(parsed, attr_value)
                     if attr_value == UrlParser.PATH:
-                        path, filename = os.path.split(value)
-                        result[UrlParser.PATH] = path
-                        result[UrlParser.FILENAME] = filename
+                        path = value
+                        result[UrlParser.FILENAME] = ""
+                        if not path.endswith("/"):
+                            filename = path[path.rfind("/") + 1:]
+                            result[UrlParser.FILENAME] = filename
+                            path = path.replace(filename, "")
+                        result[UrlParser.PATH] = path[:-1]
                     else:
                         result[attr_value] = value
         return result
 
     @staticmethod
     def build_url(url):
-        url_result = {UrlParser.QUERY: '', UrlParser.FRAGMENT: ''}
-        if not url or UrlParser.SCHEME not in url or not url[UrlParser.SCHEME]:
-            raise Exception('UrlParser:build_url', 'Url dictionary is empty or missing key values')
-
-        url_result[UrlParser.SCHEME] = url[UrlParser.SCHEME]
-
-        if UrlParser.NETLOC in url and url[UrlParser.NETLOC]:
-            if UrlParser.USERNAME in url \
-                    and url[UrlParser.USERNAME] \
-                    and url[UrlParser.USERNAME] in url[UrlParser.NETLOC]:
-                url_result[UrlParser.NETLOC] = url[UrlParser.NETLOC]
-        if UrlParser.NETLOC not in url_result:
-            url_result[UrlParser.NETLOC] = url[UrlParser.HOSTNAME]
-            if UrlParser.PORT in url and url[UrlParser.PORT]:
-                url_result[UrlParser.NETLOC] += ':{}'.format(url[UrlParser.PORT])
-            if UrlParser.USERNAME in url and url[UrlParser.USERNAME]:
-                credentials = '{}@'.format(url[UrlParser.USERNAME])
-                if UrlParser.PASSWORD in url and url[UrlParser.PASSWORD]:
-                    credentials = '{}:{}@'.format(url[UrlParser.USERNAME], url[UrlParser.PASSWORD])
-                url_result[UrlParser.NETLOC] = credentials + url_result[UrlParser.NETLOC]
-
-        url_result[UrlParser.PATH] = url[UrlParser.FILENAME]
-        if UrlParser.PATH in url and url[UrlParser.PATH]:
-            url_result[UrlParser.PATH] = url[UrlParser.PATH] + '/' + url_result[UrlParser.PATH]
-            url_result[UrlParser.PATH] = re.sub('//+', '/', url_result[UrlParser.PATH])
-
-        if UrlParser.QUERY in url and url[UrlParser.QUERY]:
-            url_result[UrlParser.QUERY] = url[UrlParser.QUERY]
-
-        result = SplitResult(**url_result)
-        return result.geturl()
+        if not url:
+            raise Exception('Url dictionary is empty.')
+        scheme = url.get(UrlParser.SCHEME, "")
+        query = url.get(UrlParser.QUERY, "")
+        fragment = url.get(UrlParser.FRAGMENT, "")
+        netloc = url.get(UrlParser.NETLOC)
+        host = url.get(UrlParser.HOSTNAME)
+        port = url.get(UrlParser.PORT)
+        username = url.get(UrlParser.USERNAME)
+        password = url.get(UrlParser.PASSWORD)
+        path = url.get(UrlParser.PATH, "")
+        filename = url.get(UrlParser.FILENAME, "")
+
+        if not scheme:
+            raise Exception('Url missing key value: scheme.')
+
+        if not netloc:
+            if not host:
+                raise Exception('Url missing key value: hostname.')
+            netloc = host
+        if port and str(port) not in netloc:
+            netloc += ':{}'.format(port)
+        if username and username not in netloc or password and password not in netloc:
+            credentials = '{}@'.format(username)
+            if password:
+                credentials = '{}:{}@'.format(username, password)
+            netloc = credentials + netloc
+
+        if path:
+            if not path.endswith("/"):
+                path = "{}/".format(path)
+            path = path + filename
+
+        return urlunsplit((scheme, netloc, path, query, fragment))
 
 
 def command_logging(func):

tests/devices/test_networking_utils.py

@@ -78,7 +78,30 @@ def test_build_url_without_scheme(self):
 
         self.assertRaisesRegexp(
             Exception,
-            'Url dictionary is empty or missing key values',
+            'Url missing key value: scheme.',
+            networking_utils.UrlParser.build_url,
+            url_data,
+        )
+
+    def test_build_url_without_host(self):
+        url_data = self.url_data.copy()
+        url_data['scheme'] = 'scp'
+        url_data['netloc'] = ''
+        url_data['hostname'] = ''
+
+        self.assertRaisesRegexp(
+            Exception,
+            'Url missing key value: hostname.',
+            networking_utils.UrlParser.build_url,
+            url_data,
+        )
+
+    def test_build_url_fail_when_url_empty(self):
+        url_data = {}
+
+        self.assertRaisesRegexp(
+            Exception,
+            'Url dictionary is empty.',
             networking_utils.UrlParser.build_url,
             url_data,
         )
@@ -93,6 +116,7 @@ def test_url_parse_and_builder_returns_the_same(self):
 
     def test_build_url_without_netloc(self):
         url_data = networking_utils.UrlParser.parse_url('http://test.com')
+        url_data["netloc"] = ""
         backup_user = 'user'  # we can add it in ConfigurationRunner
         backup_password = 'password'
         port = '22'
@@ -105,3 +129,17 @@ def test_build_url_without_netloc(self):
         url = networking_utils.UrlParser.build_url(url_data)
 
         self.assertEqual(url, 'http://user:password@test.com:22')
+
+    def test_scp_link_parsed_and_return_same_link(self):
+        url = ("scp://cisco:securePassword!1@test.host.com:"
+               "//d:/some_path/test_file_name.ext?arg=val")
+        url_data = networking_utils.UrlParser.parse_url(url)
+        new_url = networking_utils.UrlParser.build_url(url_data)
+        self.assertEqual(url, new_url)
+
+    def test_link_without_filename(self):
+        url = ("scp://cisco:securePassword!1@test.host.com"
+               "//some_path/")
+        url_data = networking_utils.UrlParser.parse_url(url)
+        new_url = networking_utils.UrlParser.build_url(url_data)
+        self.assertEqual(url, new_url)