Add support for local file urls

Change-Id: Ie54508ade52b80790d11958a820998c902d10fd7

heat/common/urlfetch.py

@@ -19,6 +19,7 @@ Utility for fetching a resource (e.g. a template) from a URL.
 
 import requests
 from requests import exceptions
+import urllib2
 import urlparse
 
 from heat.openstack.common import log as logging
@@ -27,20 +28,28 @@ from heat.openstack.common.gettextutils import _
 logger = logging.getLogger(__name__)
 
 
-def get(url):
+def get(url, allowed_schemes=('http', 'https')):
     '''
     Get the data at the specifier URL.
 
     The URL must use the http: or https: schemes.
+    The file: scheme is also supported if you override
+    the allowed_schemes argument.
     Raise an IOError if getting the data fails.
     '''
     logger.info(_('Fetching data from %s') % url)
 
     components = urlparse.urlparse(url)
 
-    if components.scheme not in ('http', 'https'):
+    if components.scheme not in allowed_schemes:
         raise IOError('Invalid URL scheme %s' % components.scheme)
 
+    if components.scheme == 'file':
+        try:
+            return urllib2.urlopen(url).read()
+        except urllib2.URLError as uex:
+            raise IOError('Failed to retrieve template: %s' % str(uex))
+
     try:
         resp = requests.get(url)
         resp.raise_for_status()

heat/tests/test_urlfetch.py

@@ -15,6 +15,8 @@
 
 import requests
 from requests import exceptions
+import urllib2
+import cStringIO
 
 from heat.common import urlfetch
 from heat.tests.common import HeatTestCase
@@ -37,11 +39,32 @@ class UrlFetchTest(HeatTestCase):
         super(UrlFetchTest, self).setUp()
         self.m.StubOutWithMock(requests, 'get')
 
-    def test_file_scheme(self):
+    def test_file_scheme_default_behaviour(self):
         self.m.ReplayAll()
         self.assertRaises(IOError, urlfetch.get, 'file:///etc/profile')
         self.m.VerifyAll()
 
+    def test_file_scheme_supported(self):
+        data = '{ "foo": "bar" }'
+        url = 'file:///etc/profile'
+
+        self.m.StubOutWithMock(urllib2, 'urlopen')
+        urllib2.urlopen(url).AndReturn(cStringIO.StringIO(data))
+        self.m.ReplayAll()
+
+        self.assertEqual(data, urlfetch.get(url, allowed_schemes=['file']))
+        self.m.VerifyAll()
+
+    def test_file_scheme_failure(self):
+        url = 'file:///etc/profile'
+
+        self.m.StubOutWithMock(urllib2, 'urlopen')
+        urllib2.urlopen(url).AndRaise(urllib2.URLError('oops'))
+        self.m.ReplayAll()
+
+        self.assertRaises(IOError, urlfetch.get, url, allowed_schemes=['file'])
+        self.m.VerifyAll()
+
     def test_http_scheme(self):
         url = 'http://example.com/template'
         data = '{ "foo": "bar" }'