Updated obi test to test NUC_SEQS views and the taxonomy API

This commit is contained in:
Celine Mercier
2017-04-21 12:09:04 +02:00
parent c2af955b78
commit 4b037ae236

View File

@ -1,7 +1,9 @@
from obitools3.apps.progress cimport ProgressBar # TODO I absolutely don't understand why it doesn't work without that line
from obitools3.dms.view.view import View, Line_selection
from obitools3.dms.view.typed_view.view_NUC_SEQS import View_NUC_SEQS
from obitools3.dms.dms import DMS
from obitools3.dms.column import Column
from obitools3.dms.taxo.taxo import OBI_Taxonomy
from obitools3.utils cimport str2bytes
import shutil
@ -9,7 +11,7 @@ import string
import random
VIEW_TYPES = [""] #, "NUC_SEQS_VIEW"]
VIEW_TYPES = ["", "NUC_SEQS_VIEW"]
#COL_TYPES = ["OBI_BOOL", "OBI_CHAR", "OBI_FLOAT", "OBI_INT", "OBI_SEQ", "OBI_STR"]
COL_TYPES = [1, 2, 3, 4, 6, 7]
NUC_SEQUENCE_COLUMN = "NUC_SEQ"
@ -17,7 +19,9 @@ ID_COLUMN = "ID"
DEFINITION_COLUMN = "DEFINITION"
QUALITY_COLUMN = "QUALITY"
SPECIAL_COLUMNS = [NUC_SEQUENCE_COLUMN, ID_COLUMN, DEFINITION_COLUMN, QUALITY_COLUMN]
#TAXDUMP = "" TODO path=?
TAXTEST = "taxtest"
NAME_MAX_LEN = 200
COL_COMMENTS_MAX_LEN = 2048
@ -25,40 +29,61 @@ MAX_INT = 2147483647 # used to generate random float values
__title__="Tests if the obitools are working properly"
default_config = {
}
def test_taxo(config, infos):
tax1 = OBI_Taxonomy.open(infos['dms'], config['obi']['taxo'], taxdump=True)
tax1.write(TAXTEST)
tax2 = OBI_Taxonomy.open(infos['dms'], TAXTEST, taxdump=False)
assert len(tax1) == len(tax2), "Length of written taxonomy != length of read taxdump : "+str(len(tax2))+" != "+str(len(tax1))
i = 0
for x in range(config['test']['nbtests']):
idx = random.randint(0, len(tax1)-1)
t1 = tax1.get_taxon_by_idx(idx)
t2 = tax2.get_taxon_by_idx(idx)
assert t1 == t2, "Taxon gotten from written taxonomy != taxon read from taxdump : "+str(t2)+" != "+str(t1)
i+=1
if (i%(config['test']['nbtests']/10)) == 0 :
print("Testing taxonomy functions......"+str(i*100/config['test']['nbtests'])+"%")
tax1.close()
tax2.close()
def random_length(max_len):
return random.randint(1, max_len)
def random_bool(config):
return random.choice([True, False])
def random_char(config):
return str2bytes(random.choice(string.ascii_lowercase))
def random_float(config):
return random.randint(0, MAX_INT) + random.random()
def random_int(config):
return random.randint(0, config['test']['maxlinenb'])
def random_seq(config):
return str2bytes(''.join(random.choice(['a','t','g','c']) for i in range(random_length(config['test']['seqmaxlen']))))
def random_bytes(config):
return random_bytes_with_max_len(config['test']['strmaxlen'])
def random_str_with_max_len(max_len):
return ''.join(random.choice(string.ascii_lowercase) for i in range(random_length(max_len)))
@ -245,8 +270,11 @@ def random_new_view(config, infos, first=False):
elif v_to_clone is not None :
infos['view'] = v_to_clone.clone(random_unique_name(infos), comments=random_str_with_max_len(config['test']['commentsmaxlen']))
else :
infos['view'] = View.new(infos['dms'], random_unique_name(infos), comments=random_str_with_max_len(config['test']['commentsmaxlen'])) # TODO random view class
if v_type == "NUC_SEQS_VIEW" :
infos['view'] = View_NUC_SEQS.new(infos['dms'], random_unique_name(infos), comments=random_str_with_max_len(config['test']['commentsmaxlen'])) # TODO quality column
else :
infos['view'] = View.new(infos['dms'], random_unique_name(infos), comments=random_str_with_max_len(config['test']['commentsmaxlen'])) # TODO quality column
print_test(config, repr(infos['view']))
if v_to_clone is not None :
if line_selection is None:
@ -280,7 +308,14 @@ def addOptions(parser):
type=str,
help="Name of the default DMS for reading and writing data. "
"Default: /tmp/test_dms")
group.add_argument('--taxo','-t',
action="store", dest="obi:taxo",
metavar='<TAXDUMP PATH>',
type=str,
help="Path to a taxdump to test the taxonomy.") # TODO
group=parser.add_argument_group('obi test specific options')
group.add_argument('--nb_tests','-n',
@ -299,7 +334,7 @@ def addOptions(parser):
help="Maximum length of DNA sequences. "
"Default: 200")
group.add_argument('--str_max_len','-t',
group.add_argument('--str_max_len','-r',
action="store", dest="test:strmaxlen",
metavar='<STR_MAX_LEN>',
default=200,
@ -370,7 +405,7 @@ def run(config):
print("Initializing the DMS and the first view...")
shutil.rmtree(config['obi']['defaultdms']+'.obidms', ignore_errors=True)
ini_dms_and_first_view(config, infos)
print_test(config, repr(infos['view']))
@ -383,7 +418,9 @@ def run(config):
print("Testing......"+str(i*100/config['test']['nbtests'])+"%")
#print(infos)
test_taxo(config, infos)
infos['view'].close()
infos['dms'].close()
shutil.rmtree(config['obi']['defaultdms']+'.obidms', ignore_errors=True)