Fix test failures when run as root

supervisor/options.py

@@ -363,6 +363,12 @@ class Options:
                 # if this is called from an RPC method, raise an error
                 raise ValueError(msg)
 
+    def exists(self, path):
+        return os.path.exists(path)
+
+    def open(self, fn, mode='r'):
+        return open(fn, mode)
+
     def get_plugins(self, parser, factory_key, section_prefix):
         factories = []
 
@@ -540,10 +546,10 @@ class ServerOptions(Options):
         section = self.configroot.supervisord
         need_close = False
         if not hasattr(fp, 'read'):
-            if not os.path.exists(fp):
+            if not self.exists(fp):
                 raise ValueError("could not find config file %s" % fp)
             try:
-                fp = open(fp, 'r')
+                fp = self.open(fp, 'r')
                 need_close = True
             except (IOError, OSError):
                 raise ValueError("could not read config file %s" % fp)
@@ -1426,9 +1432,6 @@ class ServerOptions(Options):
     def remove(self, path):
         os.remove(path)
 
-    def exists(self, path):
-        return os.path.exists(path)
-
     def _exit(self, code):
         os._exit(code)
 
@@ -1479,9 +1482,6 @@ class ServerOptions(Options):
     def process_environment(self):
         os.environ.update(self.environment or {})
 
-    def open(self, fn, mode='r'):
-        return open(fn, mode)
-
     def chdir(self, dir):
         os.chdir(dir)
 
@@ -1572,10 +1572,10 @@ class ClientOptions(Options):
         need_close = False
         if not hasattr(fp, 'read'):
             self.here = os.path.dirname(normalize_path(fp))
-            if not os.path.exists(fp):
+            if not self.exists(fp):
                 raise ValueError("could not find config file %s" % fp)
             try:
-                fp = open(fp, 'r')
+                fp = self.open(fp, 'r')
                 need_close = True
             except (IOError, OSError):
                 raise ValueError("could not read config file %s" % fp)

supervisor/tests/test_options.py

@@ -363,37 +363,34 @@ class ClientOptionsTests(unittest.TestCase):
         self.assertEqual(options.password, 'passwordhere')
         self.assertEqual(options.history_file, '/path/to/histdir/.supervisorctl.hist')
 
-    def test_unreadable_config_file(self):
-        fname = missing_but_potential_file()
-        self.assertFalse(os.path.exists(fname))
-
+    def test_read_config_not_found(self):
         instance = self._makeOne()
-        instance.stderr = StringIO()
-
-        instance.exit = dummy_exit()
-        try:
-            instance.realize(args=['-c', fname])
-        except DummyExitException as e:
-            self.assertEqual(e.exitcode, 2)
-        else:
-            self.fail("expected exception")
+        def dummy_exists(fn):
+            return False
+        instance.exists = dummy_exists
 
         try:
-            instance.read_config(fname)
+            instance.read_config('filename')
         except ValueError as e:
             self.assertTrue("could not find config file" in str(e))
         else:
             self.fail("expected exception")
 
-        tempf = tempfile.NamedTemporaryFile()
-        os.chmod(tempf.name, 0) # Removing read perms
+    def test_read_config_unreadable(self):
+        instance = self._makeOne()
+        def dummy_exists(fn):
+            return True
+        instance.exists = dummy_exists
+        def dummy_open(fn, mode):
+            raise IOError(errno.EACCES, 'Permission denied: %s' % fn)
+        instance.open = dummy_open
+
         try:
-            instance.read_config(tempf.name)
+            instance.read_config('filename')
         except ValueError as e:
             self.assertTrue("could not read config file" in str(e))
         else:
             self.fail("expected exception")
-        tempf.close()
 
     def test_options_unixsocket_cli(self):
         fp = StringIO('[supervisorctl]')
@@ -785,40 +782,34 @@ class ServerOptionsTests(unittest.TestCase):
         instance.realize(args=[])
         self.assertFalse(old_warning in instance.parse_warnings)
 
-    def test_unreadable_config_file(self):
-        # Quick and dirty way of coming up with a decent filename
-        tempf = tempfile.NamedTemporaryFile()
-        fname = tempf.name
-        tempf.close()
-        self.assertFalse(os.path.exists(fname))
-
+    def test_read_config_not_found(self):
         instance = self._makeOne()
-        instance.stderr = StringIO()
-
-        instance.exit = dummy_exit()
-        try:
-            instance.realize(args=['-c', fname])
-        except DummyExitException as e:
-            self.assertEqual(e.exitcode, 2)
-        else:
-            self.fail("expected exception")
+        def dummy_exists(fn):
+            return False
+        instance.exists = dummy_exists
 
         try:
-            instance.read_config(fname)
+            instance.read_config('filename')
         except ValueError as e:
             self.assertTrue("could not find config file" in str(e))
         else:
             self.fail("expected exception")
 
-        tempf = tempfile.NamedTemporaryFile()
-        os.chmod(tempf.name, 0) # Removing read perms
+    def test_read_config_unreadable(self):
+        instance = self._makeOne()
+        def dummy_exists(fn):
+            return True
+        instance.exists = dummy_exists
+        def dummy_open(fn, mode):
+            raise IOError(errno.EACCES, 'Permission denied: %s' % fn)
+        instance.open = dummy_open
+
         try:
-            instance.read_config(tempf.name)
+            instance.read_config('filename')
         except ValueError as e:
             self.assertTrue("could not read config file" in str(e))
         else:
             self.fail("expected exception")
-        tempf.close()
 
     def test_readFile_failed(self):
         from supervisor.options import readFile