working on removing: # pragma: no cover from pserve

2 files changed, 151 insertions, 34 deletions

diff --git a/pyramid/scripts/pserve.py b/pyramid/scripts/pserve.py
index 31a07c46f..dcc095d1a 100644
--- a/pyramid/scripts/pserve.py
+++ b/pyramid/scripts/pserve.py
@@ -303,7 +303,7 @@ class PServeCommand(object):
                 if self.verbose > 1:
                     raise
                 if str(e):
-                    msg = ' '+str(e)
+                    msg = ' ' + str(e)
                 else:
                     msg = ''
                 self.out('Exiting%s (-v to see traceback)' % msg)
@@ -396,8 +396,7 @@ class PServeCommand(object):
         os.dup2(0, 1)  # standard output (1)
         os.dup2(0, 2)  # standard error (2)
 
-    def _remove_pid_file(self, written_pid, filename,
-                         verbosity): # pragma: no cover
+    def _remove_pid_file(self, written_pid, filename, verbosity):
         current_pid = os.getpid()
         if written_pid != current_pid:
             # A forked process must be exiting, not the process that
@@ -405,17 +404,16 @@ class PServeCommand(object):
             return
         if not os.path.exists(filename):
             return
-        f = open(filename)
-        content = f.read().strip()
-        f.close()
+        with open(filename) as f:
+            content = f.read().strip()
         try:
             pid_in_file = int(content)
         except ValueError:
             pass
         else:
             if pid_in_file != current_pid:
-                self.out("PID file %s contains %s, not expected PID %s" % (
-                    filename, pid_in_file, current_pid))
+                msg = "PID file %s contains %s, not expected PID %s"
+                self.out(msg % (filename, pid_in_file, current_pid))
                 return
         if verbosity > 0:
             self.out("Removing PID file %s" % filename)
@@ -424,24 +422,22 @@ class PServeCommand(object):
             return
         except OSError as e:
             # Record, but don't give traceback
-            self.out("Cannot remove PID file: %s" % e)
+            self.out("Cannot remove PID file: (%s)" % e)
         # well, at least lets not leave the invalid PID around...
         try:
-            f = open(filename, 'w')
-            f.write('')
-            f.close()
+            with open(filename, 'w') as f:
+                f.write('')
         except OSError as e:
-            self.out('Stale PID left in file: %s (%e)' % (filename, e))
+            self.out('Stale PID left in file: %s (%s)' % (filename, e))
         else:
             self.out('Stale PID removed')
 
-    def record_pid(self, pid_file): # pragma: no cover
+    def record_pid(self, pid_file):
         pid = os.getpid()
         if self.verbose > 1:
             self.out('Writing PID %s to %s' % (pid, pid_file))
-        f = open(pid_file, 'w')
-        f.write(str(pid))
-        f.close()
+        with open(pid_file, 'w') as f:
+            f.write(str(pid))
         atexit.register(self._remove_pid_file, pid, pid_file, self.verbose)
 
     def stop_daemon(self): # pragma: no cover
@@ -533,7 +529,7 @@ class PServeCommand(object):
                 if exit_code != 3:
                     return exit_code
             if self.verbose > 0:
-                self.out('%s %s %s' % ('-'*20, 'Restarting', '-'*20))
+                self.out('%s %s %s' % ('-' * 20, 'Restarting', '-' * 20))
 
     def change_user_group(self, user, group): # pragma: no cover
         if not user and not group:
@@ -795,7 +791,7 @@ class Monitor(object): # pragma: no cover
 
     def periodic_reload(self):
         while True:
-            if not self.check_reload(): 
+            if not self.check_reload():
                 self._exit()
                 break
             time.sleep(self.poll_interval)
@@ -965,7 +961,7 @@ def cherrypy_server_runner(
     try:
         protocol = is_ssl and 'https' or 'http'
         if host == '0.0.0.0':
-            print('serving on 0.0.0.0:%s view at %s://127.0.0.1:%s' % 
+            print('serving on 0.0.0.0:%s view at %s://127.0.0.1:%s' %
                   (port, protocol, port))
         else:
             print('serving on %s://%s:%s' % (protocol, host, port))
diff --git a/pyramid/tests/test_scripts/test_pserve.py b/pyramid/tests/test_scripts/test_pserve.py
index e21a703d9..534bb37aa 100644
--- a/pyramid/tests/test_scripts/test_pserve.py
+++ b/pyramid/tests/test_scripts/test_pserve.py
@@ -1,15 +1,21 @@
-import unittest
+import __builtin__
 import os
 import tempfile
+import unittest
 
 class TestPServeCommand(unittest.TestCase):
     def setUp(self):
         from pyramid.compat import NativeIO
         self.out_ = NativeIO()
+        self.pid_file = None
+
+    def tearDown(self):
+        if self.pid_file and os.path.exists(self.pid_file):
+            os.remove(self.pid_file)
 
     def out(self, msg):
         self.out_.write(msg)
-        
+
     def _getTargetClass(self):
         from pyramid.scripts.pserve import PServeCommand
         return PServeCommand
@@ -21,6 +27,124 @@ class TestPServeCommand(unittest.TestCase):
         cmd.out = self.out
         return cmd
 
+    def _makeOneWithPidFile(self, pid):
+        self.pid_file = tempfile.mktemp()
+        inst = self._makeOne()
+        with open(self.pid_file, 'w') as f:
+            f.write(str(pid))
+        return inst
+
+    def test_remove_pid_file_verbose(self):
+        inst = self._makeOneWithPidFile(os.getpid())
+        inst._remove_pid_file(os.getpid(), self.pid_file, verbosity=1)
+        self._assert_pid_file_removed(verbose=True)
+
+    def test_remove_pid_file_not_verbose(self):
+        inst = self._makeOneWithPidFile(os.getpid())
+        inst._remove_pid_file(os.getpid(), self.pid_file, verbosity=0)
+        self._assert_pid_file_removed(verbose=False)
+
+    def test_remove_pid_not_a_number(self):
+        inst = self._makeOneWithPidFile('not a number')
+        inst._remove_pid_file(os.getpid(), self.pid_file, verbosity=1)
+        self._assert_pid_file_removed(verbose=True)
+
+    def test_remove_pid_current_pid_is_not_written_pid(self):
+        inst = self._makeOneWithPidFile(os.getpid())
+        inst._remove_pid_file('99999', self.pid_file, verbosity=1)
+        self._assert_pid_file_not_removed('')
+
+    def test_remove_pid_current_pid_is_not_pid_in_file(self):
+        inst = self._makeOneWithPidFile('99999')
+        inst._remove_pid_file(os.getpid(), self.pid_file, verbosity=1)
+        msg = 'PID file %s contains 99999, not expected PID %s'
+        self._assert_pid_file_not_removed(msg % (self.pid_file, os.getpid()))
+
+    def test_remove_pid_no_pid_file(self):
+        inst = self._makeOne()
+        self.pid_file = 'some unknown path'
+        inst._remove_pid_file(os.getpid(), self.pid_file, verbosity=1)
+        self._assert_pid_file_removed(verbose=False)
+
+    def test_remove_pid_file_unlink_exception(self):
+        inst = self._makeOneWithPidFile(os.getpid())
+        self._remove_pid_unlink_exception(inst)
+        msg = [
+            'Removing PID file %s' % (self.pid_file),
+            'Cannot remove PID file: (Some OSError - unlink)',
+            'Stale PID removed']
+        self._assert_pid_file_not_removed(msg=''.join(msg))
+
+    def test_remove_pid_file_stale_pid_write_exception(self):
+        inst = self._makeOneWithPidFile(os.getpid())
+        self._remove_pid_unlink_and_write_exceptions(inst)
+        msg = [
+            'Removing PID file %s' % (self.pid_file),
+            'Cannot remove PID file: (Some OSError - unlink)',
+            'Stale PID left in file: %s ' % (self.pid_file),
+            '(Some OSError - open)']
+        self._assert_pid_file_not_removed(msg=''.join(msg))
+
+    def test_record_pid_verbose(self):
+        self._assert_record_pid(verbosity=2, msg='Writing PID %d to %s')
+
+    def test_record_pid_not_verbose(self):
+        self._assert_record_pid(verbosity=1, msg='')
+
+    def _remove_pid_unlink_exception(self, inst):
+        old_unlink = os.unlink
+
+        def fake_unlink(filename):
+            raise OSError('Some OSError - unlink')
+
+        try:
+            os.unlink = fake_unlink
+            inst._remove_pid_file(os.getpid(), self.pid_file, verbosity=1)
+        finally:
+            os.unlink = old_unlink
+
+    def _remove_pid_unlink_and_write_exceptions(self, inst):
+        old_unlink = os.unlink
+        old_open = __builtin__.open
+
+        def fake_unlink(filename):
+            raise OSError('Some OSError - unlink')
+
+        run_already = []
+        def fake_open(*args):
+            if not run_already:
+                run_already.append(True)
+                return old_open(*args)
+            raise OSError('Some OSError - open')
+
+        try:
+            os.unlink = fake_unlink
+            __builtin__.open = fake_open
+            inst._remove_pid_file(os.getpid(), self.pid_file, verbosity=1)
+        finally:
+            os.unlink = old_unlink
+            __builtin__.open = old_open
+
+    def _assert_pid_file_removed(self, verbose=False):
+        self.assertFalse(os.path.exists(self.pid_file))
+        msg = 'Removing PID file %s' % (self.pid_file) if verbose else ''
+        self.assertEqual(self.out_.getvalue(), msg)
+
+    def _assert_pid_file_not_removed(self, msg):
+        self.assertTrue(os.path.exists(self.pid_file))
+        self.assertEqual(self.out_.getvalue(), msg)
+
+    def _assert_record_pid(self, verbosity, msg):
+        self.pid_file = tempfile.mktemp()
+        pid = os.getpid()
+        inst = self._makeOne()
+        inst.verbose = verbosity
+        inst.record_pid(self.pid_file)
+        msg = msg % (pid, self.pid_file) if msg else ''
+        self.assertEqual(self.out_.getvalue(), msg)
+        with open(self.pid_file) as f:
+            self.assertEqual(int(f.read()), pid)
+
     def test_run_no_args(self):
         inst = self._makeOne()
         result = inst.run()
@@ -31,15 +155,15 @@ class TestPServeCommand(unittest.TestCase):
         path = os.path.join(os.path.dirname(__file__), 'wontexist.pid')
         inst = self._makeOne('--stop-daemon', '--pid-file=%s' % path)
         inst.run()
-        self.assertEqual(self.out_.getvalue(),'No PID file exists in %s' %
-                         path)
-        
+        msg = 'No PID file exists in %s' % path
+        self.assertEqual(self.out_.getvalue(), msg)
+
     def test_run_stop_daemon_bad_pid_file(self):
         path = __file__
         inst = self._makeOne('--stop-daemon', '--pid-file=%s' % path)
         inst.run()
-        self.assertEqual(
-            self.out_.getvalue(),'Not a valid PID file in %s' % path)
+        msg = 'Not a valid PID file in %s' % path
+        self.assertEqual(self.out_.getvalue(), msg)
 
     def test_run_stop_daemon_invalid_pid_in_file(self):
         fn = tempfile.mktemp()
@@ -48,15 +172,15 @@ class TestPServeCommand(unittest.TestCase):
         tmp.close()
         inst = self._makeOne('--stop-daemon', '--pid-file=%s' % fn)
         inst.run()
-        self.assertEqual(self.out_.getvalue(),
-                         'PID in %s is not valid (deleting)' % fn)
+        msg = 'PID in %s is not valid (deleting)' % fn
+        self.assertEqual(self.out_.getvalue(), msg)
 
     def test_parse_vars_good(self):
         vars = ['a=1', 'b=2']
         inst = self._makeOne('development.ini')
         result = inst.parse_vars(vars)
         self.assertEqual(result, {'a': '1', 'b': '2'})
-        
+
     def test_parse_vars_bad(self):
         vars = ['a']
         inst = self._makeOne('development.ini')
@@ -85,7 +209,7 @@ class TestLazyWriter(unittest.TestCase):
         finally:
             fp.close()
             os.remove(filename)
-        
+
     def test_write(self):
         filename = tempfile.mktemp()
         try:
@@ -138,6 +262,3 @@ class Test__methodwrapper(unittest.TestCase):
         class Bar(object): pass
         wrapper = self._makeOne(foo, Bar, None)
         self.assertRaises(AssertionError, wrapper, cls=1)
-        
-        
-