Use modern unittest methods

test/backend/test_config.py

@@ -88,7 +88,7 @@ def test_send_email(self):
                 conf = get_config(config_class=config.Config)
                 conf.send_email(to, 'testsubj', body)
                 mail = conf.get_mail_output()
-                self.assert_(re.search('Subject: testsubj.*From: '
+                self.assertTrue(re.search('Subject: testsubj.*From: '
                                    'test@salilab\.org.*To: testto.*testbody',
                                    mail, flags=re.DOTALL),
                              'Unexpected mail output: ' + mail)
@@ -133,7 +133,7 @@ def test_time_deltas(self):
         self.assertEqual(conf.oldjobs['expire'].days, 1000)
         # Check NEVER
         conf = get_config(expire='never')
-        self.assertEqual(conf.oldjobs['expire'], None)
+        self.assertIsNone(conf.oldjobs['expire'])
         # Other suffixes or non-floats should raise an error
         self.assertRaises(ValueError, get_config, expire='8s')
         self.assertRaises(ValueError, get_config, expire='4.garbageh')

test/backend/test_database.py

@@ -157,7 +157,7 @@ def _connect(self, config):
         self.assertRaises(DummyError, db._execute, 'execute exception')
         # Return value should be the cursor
         c = db._execute('CREATE TABLE jobs (test TEXT)')
-        self.assert_(isinstance(c, DummyCursor))
+        self.assertIsInstance(c, DummyCursor)
 
         # If a 'gone away' error is encountered and reraised, the connection
         # should have been reestablished
@@ -170,7 +170,7 @@ def _connect(self, config):
         oldconn = db.conn
         c = db._execute('reconnect succeeds')
         self.assertNotEqual(id(db.conn), id(oldconn))
-        self.assert_(isinstance(c, DummyCursor))
+        self.assertIsInstance(c, DummyCursor)
 
     def test_drop_tables(self):
         """Check Database._drop_tables()"""
@@ -270,7 +270,7 @@ def test_update_job(self):
         db._update_job(job._metadata, 'INCOMING')
         # Get a fresh copy of the job from the database
         newjob = list(db._get_all_jobs_in_state('INCOMING'))[0]
-        self.assert_(job is not newjob)
+        self.assertIsNot(job, newjob)
         self.assertEqual(newjob._metadata['runner_id'], 'new-SGE-ID')
 
     def test_get_job_dependencies(self):

test/backend/test_delete_all_jobs.py

@@ -26,11 +26,11 @@ class DelJobTest(unittest.TestCase):
     def test_check_not_running(self):
         """Test check_not_running function"""
         web = DummyWeb(None)
-        self.assertEqual(check_not_running(web), None)
+        self.assertIsNone(check_not_running(web))
         web = DummyWeb(99)
         self.assertRaises(ValueError, check_not_running, web)
         web = DummyWeb(None, err=True)
-        self.assertEqual(check_not_running(web), None)
+        self.assertIsNone(check_not_running(web))
 
     def test_main(self):
         """Test delete_all_jobs main()"""

test/backend/test_do_nothing_runner.py

@@ -24,7 +24,7 @@ def test_run(self):
         self.assertEqual(DoNothingRunner._check_completed('none', ''), True)
         event = ws._event_queue.get()
         event.process()
-        self.assertEqual(ws._exception, None)
+        self.assertIsNone(ws._exception)
         os.unlink('job-state')
 
 if __name__ == '__main__':

test/backend/test_events.py

@@ -10,12 +10,12 @@ class EventsTest(unittest.TestCase):
     def test_event_queue(self):
         """Check the _EventQueue class"""
         e = saliweb.backend.events._EventQueue()
-        self.assertEqual(e.get(0), None)
+        self.assertIsNone(e.get(0))
         e.put('a')
         e.put('b')
         self.assertEqual(e.get(), 'a')
         self.assertEqual(e.get(), 'b')
-        self.assertEqual(e.get(0), None)
+        self.assertIsNone(e.get(0))
 
     def test_incoming_jobs_event(self):
         """Check the _IncomingJobsEvent class"""
@@ -59,7 +59,7 @@ def _get_job_by_runner_id(self, runner, runid):
         ws = DummyWebService()
         ev = saliweb.backend.events._CompletedJobEvent(ws, None, 'good', None)
         ev.process()
-        self.assertEqual(ws.run_exception, None)
+        self.assertIsNone(ws.run_exception)
 
         ws = DummyWebService()
         ev = saliweb.backend.events._CompletedJobEvent(ws, None, 'good', 'foo')
@@ -85,8 +85,8 @@ def _get_oldjob_interval(self): return 0.02
         # Should have added 2 events
         for i in range(2):
             x = q.get(timeout=0.)
-            self.assert_(isinstance(x, saliweb.backend.events._OldJobsEvent))
-        self.assertEqual(q.get(timeout=0.), None)
+            self.assertIsInstance(x, saliweb.backend.events._OldJobsEvent)
+        self.assertIsNone(q.get(timeout=0.))
 
     def test_periodic_check(self):
         """Check the _PeriodicCheck class"""
@@ -103,9 +103,9 @@ class dummy: pass
         # Should have added 2 events
         for i in range(2):
             x = q.get(timeout=0.)
-            self.assert_(isinstance(x,
-                                 saliweb.backend.events._PeriodicCheckEvent))
-        self.assertEqual(q.get(timeout=0.), None)
+            self.assertIsInstance(x,
+                                 saliweb.backend.events._PeriodicCheckEvent)
+        self.assertIsNone(q.get(timeout=0.))
 
     def test_periodic_check_event(self):
         """Check the _PeriodicCheckEvent class"""
@@ -131,9 +131,9 @@ def _get_cleanup_incoming_job_times(self): return (0.02, 0.02)
         # Should have added 2 events
         for i in range(2):
             x = q.get(timeout=0.)
-            self.assert_(isinstance(x,
-                         saliweb.backend.events._CleanupIncomingJobsEvent))
-        self.assertEqual(q.get(timeout=0.), None)
+            self.assertIsInstance(x,
+                         saliweb.backend.events._CleanupIncomingJobsEvent)
+        self.assertIsNone(q.get(timeout=0.))
 
     def test_incoming_jobs(self):
         """Check the _IncomingJobs class"""
@@ -158,8 +158,8 @@ class dummy: pass
         time.sleep(0.05)
         # Should have added 1 event
         x = q.get(timeout=0.)
-        self.assert_(isinstance(x, saliweb.backend.events._IncomingJobsEvent))
-        self.assertEqual(q.get(timeout=0.), None)
+        self.assertIsInstance(x, saliweb.backend.events._IncomingJobsEvent)
+        self.assertIsNone(q.get(timeout=0.))
         os.unlink('test.sock')
 
 if __name__ == '__main__':

test/backend/test_job.py

@@ -282,7 +282,7 @@ class JobTest(unittest.TestCase):
     """Check Job class"""
 
     def assert_fail_msg(self, failre, job):
-        self.assert_(re.search(failre, job._metadata['failure'],
+        self.assertTrue(re.search(failre, job._metadata['failure'],
                                flags=re.DOTALL),
                      'Unexpected failure message: ' + job._metadata['failure'])
 
@@ -313,8 +313,8 @@ def test_ok_startup(self):
         # New fields should have been populated in the database
         self.assertEqual(job._metadata['testfield'], 'run')
         self.assertEqual(job._metadata['runner_id'], 'mock:MyJob ID')
-        self.assertNotEqual(job._metadata['preprocess_time'], None)
-        self.assertNotEqual(job._metadata['run_time'], None)
+        self.assertIsNotNone(job._metadata['preprocess_time'])
+        self.assertIsNotNone(job._metadata['run_time'])
         # Both preprocess and run methods in MyJob should have triggered
         os.unlink(os.path.join(runjobdir, 'preproc'))
         os.unlink(os.path.join(runjobdir, 'job-output'))
@@ -332,7 +332,7 @@ def test_sanity_check_no_directory(self):
         db.conn.commit()
         web._process_incoming_jobs()
         job = web.get_job_by_name('FAILED', 'job1')
-        self.assertEqual(job.directory, None)
+        self.assertIsNone(job.directory)
         self.assert_fail_msg('Python exception:.*Traceback.*' \
                              + 'SanityError: .*did not set the directory', job)
         cleanup_webservice(conf, tmpdir)
@@ -348,7 +348,7 @@ def test_sanity_check_invalid_directory(self):
         db.conn.commit()
         web._process_incoming_jobs()
         job = web.get_job_by_name('FAILED', 'job2')
-        self.assertEqual(job.directory, None)
+        self.assertIsNone(job.directory)
         self.assert_fail_msg('Python exception:.*Traceback.*' \
                              + 'SanityError: .*is not a directory', job)
         cleanup_webservice(conf, tmpdir)
@@ -405,14 +405,14 @@ def test_preprocess_failure(self):
         job = web.get_job_by_name('FAILED', 'fail-preprocess')
         failjobdir = os.path.join(conf.directories['FAILED'], 'fail-preprocess')
         self.assertEqual(job.directory, failjobdir)
-        self.assertEqual(job._metadata['runner_id'], None)
+        self.assertIsNone(job._metadata['runner_id'])
         self.assert_fail_msg('Python exception:.*Traceback.*' \
                              + 'ValueError: Failure in preprocessing', job)
         os.rmdir(failjobdir)
         cleanup_webservice(conf, tmpdir)
         # Make sure that the admin got a failed job email
         mail = conf.get_mail_output()
-        self.assert_(re.search('Subject: .*From: testadmin.*To: testadmin' \
+        self.assertTrue(re.search('Subject: .*From: testadmin.*To: testadmin' \
                                + '.*Failure in preprocessing', mail,
                                flags=re.DOTALL),
                      'Unexpected mail output: ' + mail)
@@ -427,19 +427,19 @@ def test_preprocess_logging(self):
         job = web.get_job_by_name('RUNNING', 'log-preprocess')
         # Make sure that the logger file was closed
         for f in testutil.get_open_files():
-            self.assert_('framework.log' not in f,
+            self.assertTrue('framework.log' not in f,
                          "log file %s is still open" % f)
         # Check get_log_handler method
         hdlr = job.get_log_handler()
-        self.assert_(isinstance(hdlr, logging.Handler))
+        self.assertIsInstance(hdlr, logging.Handler)
         jobdir = os.path.join(conf.directories['RUNNING'], 'log-preprocess')
         self.assertEqual(job.directory, jobdir)
         # Both preprocess and run methods in MyJob should have triggered
         os.unlink(os.path.join(jobdir, 'preproc'))
         os.unlink(os.path.join(jobdir, 'job-output'))
         # All logging messages above the threshold should be in framework.log
         logs = open(os.path.join(jobdir, 'framework.log')).read()
-        self.assert_(re.match(
+        self.assertTrue(re.match(
                     '\d+\-\d+\-\d+ \d+:\d+:\d+,\d+ WARNING warning message\n'
                     '\d+\-\d+\-\d+ \d+:\d+:\d+,\d+ ERROR error message\n'
                     '\d+\-\d+\-\d+ \d+:\d+:\d+,\d+ CRITICAL critical message\n',
@@ -476,9 +476,9 @@ def test_preprocess_complete(self):
         os.unlink(os.path.join(compjobdir, 'preproc'))
         os.unlink(os.path.join(compjobdir, 'complete'))
         self.assertEqual(job._metadata['testfield'], 'complete')
-        self.assertEqual(job._metadata['runner_id'], None)
-        self.assertEqual(job._metadata['run_time'], None)
-        self.assertEqual(job._metadata['postprocess_time'], None)
+        self.assertIsNone(job._metadata['runner_id'])
+        self.assertIsNone(job._metadata['run_time'])
+        self.assertIsNone(job._metadata['postprocess_time'])
         os.rmdir(compjobdir)
         cleanup_webservice(conf, tmpdir)
 
@@ -493,7 +493,7 @@ def test_run_failure(self):
         failjobdir = os.path.join(conf.directories['FAILED'], 'fail-run')
         self.assertEqual(job.directory, failjobdir)
         self.assertEqual(job._metadata['testfield'], 'preprocess')
-        self.assertEqual(job._metadata['runner_id'], None)
+        self.assertIsNone(job._metadata['runner_id'])
         self.assert_fail_msg('Python exception:.*Traceback.*' \
                              + 'ValueError: Failure in running', job)
         # Just the preprocess method in MyJob should have triggered
@@ -515,10 +515,10 @@ def test_ok_complete(self):
         self.assertEqual(id(job.config), id(conf))
         # New fields should have been populated in the database
         self.assertEqual(job._metadata['testfield'], 'complete')
-        self.assertNotEqual(job._metadata['postprocess_time'], None)
-        self.assertNotEqual(job._metadata['end_time'], None)
-        self.assertNotEqual(job._metadata['archive_time'], None)
-        self.assertNotEqual(job._metadata['expire_time'], None)
+        self.assertIsNotNone(job._metadata['postprocess_time'])
+        self.assertIsNotNone(job._metadata['end_time'])
+        self.assertIsNotNone(job._metadata['archive_time'])
+        self.assertIsNotNone(job._metadata['expire_time'])
         # postprocess, finalize, complete methods in MyJob should have triggered
         os.unlink(os.path.join(compjobdir, 'postproc'))
         os.unlink(os.path.join(compjobdir, 'finalize'))
@@ -529,7 +529,7 @@ def test_ok_complete(self):
         cleanup_webservice(conf, tmpdir)
         # User should have been notified by email
         mail = conf.get_mail_output()
-        self.assert_(re.search('Subject: .*From: testadmin.*To: testuser' \
+        self.assertTrue(re.search('Subject: .*From: testadmin.*To: testuser' \
                                + '.*Your job job1 has finished.*http://testurl',
                                mail, flags=re.DOTALL),
                      'Unexpected mail output: ' + mail)
@@ -546,8 +546,8 @@ def test_complete_no_expire(self):
         jobdir = os.path.join(conf.directories['COMPLETED'], 'job1')
         self.assertEqual(job.directory, jobdir)
         # archive/expire times should still be NULL
-        self.assertEqual(job._metadata['archive_time'], None)
-        self.assertEqual(job._metadata['expire_time'], None)
+        self.assertIsNone(job._metadata['archive_time'])
+        self.assertIsNone(job._metadata['expire_time'])
         os.unlink(os.path.join(jobdir, 'postproc'))
         os.unlink(os.path.join(jobdir, 'finalize'))
         os.unlink(os.path.join(jobdir, 'complete'))
@@ -720,7 +720,7 @@ def test_ok_expire(self):
 
         # Job should now have moved from ARCHIVED to EXPIRED
         job = web.get_job_by_name('EXPIRED', 'job1')
-        self.assertEqual(job.directory, None)
+        self.assertIsNone(job.directory)
         self.assertEqual(job._metadata['testfield'], 'expire')
         # expire method in MyJob should have triggered
         os.unlink('expire')
@@ -749,7 +749,7 @@ def test_expire_failure(self):
         # Job should now have moved from ARCHIVED to FAILED
         job = web.get_job_by_name('FAILED', 'fail-expire')
         # Job directory should be None since EXPIRED state was visited
-        self.assertEqual(job.directory, None)
+        self.assertIsNone(job.directory)
         self.assert_fail_msg('Python exception:.*Traceback.*' \
                              + 'ValueError: Failure in expiry', job)
         cleanup_webservice(conf, tmpdir)
@@ -783,10 +783,10 @@ def test_delete_all(self):
         open(os.path.join(db.config.directories['PREPROCESSING'],
                           'tempfile'), 'w').write('foo')
         web._delete_all_jobs()
-        self.assertEqual(web.get_job_by_name('FAILED', 'job1'), None)
-        self.assertEqual(web.get_job_by_name('INCOMING', 'job2'), None)
-        self.assertEqual(web.get_job_by_name('EXPIRED', 'job3'), None)
-        self.assertEqual(web.get_job_by_name('RUNNING', 'job4'), None)
+        self.assertIsNone(web.get_job_by_name('FAILED', 'job1'))
+        self.assertIsNone(web.get_job_by_name('INCOMING', 'job2'))
+        self.assertIsNone(web.get_job_by_name('EXPIRED', 'job3'))
+        self.assertIsNone(web.get_job_by_name('RUNNING', 'job4'))
         for typ in ('INCOMING', 'FAILED', 'RUNNING'):
             g = glob.glob(os.path.join(db.config.directories['INCOMING'], '*'))
             self.assertEqual(g, [])
@@ -804,10 +804,10 @@ def test_delete(self):
         job.delete()
         # Jobs should now be removed
         job = web.get_job_by_name('FAILED', 'job1')
-        self.assertEqual(job, None)
+        self.assertIsNone(job)
         job = web.get_job_by_name('EXPIRED', 'job2')
-        self.assertEqual(job, None)
-        self.assert_(not os.path.exists(injobdir))
+        self.assertIsNone(job)
+        self.assertFalse(os.path.exists(injobdir))
         # Make sure that the database rows really went away
         c = db.conn.cursor()
         c.execute('SELECT COUNT(*) FROM jobs')
@@ -861,13 +861,13 @@ def test_reschedule(self):
         os.unlink(os.path.join(jobdir, 'batch_complete'))
         # Rescheduled jobs should *not* set run_time but should set
         # postprocess_time
-        self.assertEqual(job._metadata['run_time'], None)
-        self.assertNotEqual(job._metadata['postprocess_time'], None)
+        self.assertIsNone(job._metadata['run_time'])
+        self.assertIsNotNone(job._metadata['postprocess_time'])
         os.rmdir(jobdir)
         cleanup_webservice(conf, tmpdir)
         # User should *not* been notified by email
         mail = conf.get_mail_output()
-        self.assertEqual(mail, None)
+        self.assertIsNone(mail)
 
     def test_reschedule_run(self):
         """Check Job.reschedule_run method"""

test/backend/test_local_runner.py

@@ -26,7 +26,7 @@ def test_run_wait(self):
         ws = DummyWebService()
         r = LocalRunner(['/bin/sleep', '60'])
         pid = r._run(ws)
-        self.assert_(isinstance(pid, str))
+        self.assertIsInstance(pid, str)
         # Give the waiter thread enough time to start up
         for i in range(20):
             if pid in LocalRunner._waited_jobs:
@@ -44,7 +44,7 @@ def test_run_wait(self):
         # Make sure that non-zero return code causes a job failure
         event = ws._event_queue.get(timeout=0)
         event.process()
-        self.assert_(isinstance(ws._exception, OSError))
+        self.assertIsInstance(ws._exception, OSError)
         ws._exception = None
 
         # Check successful completion
@@ -62,7 +62,7 @@ def test_run_wait(self):
             self.assertEqual(event2.runid, pid1)
         self.assertEqual(type(event1.runner), LocalRunner)
         self.assertEqual(type(event2.runner), LocalRunner)
-        self.assertEqual(ws._exception, None)
+        self.assertIsNone(ws._exception)
 
     def test_run_directory(self):
         """Make sure that LocalRunner runs in the right directory"""

test/backend/test_logging.py

@@ -12,15 +12,15 @@ def test_init(self):
         """Check create of _DelayFileStream objects"""
         dfs = _DelayFileStream('foo')
         self.assertEqual(dfs.filename, os.path.join(os.getcwd(), 'foo'))
-        self.assertEqual(dfs.stream, None)
-        self.assertEqual(os.path.exists('foo'), False)
+        self.assertIsNone(dfs.stream)
+        self.assertFalse(os.path.exists('foo'))
         # flush should be a no-op
         dfs.flush()
-        self.assertEqual(os.path.exists('foo'), False)
+        self.assertFalse(os.path.exists('foo'))
         # File should appear on the first write
         print("test text", file=dfs)
         dfs.flush()
-        self.assertEqual(os.path.exists('foo'), True)
+        self.assertTrue(os.path.exists('foo'))
         contents = open('foo').read()
         self.assertEqual(contents, 'test text\n')
 

test/backend/test_metadata.py

@@ -16,7 +16,7 @@ def test_get(self):
         self.assertEqual(m['key1'], 'value1')
         self.assertEqual(m.get('key1'), 'value1')
         self.assertEqual(m.get('nokey', 'bar'), 'bar')
-        self.assertEqual(m.get('nokey'), None)
+        self.assertIsNone(m.get('nokey'))
         self.assertRaises(KeyError, m.__getitem__, 'nokey')
         k = sorted(m.keys())
         self.assertEqual(k, ['key1', 'key2'])