From e5d6f33a01153c33ae1cbe0a105e0bc9e309c00f Mon Sep 17 00:00:00 2001
From: Maximilian Press <max@phasegenomics.com>
Date: Fri, 2 Apr 2021 12:38:22 -0700
Subject: [PATCH] Unit test fixes accommodating bugfix
* changes to unit-tests/test_functions_PhageTerm_for_multi.test_checkpoint_creation() accommodating that the first read is now considered. e.g. total match length goes from 291333 --> 291441 due to first read length of 108.
* minor restructure of unit tests to include standard setUp() and tearDown() methods, simplifying the creation/deletion of tmp dirs. This means that the unit tests don't fail if they detect leftover tmp directories leftover from a previous run that didn't cleanup because it failed.
---
.../test_functions_PhageTerm_for_multi.py | 42 +++++++++----------
1 file changed, 20 insertions(+), 22 deletions(-)
diff --git a/unit-tests/test_functions_PhageTerm_for_multi.py b/unit-tests/test_functions_PhageTerm_for_multi.py
index a7c723d..8fa7326 100755
--- a/unit-tests/test_functions_PhageTerm_for_multi.py
+++ b/unit-tests/test_functions_PhageTerm_for_multi.py
@@ -22,11 +22,18 @@ from _modules.debug_utils import ReadMappingInfoLogger
class Test_functions_Phageterm (unittest.TestCase):
# Test that readsCoverage results are saved in the appropriate structure and can be read again for later work.
+ def setUp(self):
+ self.dir_tmp = os.path.join(os.getcwd(), "tmp")
+ os.mkdir(self.dir_tmp)
+
+ def tearDown(self):
+ # this will remove these dirs at end of each test, even if test fails.
+ shutil.rmtree(self.dir_tmp)
+
def testDumpAndReadAgainRes(self):
l_data = buildTestData()
chk_handler = RCCheckpoint_handler(0,None,False)
- dir_cov_res = os.path.join(os.getcwd(),"tmp")
- os.mkdir(os.path.join(os.getcwd(),"tmp")) # TODO: when switching to python3, use the tempfile module
+ dir_cov_res = self.dir_tmp # TODO: when switching to python3, use the tempfile module
for d in l_data:
idx_seq = 0
for refseq in d.refseq_list:
@@ -67,7 +74,7 @@ class Test_functions_Phageterm (unittest.TestCase):
self.assertTrue(np.array_equal(r1b.paired_mismatch,r1.paired_mismatch))
self.assertTrue(np.array_equal(r1b.reads_tested,r1.reads_tested))
idx_seq += 1
- shutil.rmtree(dir_cov_res)
+
#
# # VL: start basic testing of sum_reads_coverage_for_seq for multimachine mode
def test_basic_sum_readsCoverage_for_seq(self):
@@ -83,17 +90,16 @@ class Test_functions_Phageterm (unittest.TestCase):
os.mkdir(DR_path)
sum_readsCoverage_for_seq(dir_cov_res, idx_refseq, nb_pieces, d.inDArgs, d.fParms, d.inDRawArgs, dir_seq_res, DR_path)
self.assertTrue(os.path.exists("DR/UNKNOWN/NODE_1_length_422445"))
+ # ideally these would be handled in self.tearDown() but they are not currently a problem.
shutil.rmtree(dir_seq_res)
shutil.rmtree(DR_path)
-
# Checks that checkpoints are created and that their content is correct.
def test_checkpoint_creation(self):
# last line is 11150
d=Data4Test("../test-data/COS-5.fastq", "../test-data/COS-5.fasta")
- dir_chk = os.path.join(os.getcwd(), "tmp2")
- os.mkdir(dir_chk)
+ dir_chk = self.dir_tmp
d.tParms.dir_chk=dir_chk
d.tParms.test_mode=True
return_dict = dict()
@@ -101,26 +107,24 @@ class Test_functions_Phageterm (unittest.TestCase):
chk_handler = RCCheckpoint_handler(d.tParms.chk_freq, d.tParms.dir_chk, d.tParms.test_mode)
readsCoverage(d.inDRawArgs, d.refseq_list[0], d.inDArgs, d.fParms, return_dict, 0, d.line_start, d.line_end, \
d.tParms, chk_handler, 0, None)
- fic_name = "chk_0_0_11150_291333.npz"
+ fic_name = "chk_0_0_11150_291441.npz" # was chk_0_0_11150_291333.npz - length of first read is 108.
full_fname = os.path.join(dir_chk, fic_name)
- self.assertTrue(os.path.exists(full_fname))
list_f = os.listdir(dir_chk)
+ print("this is the list of files", list_f)
+ self.assertTrue(os.path.exists(full_fname))
self.assertTrue(len(list_f)==1)
wr=chk_handler.load(0,0)
- self.assertEqual(wr.read_match,291333)
+ self.assertEqual(wr.read_match,291441) # 291333 + 108 (first read length) = 291441
self.assertEqual(wr.count_line,11150)
self.assertEqual(wr.interm_res.host_len,0)
self.assertEqual(wr.interm_res.gen_len,3012)
- self.assertEqual(int(wr.interm_res.reads_tested),2787) # 2796?
- shutil.rmtree(dir_chk)
-
+ self.assertEqual(int(wr.interm_res.reads_tested),2788) # 2787+1 # orig comment: 2796?
## Checks thst in production mode, all checkpoints are deleted at the end.
def test_checkpoint_end(self):
# last line is 11150
d=Data4Test("../test-data/COS-5.fastq", "../test-data/COS-5.fasta")
- dir_chk = os.path.join(os.getcwd(), "tmp3")
- os.mkdir(dir_chk)
+ dir_chk = self.dir_tmp
d.tParms.chk_freq=1
d.tParms.dir_chk=dir_chk
d.tParms.test_mode=False
@@ -131,13 +135,11 @@ class Test_functions_Phageterm (unittest.TestCase):
d.tParms, chk_handler, 0, None)
list_f = os.listdir(dir_chk)
self.assertTrue(len(list_f)==0)
- shutil.rmtree(dir_chk)
# ## checks that readsCoverage restarts from ceckpoint and not from the beginning.
def test_restart_from_checkpoint(self):
d = Data4Test("../test-data/COS-5.fastq", "../test-data/COS-5.fasta")
- dir_chk = os.path.join(os.getcwd(), "tmp4")
- os.mkdir(dir_chk)
+ dir_chk = self.dir_tmp
d.tParms.dir_chk = dir_chk
d.tParms.chk_freq=1
d.tParms.test_mode = False
@@ -156,12 +158,10 @@ class Test_functions_Phageterm (unittest.TestCase):
self.assertEqual(wr.interm_res.host_len,0)
self.assertEqual(wr.interm_res.gen_len,3012)
self.assertEqual(int(wr.interm_res.reads_tested),9)
- shutil.rmtree(dir_chk)
def test_restart_from_checkpoint2(self):
d=Data4Test("../test-data/Virome.fastq","../test-data/Virome.fasta")
- dir_chk = os.path.join(os.getcwd(), "tmp5")
- os.mkdir(dir_chk)
+ dir_chk = self.dir_tmp
d.tParms.dir_chk = dir_chk
d.tParms.chk_freq = 5
d.tParms.test_mode = False
@@ -179,8 +179,6 @@ class Test_functions_Phageterm (unittest.TestCase):
d.tParms, chk_handler, idx_seq, logger)
idx_seq+=1
self.assertEqual(idx_seq,5)
- shutil.rmtree(dir_chk)
-
if __name__ == "__main__":
--
GitLab