Skip to content
Snippets Groups Projects
Commit e5d6f33a authored by Maximilian Press's avatar Maximilian Press
Browse files

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.
parent 593f9e2a
No related branches found
No related tags found
No related merge requests found
Pipeline #83944 failed with stage
in 2 hours, 10 minutes, and 46 seconds
......@@ -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__":
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment