Merge "Add file protocol handling to url checker"

fuelmenu/fuelmenu/modules/bootstrapimg.py

@@ -66,7 +66,8 @@ class bootstrapimg(urwid.WidgetWrap):
                 "value": "http://archive.ubuntu.com/ubuntu"},
             "BOOTSTRAP/MIRROR_MOS": {
                 "label": "MOS mirror",
-                "tooltip": "MOS APT repo URL",
+                "tooltip": ("MOS APT repo URL (can use file:// protocol, will"
+                            "use local mirror in such case"),
                 "value": mos_repo_dflt},
             "BOOTSTRAP/HTTP_PROXY": {
                 "label": "HTTP proxy",

network_checker/test-requirements.txt

@@ -1,3 +1,5 @@
 hacking==0.7
 mock==1.0.1
 pytest
+unittest2
+requests-mock

network_checker/url_access_checker/api.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 import json
+import os
 import socket
 
 import requests
@@ -20,11 +21,20 @@ import requests
 import url_access_checker.errors as errors
 
 
+FILE_PREFIX = 'file://'
+
+
 def check_urls(urls, proxies=None, timeout=60):
-    """Checks a set of urls to see if they are up and returning 200
+    """Checks a set of urls to see if they are valid
+
+    Url is valid if
+    - it returns 200 upon requesting it with http (if it doesn't specify
+    protocol as "file")
+    - it is an existing file or directory (if the protocol used in url is
+    "file")
 
     Arguments:
-    url -- a string containing url for testing
+    urls -- an iterable containing urls for testing
     proxies -- proxy servers to use for the request
     timeout -- the max time to wait for a response, default 60 seconds
     """
@@ -43,7 +53,7 @@ def _get_response_tuple(url, proxies=None, timeout=60):
     """Return a tuple which contains a result of url test
 
     Arguments:
-    url -- a string containing url for testing
+    url -- a string containing url for testing, can be local file
     proxies -- proxy servers to use for the request
     timeout -- the max time to wait for a response, default 60 seconds
 
@@ -51,6 +61,19 @@ def _get_response_tuple(url, proxies=None, timeout=60):
         result[0] -- boolean value, True if the url is deemed failed
         result[1] -- unchange url argument
     """
+
+    if url.startswith(FILE_PREFIX):
+        return _get_file_existence_tuple(url)
+    else:
+        return _get_http_response_tuple(url, proxies, timeout)
+
+
+def _get_file_existence_tuple(url):
+    path = url[len(FILE_PREFIX):]
+    return (not os.path.exists(path), url)
+
+
+def _get_http_response_tuple(url, proxies=None, timeout=60):
     try:
         # requests seems to correctly handle various corner cases:
         # proxies=None or proxies={} mean 'use the default' rather than

network_checker/url_access_checker/tests/unit/test_api.py

@@ -0,0 +1,58 @@
+#    Copyright 2015 Mirantis, 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 unittest2
+
+import mock
+import requests_mock
+
+from url_access_checker import api
+from url_access_checker import errors
+
+
+class TestApi(unittest2.TestCase):
+
+    def setUp(self):
+        self.urls = ['http://url{0}'.format(i) for i in range(10)]
+        self.paths = ['file:///tmp/test_api{0}'.format(i) for i in range(10)]
+
+    @requests_mock.Mocker()
+    def test_check_urls(self, req_mocker):
+        for url in self.urls:
+            req_mocker.get(url, status_code=200)
+
+        check_result = api.check_urls(self.urls)
+
+        self.assertTrue(check_result)
+
+    @requests_mock.Mocker()
+    def test_check_urls_fail(self, req_mocker):
+        for url in self.urls:
+            req_mocker.get(url, status_code=404)
+
+        with self.assertRaises(errors.UrlNotAvailable):
+            api.check_urls(self.urls)
+
+    @mock.patch('os.path.exists')
+    def test_check_paths(self, mock_exists):
+        mock_exists.return_value = True
+        check_result = api.check_urls(self.paths)
+
+        self.assertTrue(check_result)
+
+    @mock.patch('os.path.exists')
+    def test_check_paths_fail(self, mock_exists):
+        mock_exists.return_value = False
+        with self.assertRaises(errors.UrlNotAvailable):
+            api.check_urls(self.paths)