Add tests

resync/url_or_file_open.py

@@ -8,7 +8,7 @@ from . import __version__
 
 
 # Global configuration settings
-FIRST_REQUEST = False
+NUM_REQUESTS = 0
 CONFIG = {
     'bearer_token': None,
     'delay': None
@@ -22,20 +22,24 @@ def set_url_or_file_open_config(key, value):
 
 
 def url_or_file_open(uri):
-    """Wrapper around urlopen() to prepend file: if no scheme provided."""
+    """Wrapper around urlopen() to prepend file: if no scheme provided.
+
+    Can be used as a context manager because the return value from urlopen(...)
+    supports both that and straightforwrd use as simple file handle object.
+    """
     if (not re.match(r'''\w+:''', uri)):
         uri = 'file:' + uri
+    headers = {'User-Agent': 'resync/' + __version__}
     # Do we need to send an Authorization header?
     # FIXME - This token will be added blindy to all requests. This is insecure
     # if the --noauth setting is used allowing requests across different domains.
     # It would be better to have some scheme where a token is tied to a particular
     # domain, or domain pattern.
-    headers = {'User-Agent': 'resync/' + __version__}
     if CONFIG['bearer_token'] is not None:
         headers['Authorization'] = 'Bearer ' + CONFIG['bearer_token']
     # Have we got a delay set? Apply only to web requests after first
-    global FIRST_REQUEST
-    if not FIRST_REQUEST and CONFIG['delay'] is not None and not uri.startswith('file:'):
-        FIRST_REQUEST = False
+    global NUM_REQUESTS
+    if NUM_REQUESTS != 0 and CONFIG['delay'] is not None and not uri.startswith('file:'):
         time.sleep(CONFIG['delay'])
-    return(urlopen(Request(url=uri, headers=headers)))
+    NUM_REQUESTS += 1
+    return urlopen(Request(url=uri, headers=headers))

tests/test_url_or_file_open.py

@@ -0,0 +1,37 @@
+"""Tests for resync.url_or_file_open."""
+from .testlib import TestCase, webserver
+import time
+
+from resync.url_or_file_open import NUM_REQUESTS, CONFIG, set_url_or_file_open_config, url_or_file_open
+
+
+class TestUrlOrFileOpen(TestCase):
+
+    def test_set_url_or_file_open_config(self):
+        """Test set_url_or_file_open_config function."""
+        self.assertEqual(CONFIG['bearer_token'], None)
+        self.assertEqual(CONFIG['delay'], None)
+        self.assertNotIn('my_thing', CONFIG)
+        set_url_or_file_open_config('bearer_token', 'open seasame')
+        self.assertEqual(CONFIG['bearer_token'], 'open seasame')
+        set_url_or_file_open_config('my_thing', 'special')
+        self.assertEqual(CONFIG['my_thing'], 'special')
+
+    def test_url_or_file_open(self):
+        """Test basic operation of url_or_file_open function."""
+        # Open file
+        fh = url_or_file_open('tests/testdata/dir1/file_a')
+        self.assertIn(b'I am file a', fh.read())
+        fh.close()
+        with url_or_file_open('file:tests/testdata/dir1/file_b') as fh:
+            self.assertIn(b'I am file b', fh.read())
+        # Open URL
+        with webserver('tests/testdata', 'localhost', 9999):
+            with url_or_file_open('http://localhost:9999/dir2/file_x') as fh:
+                self.assertIn(b'I am the mysterious file_x', fh.read())
+            # test delay of 0.1s
+            set_url_or_file_open_config('delay', 0.1)
+            before = time.time()
+            with url_or_file_open('http://localhost:9999/dir1/file_a') as fh:
+                self.assertIn(b'I am file a', fh.read())
+            self.assertGreater(time.time() - before, 0.099)