diff --git a/datatypes_conf.xml.sample b/datatypes_conf.xml.sample index 8c6548f4728..3d4d25feb68 100644 --- a/datatypes_conf.xml.sample +++ b/datatypes_conf.xml.sample @@ -1,15 +1,15 @@ - + - + - + @@ -49,12 +49,13 @@ - + + - + @@ -190,6 +191,7 @@ defined format first, followed by next-most rigidly defined, and so on. --> + diff --git a/lib/galaxy/datatypes/binary.py b/lib/galaxy/datatypes/binary.py new file mode 100644 index 00000000000..7a61bc4222b --- /dev/null +++ b/lib/galaxy/datatypes/binary.py @@ -0,0 +1,156 @@ +""" +Binary classes +""" + +import data, logging, binascii +from galaxy.datatypes.metadata import MetadataElement +from galaxy.datatypes import metadata +from galaxy.datatypes.sniff import * +from urllib import urlencode, quote_plus +import zipfile +import os, subprocess, tempfile + +log = logging.getLogger(__name__) + +sniffable_binary_formats = [ 'sff' ] +# Currently these supported binary data types must be manually set on upload +unsniffable_binary_formats = [ 'ab1', 'scf' ] + +class Binary( data.Data ): + """Binary data""" + def set_peek( self, dataset ): + """Set the peek and blurb text""" + if not dataset.dataset.purged: + dataset.peek = 'binary data' + dataset.blurb = 'data' + else: + dataset.peek = 'file does not exist' + dataset.blurb = 'file purged from disk' + +class Ab1( Binary ): + """Class describing an ab1 binary sequence file""" + file_ext = "ab1" + def set_peek( self, dataset ): + if not dataset.dataset.purged: + export_url = "/history_add_to?" + urlencode( {'history_id':dataset.history_id,'ext':'ab1','name':'ab1 sequence','info':'Sequence file','dbkey':dataset.dbkey} ) + dataset.peek = "Binary ab1 sequence file" + dataset.blurb = data.nice_size( dataset.get_size() ) + else: + dataset.peek = 'file does not exist' + dataset.blurb = 'file purged from disk' + def display_peek( self, dataset ): + try: + return dataset.peek + except: + return "Binary ab1 sequence file (%s)" % ( data.nice_size( dataset.get_size() ) ) + +class Bam( Binary ): + """Class describing a BAM binary file""" + file_ext = "bam" + MetadataElement( name="bam_index", desc="BAM Index File", param=metadata.FileParameter, readonly=True, no_value=None, visible=False, optional=True ) + def init_meta( self, dataset, copy_from=None ): + Binary.init_meta( self, dataset, copy_from=copy_from ) + def set_meta( self, dataset, overwrite = True, **kwd ): + """ + Sets index for BAM file. + """ + index_file = dataset.metadata.bam_index + if not index_file: + index_file = dataset.metadata.spec['bam_index'].param.new_file( dataset = dataset ) + tmp_dir = tempfile.gettempdir() + tmpf1 = tempfile.NamedTemporaryFile( dir=tmp_dir ) + tmpf1bai = '%s.bai' % tmpf1.name + try: + os.system( 'cd %s' % tmp_dir ) + os.system( 'cp %s %s' % ( dataset.file_name, tmpf1.name ) ) + os.system( 'samtools index %s' % tmpf1.name ) + os.system( 'cp %s %s' % ( tmpf1bai, index_file.file_name ) ) + except Exception, ex: + sys.stderr.write( 'There was a problem creating the index for the BAM file\n%s\n' + str( ex ) ) + tmpf1.close() + if os.path.exists( tmpf1bai ): + os.remove( tmpf1bai ) + dataset.metadata.bam_index = index_file + def set_peek( self, dataset ): + if not dataset.dataset.purged: + export_url = "/history_add_to?" + urlencode( {'history_id':dataset.history_id,'ext':'bam','name':'bam alignments','info':'Alignments file','dbkey':dataset.dbkey} ) + dataset.peek = "Binary bam alignments file" + dataset.blurb = data.nice_size( dataset.get_size() ) + else: + dataset.peek = 'file does not exist' + dataset.blurb = 'file purged from disk' + def display_peek( self, dataset ): + try: + return dataset.peek + except: + return "Binary bam alignments file (%s)" % ( data.nice_size( dataset.get_size() ) ) + def get_mime( self ): + """Returns the mime type of the datatype""" + return 'application/octet-stream' + +class Binseq( Binary ): + """Class describing a zip archive of binary sequence files""" + file_ext = "binseq.zip" + def set_peek( self, dataset ): + if not dataset.dataset.purged: + zip_file = zipfile.ZipFile( dataset.file_name, "r" ) + num_files = len( zip_file.namelist() ) + dataset.peek = "Archive of %s binary sequence files" % ( str( num_files ) ) + dataset.blurb = data.nice_size( dataset.get_size() ) + else: + dataset.peek = 'file does not exist' + dataset.blurb = 'file purged from disk' + def display_peek( self, dataset ): + try: + return dataset.peek + except: + return "Binary sequence file archive (%s)" % ( data.nice_size( dataset.get_size() ) ) + def get_mime( self ): + """Returns the mime type of the datatype""" + return 'application/zip' + +class Scf( Binary ): + """Class describing an scf binary sequence file""" + file_ext = "scf" + def set_peek( self, dataset ): + if not dataset.dataset.purged: + export_url = "/history_add_to?" + urlencode({'history_id':dataset.history_id,'ext':'scf','name':'scf sequence','info':'Sequence file','dbkey':dataset.dbkey}) + dataset.peek = "Binary scf sequence file" + dataset.blurb = data.nice_size( dataset.get_size() ) + else: + dataset.peek = 'file does not exist' + dataset.blurb = 'file purged from disk' + def display_peek( self, dataset ): + try: + return dataset.peek + except: + return "Binary scf sequence file (%s)" % ( data.nice_size( dataset.get_size() ) ) + +class Sff( Binary ): + """ Standard Flowgram Format (SFF) """ + file_ext = "sff" + def __init__( self, **kwd ): + Binary.__init__( self, **kwd ) + def sniff( self, filename ): + # The first 4 bytes of any sff file is '.sff', and the file is binary. For details + # about the format, see http://www.ncbi.nlm.nih.gov/Traces/trace.cgi?cmd=show&f=formats&m=doc&s=format + try: + header = open( filename ).read(4) + if binascii.b2a_hex( header ) == binascii.hexlify( '.sff' ): + return True + return False + except Exception, e: + return False + def set_peek( self, dataset ): + if not dataset.dataset.purged: + export_url = "/history_add_to?" + urlencode( {'history_id':dataset.history_id,'ext':'sff','name':'sff file','info':'sff file','dbkey':dataset.dbkey} ) + dataset.peek = "Binary sff file" + dataset.blurb = data.nice_size( dataset.get_size() ) + else: + dataset.peek = 'file does not exist' + dataset.blurb = 'file purged from disk' + def display_peek( self, dataset ): + try: + return dataset.peek + except: + return "Binary sff file (%s)" % ( data.nice_size( dataset.get_size() ) ) diff --git a/lib/galaxy/datatypes/data.py b/lib/galaxy/datatypes/data.py index 539aded117b..2c57eedbdfa 100644 --- a/lib/galaxy/datatypes/data.py +++ b/lib/galaxy/datatypes/data.py @@ -1,4 +1,4 @@ -import logging, os, sys, time, tempfile, binascii +import logging, os, sys, time, tempfile from galaxy import util from galaxy.util.odict import odict from galaxy.util.bunch import Bunch @@ -40,20 +40,18 @@ class Data( object ): """ __metaclass__ = DataMeta - - """Add metadata elements""" + # Add metadata elements MetadataElement( name="dbkey", desc="Database/Build", default="?", param=metadata.DBKeyParameter, multiple=False, no_value="?" ) - - """Stores the set of display applications, and viewing methods, supported by this datatype """ + # Stores the set of display applications, and viewing methods, supported by this datatype supported_display_apps = {} - - """If False, the peek is regenerated whenever a dataset of this type is copied""" + # If False, the peek is regenerated whenever a dataset of this type is copied copy_safe_peek = True - - is_binary = True #The dataset contains binary data --> do not space_to_tab or convert newlines, etc. Allow binary file uploads of this type when True. - - allow_datatype_change = True #Allow user to change between this datatype and others. If False, this datatype cannot be changed from or into. - + # The dataset contains binary data --> do not space_to_tab or convert newlines, etc. + # Allow binary file uploads of this type when True. + is_binary = True + # Allow user to change between this datatype and others. If False, this datatype + # cannot be changed from or into. + allow_datatype_change = True #Composite datatypes composite_type = None composite_files = odict() @@ -270,8 +268,6 @@ class Data( object ): def add_composite_file( self, name, **kwds ): #self.composite_files = self.composite_files.copy() self.composite_files[ name ] = self.__new_composite_file( name, **kwds ) - - def __substitute_composite_key( self, key, composite_file, dataset = None ): if composite_file.substitute_name_with_metadata: if dataset: @@ -303,7 +299,6 @@ class Data( object ): return files def generate_auto_primary_file( self, dataset = None ): raise Exception( "generate_auto_primary_file is not implemented for this datatype." ) - @property def has_resolution(self): return False @@ -364,23 +359,37 @@ class Text( Data ): dataset.peek = 'file does not exist' dataset.blurb = 'file purged from disk' -class Binary( Data ): - """Binary data""" +class Txtseq( Data ): + """Class describing a zip archive of text sequence files""" + file_ext = "txtseq.zip" def set_peek( self, dataset ): - """Set the peek and blurb text""" if not dataset.dataset.purged: - dataset.peek = 'binary data' - dataset.blurb = 'data' + zip_file = zipfile.ZipFile( dataset.file_name, "r" ) + num_files = len( zip_file.namelist() ) + dataset.peek = "Archive of %s text sequence files" % ( str( num_files ) ) + dataset.blurb = data.nice_size( dataset.get_size() ) else: dataset.peek = 'file does not exist' dataset.blurb = 'file purged from disk' + def display_peek(self, dataset): + try: + return dataset.peek + except: + return "Text sequence file archive (%s)" % ( data.nice_size( dataset.get_size() ) ) + def get_mime(self): + """Returns the mime type of the datatype""" + return 'application/zip' + +class Newick( Text ): + pass + +# ------------- Utility methods -------------- def get_test_fname( fname ): """Returns test data filename""" path, name = os.path.split(__file__) full_path = os.path.join( path, 'test', fname ) return full_path - def nice_size(size): """ Returns a readably formatted string with the size @@ -406,7 +415,6 @@ def nice_size(size): out = "%.1f %s" % (size, word) return out return '??? bytes' - def get_file_peek( file_name, is_multi_byte=False, WIDTH=256, LINE_COUNT=5 ): """ Returns the first LINE_COUNT lines wrapped to WIDTH @@ -443,7 +451,6 @@ def get_file_peek( file_name, is_multi_byte=False, WIDTH=256, LINE_COUNT=5 ): else: text = unicode( '\n'.join( lines ), 'utf-8' ) return text - def get_line_count(file_name): """Returns the number of lines in a file that are neither null nor comments""" count = 0 @@ -452,38 +459,3 @@ def get_line_count(file_name): if line and line[0] != '#': count += 1 return count - -class Newick( Text ): - pass - -class Sff( Binary ): - """ Standard Flowgram Format (SFF) """ - file_ext = "sff" - def __init__( self, **kwd ): - Binary.__init__(self, **kwd) - def init_meta( self, dataset, copy_from=None ): - Binary.init_meta( self, dataset, copy_from=copy_from ) - def sniff( self, filename ): - ''' - The first 4 bytes of any sff file is '.sff' - - >>> fname = get_test_fname( '1.sff' ) - >>> Sff().sniff( fname ) - True - ''' - header = open( filename ).read(4) - if binascii.b2a_hex( header ) == binascii.hexlify( '.sff' ): - return True - return False - def set_peek( self, dataset ): - if not dataset.dataset.purged: - dataset.peek = "Binary sff file" - dataset.blurb = nice_size( dataset.get_size() ) - else: - dataset.peek = 'file does not exist' - dataset.blurb = 'file purged from disk' - def display_peek(self, dataset): - try: - return dataset.peek - except: - return "sff file (%s)" % ( nice_size( dataset.get_size() ) ) diff --git a/lib/galaxy/datatypes/genetics.py b/lib/galaxy/datatypes/genetics.py index 0e0b739f9eb..cfcd981c516 100644 --- a/lib/galaxy/datatypes/genetics.py +++ b/lib/galaxy/datatypes/genetics.py @@ -48,10 +48,8 @@ class GenomeGraphs(Interval): """Initialize datatype, by adding GBrowse display app""" Interval.__init__(self, **kwd) self.add_display_app ( 'ucsc', 'display at UCSC', 'as_ucsc_display_file', 'ucsc_links' ) - def as_ucsc_display_file( self, dataset, **kwd ): return open( dataset.file_name ) - def set_meta( self, dataset, overwrite = True, **kwd ): i = 0 for i, line in enumerate( file ( dataset.file_name ) ): @@ -66,7 +64,6 @@ class GenomeGraphs(Interval): except: pass Interval.set_meta( self, dataset, overwrite = overwrite, skip = i ) - def make_html_table( self, dataset, skipchars=[] ): """Create HTML table, used for displaying peek""" out = [''] @@ -82,7 +79,6 @@ class GenomeGraphs(Interval): except Exception, exc: out = "Can't create peek %s" % exc return out - def get_estimated_display_viewport( self, dataset ): """ Return a chrom, start, stop tuple for viewing a file. There are slight differences between gff 2 and gff 3 @@ -118,7 +114,6 @@ class GenomeGraphs(Interval): return ( seqid, str( start ), str( stop ) ) else: return ( '', '', '' ) - def gbrowse_links( self, dataset, type, app, base_url ): ret_val = [] if dataset.has_data: @@ -132,7 +127,6 @@ class GenomeGraphs(Interval): link = "%s?start=%s&stop=%s&ref=%s&dbkey=%s" % ( site_url, start, stop, seqid, dataset.dbkey ) ret_val.append( ( site_name, link ) ) return ret_val - def ucsc_links( self, dataset, type, app, base_url ): ret_val = [] if dataset.has_data: @@ -162,8 +156,6 @@ class GenomeGraphs(Interval): else: gal_Log.debug('@@@ gg ucsc_links - no viewport_tuple') return ret_val - - def sniff( self, filename ): """ Determines whether the file is in gff format @@ -202,20 +194,16 @@ class GenomeGraphs(Interval): except: return False - - class rgTabList(Tabular): """ for sampleid and for featureid lists of exclusions or inclusions in the clean tool featureid subsets on statistical criteria -> specialized display such as gg """ file_ext = "rgTList" - def __init__(self, **kwd): """Initialize featurelistt datatype""" Tabular.__init__( self, **kwd ) self.column_names = [] - def make_html_table( self, dataset, skipchars=[] ): """Create HTML table, used for displaying peek""" out = ['
'] @@ -236,7 +224,6 @@ class rgTabList(Tabular): out = "Can't create peek %s" % exc return out - class rgSampleList(rgTabList): """ for sampleid exclusions or inclusions in the clean tool output from QC eg excess het, gender error, ibd pair member,eigen outlier,excess mendel errors,... @@ -252,7 +239,6 @@ class rgSampleList(rgTabList): self.column_names[0] = 'FID' self.column_names[1] = 'IID' # this is what Plink wants as at 2009 - def sniff(self,filename): """ """ @@ -276,26 +262,22 @@ class rgFeatureList( rgTabList ): rgTabList.__init__( self, **kwd ) for i,s in enumerate(['#FeatureId', 'Chr', 'Genpos', 'Mappos']): self.column_names[i] = s - class Rgenetics(Html): """class to use for rgenetics""" - MetadataElement( name="base_name", desc="base name for all transformed versions of this genetic dataset", default="rgenetics", - readonly=True, set_in_upload=True) + MetadataElement( name="base_name", desc="base name for all transformed versions of this genetic dataset", default="rgenetics", readonly=True, set_in_upload=True) composite_type = 'auto_primary_file' allow_datatype_change = False file_ext = 'rgenetics' - def missing_meta( self, dataset=None, **kwargs): """Checks for empty meta values""" for key, value in dataset.metadata.items(): if not value: return True return False - def generate_primary_file( self, dataset = None ): rval = ['Rgenetics Galaxy Composite Dataset

'] rval.append('

This composite dataset is composed of the following files:

    ') @@ -306,7 +288,6 @@ class Rgenetics(Html): rval.append( '
  • %s%s' % ( composite_name, composite_name, opt_text ) ) rval.append( '
' ) return "\n".join( rval ) - def regenerate_primary_file(self,dataset): """cannot do this until we are setting metadata """ @@ -332,12 +313,8 @@ class Rgenetics(Html): f.write("\n".join( rval )) f.write('\n') f.close() - def set_meta( self, dataset, **kwd ): - - """for lped/pbed eg - - """ + """for lped/pbed eg""" if kwd.get('overwrite') == False: if verbose: gal_Log.debug('@@@ rgenetics set_meta called with overwrite = False') @@ -349,9 +326,10 @@ class Rgenetics(Html): gal_Log.debug('@@@rgenetics set_meta failed %s - dataset %s has no efp ?' % (sys.exc_info()[0], dataset.name)) return False try: - flist = os.listdir(efp) - except: - if verbose: gal_Log.debug('@@@rgenetics set_meta failed %s - dataset %s has no efp ?' % (sys.exc_info()[0],dataset.name)) + flist = os.listdir(efp) + except: + if verbose: + gal_Log.debug('@@@rgenetics set_meta failed %s - dataset %s has no efp ?' % (sys.exc_info()[0],dataset.name)) return False if len(flist) == 0: if verbose: @@ -372,7 +350,6 @@ class Rgenetics(Html): dataset.blurb = 'Composite file - Rgenetics Galaxy toolkit' return True - class SNPMatrix(Rgenetics): """fake class to distinguish different species of Rgenetics data collections """ @@ -385,7 +362,6 @@ class SNPMatrix(Rgenetics): else: dataset.peek = 'file does not exist' dataset.blurb = 'file purged from disk' - def sniff(self,filename): """ need to check the file header hex code """ @@ -397,7 +373,6 @@ class SNPMatrix(Rgenetics): else: return True - class Lped(Rgenetics): """fake class to distinguish different species of Rgenetics data collections """ @@ -408,7 +383,6 @@ class Lped(Rgenetics): self.add_composite_file( '%s.ped', description = 'Pedigree File', substitute_name_with_metadata = 'base_name', is_binary = True ) self.add_composite_file( '%s.map', description = 'Map File', substitute_name_with_metadata = 'base_name', is_binary = True ) - class Pphe(Rgenetics): """fake class to distinguish different species of Rgenetics data collections """ @@ -418,7 +392,6 @@ class Pphe(Rgenetics): Rgenetics.__init__(self, **kwd) self.add_composite_file( '%s.pphe', description = 'Plink Phenotype File', substitute_name_with_metadata = 'base_name' ) - class Lmap(Rgenetics): """fake class to distinguish different species of Rgenetics data collections """ @@ -442,8 +415,6 @@ class Phe(Rgenetics): Rgenetics.__init__(self, **kwd) self.add_composite_file( '%s.phe', description = 'Phenotype File', substitute_name_with_metadata = 'base_name' ) - - class Fped(Rgenetics): """fake class to distinguish different species of Rgenetics data collections """ @@ -453,7 +424,6 @@ class Fped(Rgenetics): Rgenetics.__init__(self, **kwd) self.add_composite_file( '%s.fped', description = 'FBAT format pedfile', substitute_name_with_metadata = 'base_name' ) - class Pbed(Rgenetics): """fake class to distinguish different species of Rgenetics data collections """ @@ -475,8 +445,6 @@ class Eigenstratgeno(Rgenetics): self.add_composite_file( '%s.eigenstratgeno', substitute_name_with_metadata = 'base_name', is_binary = True ) self.add_composite_file( '%s.ind', substitute_name_with_metadata = 'base_name', is_binary = True ) self.add_composite_file( '%s.map', substitute_name_with_metadata = 'base_name', is_binary = True ) - - class Eigenstratpca(Rgenetics): """fake class to distinguish different species of Rgenetics data collections @@ -487,20 +455,17 @@ class Eigenstratpca(Rgenetics): Rgenetics.__init__(self, **kwd) self.add_composite_file( '%s.eigenstratpca', description = 'Eigenstrat PCA file', substitute_name_with_metadata = 'base_name' ) - class Snptest(Rgenetics): """fake class to distinguish different species of Rgenetics data collections """ file_ext="snptest" - class Pheno(Tabular): """ base class for pheno files """ file_ext = 'pheno' - class RexpBase( Html ): """base class for BioC data structures in Galaxy must be constructed with the pheno data in place since that @@ -518,18 +483,15 @@ class RexpBase( Html ): composite_type = 'auto_primary_file' allow_datatype_change = False - def __init__( self, **kwd ): Html.__init__(self,**kwd) self.add_composite_file( '%s.pheno', description = 'Phenodata tab text file', substitute_name_with_metadata = 'base_name', is_binary=True) - def generate_primary_file( self, dataset = None ): """ This is called only at upload to write the html file cannot rename the datasets here - they come with the default unfortunately """ return 'AutoGenerated Primary File for Composite Dataset' - def get_phecols(self, phenolist=[], maxConc=20): """ sept 2009: cannot use whitespace to split - make a more complex structure here @@ -601,8 +563,6 @@ class RexpBase( Html ): res = [('no usable phenotype columns found',[('?',0),]),] return res - - def get_pheno(self,dataset): """expects a .pheno file in the extra_files_dir - ugh note that R is wierd and adds the row.name in @@ -620,7 +580,6 @@ class RexpBase( Html ): else: p = [] return '\n'.join(p) - def set_peek( self, dataset ): """expects a .pheno file in the extra_files_dir - ugh note that R is wierd and does not include the row.name in @@ -636,7 +595,6 @@ class RexpBase( Html ): else: dataset.peek = 'file does not exist\n' dataset.blurb = 'file purged from disk' - def get_peek( self, dataset ): """expects a .pheno file in the extra_files_dir - ugh """ @@ -646,7 +604,6 @@ class RexpBase( Html ): except: p = ['##failed to find %s' % pp] return ''.join(p[:5]) - def get_file_peek(self,filename): """ can't really peek at a filename - need the extra_files_path and such? @@ -657,7 +614,6 @@ class RexpBase( Html ): except: pass return ''.join(h[:5]) - def regenerate_primary_file(self,dataset): """cannot do this until we are setting metadata """ @@ -672,18 +628,14 @@ class RexpBase( Html ): f.write("\n".join( rval )) f.write('\n') f.close() - - """Add metadata elements""" def init_meta( self, dataset, copy_from=None ): + """Add metadata elements""" if copy_from: dataset.metadata = copy_from.metadata - def set_meta( self, dataset, **kwd ): - """ NOTE we apply the tabular machinary to the phenodata extracted from a BioC eSet or affybatch. - """ try: flist = os.listdir(dataset.extra_files_path) @@ -727,7 +679,6 @@ class RexpBase( Html ): if not dataset.blurb: dataset.blurb = 'R loadable BioC expression object for the Rexpression Galaxy toolkit' return True - def make_html_table( self, pp='nothing supplied from peek\n'): """Create HTML table, used for displaying peek""" out = ['
',] @@ -750,16 +701,13 @@ class RexpBase( Html ): except Exception, exc: out = "Can't create html table %s" % str( exc ) return out - def display_peek( self, dataset ): """Returns formatted html of peek""" out=self.make_html_table(dataset.peek) return out - def get_mime(self): """Returns the mime type of the datatype""" return 'text/html' - class Affybatch( RexpBase ): """derived class for BioC data structures in Galaxy """ @@ -790,9 +738,6 @@ class MAlist( RexpBase ): self.add_composite_file( '%s.malist', description = 'MAlist R object saved to file', substitute_name_with_metadata = 'base_name', is_binary = True ) - if __name__ == '__main__': import doctest, sys doctest.testmod(sys.modules[__name__]) - - diff --git a/lib/galaxy/datatypes/images.py b/lib/galaxy/datatypes/images.py index 17d0f659a8d..a5e17965446 100644 --- a/lib/galaxy/datatypes/images.py +++ b/lib/galaxy/datatypes/images.py @@ -13,82 +13,6 @@ import os, subprocess, tempfile log = logging.getLogger(__name__) -class Ab1( data.Data ): - """Class describing an ab1 binary sequence file""" - file_ext = "ab1" - def set_peek( self, dataset ): - if not dataset.dataset.purged: - export_url = "/history_add_to?" + urlencode({'history_id':dataset.history_id,'ext':'ab1','name':'ab1 sequence','info':'Sequence file','dbkey':dataset.dbkey}) - dataset.peek = "Binary ab1 sequence file" - dataset.blurb = data.nice_size( dataset.get_size() ) - else: - dataset.peek = 'file does not exist' - dataset.blurb = 'file purged from disk' - def display_peek(self, dataset): - try: - return dataset.peek - except: - return "Binary ab1 sequence file (%s)" % ( data.nice_size( dataset.get_size() ) ) - -class Scf( data.Data ): - """Class describing an scf binary sequence file""" - file_ext = "scf" - def set_peek( self, dataset ): - if not dataset.dataset.purged: - export_url = "/history_add_to?" + urlencode({'history_id':dataset.history_id,'ext':'scf','name':'scf sequence','info':'Sequence file','dbkey':dataset.dbkey}) - dataset.peek = "Binary scf sequence file" - dataset.blurb = data.nice_size( dataset.get_size() ) - else: - dataset.peek = 'file does not exist' - dataset.blurb = 'file purged from disk' - def display_peek(self, dataset): - try: - return dataset.peek - except: - return "Binary scf sequence file (%s)" % ( data.nice_size( dataset.get_size() ) ) - -class Binseq( data.Data ): - """Class describing a zip archive of binary sequence files""" - file_ext = "binseq.zip" - def set_peek( self, dataset ): - if not dataset.dataset.purged: - zip_file = zipfile.ZipFile( dataset.file_name, "r" ) - num_files = len( zip_file.namelist() ) - dataset.peek = "Archive of %s binary sequence files" % ( str( num_files ) ) - dataset.blurb = data.nice_size( dataset.get_size() ) - else: - dataset.peek = 'file does not exist' - dataset.blurb = 'file purged from disk' - def display_peek(self, dataset): - try: - return dataset.peek - except: - return "Binary sequence file archive (%s)" % ( data.nice_size( dataset.get_size() ) ) - def get_mime(self): - """Returns the mime type of the datatype""" - return 'application/zip' - -class Txtseq( data.Data ): - """Class describing a zip archive of text sequence files""" - file_ext = "txtseq.zip" - def set_peek( self, dataset ): - if not dataset.dataset.purged: - zip_file = zipfile.ZipFile( dataset.file_name, "r" ) - num_files = len( zip_file.namelist() ) - dataset.peek = "Archive of %s text sequence files" % ( str( num_files ) ) - dataset.blurb = data.nice_size( dataset.get_size() ) - else: - dataset.peek = 'file does not exist' - dataset.blurb = 'file purged from disk' - def display_peek(self, dataset): - try: - return dataset.peek - except: - return "Text sequence file archive (%s)" % ( data.nice_size( dataset.get_size() ) ) - def get_mime(self): - """Returns the mime type of the datatype""" - return 'application/zip' - class Image( data.Data ): """Class describing an image""" def set_peek( self, dataset ): @@ -236,47 +160,3 @@ class Laj( data.Text ): return dataset.peek except: return "peek unavailable" - -class Bam( data.Binary ): - """Class describing a BAM binary file""" - file_ext = "bam" - MetadataElement( name="bam_index", desc="BAM Index File", param=metadata.FileParameter, readonly=True, no_value=None, visible=False, optional=True ) - def init_meta( self, dataset, copy_from=None ): - data.Binary.init_meta( self, dataset, copy_from=copy_from ) - def set_meta( self, dataset, overwrite = True, **kwd ): - """ - Sets index for BAM file. - """ - index_file = dataset.metadata.bam_index - if not index_file: - index_file = dataset.metadata.spec['bam_index'].param.new_file( dataset = dataset ) - tmp_dir = tempfile.gettempdir() - tmpf1 = tempfile.NamedTemporaryFile(dir=tmp_dir) - tmpf1bai = '%s.bai' % tmpf1.name - try: - os.system('cd %s' % tmp_dir) - os.system('cp %s %s' % (dataset.file_name, tmpf1.name)) - os.system('samtools index %s' % tmpf1.name) - os.system('cp %s %s' % (tmpf1bai, index_file.file_name)) - except Exception, ex: - sys.stderr.write('There was a problem creating the index for the BAM file\n%s\n' + str(ex)) - tmpf1.close() - if os.path.exists(tmpf1bai): - os.remove(tmpf1bai) - dataset.metadata.bam_index = index_file - def set_peek( self, dataset ): - if not dataset.dataset.purged: - export_url = "/history_add_to?" + urlencode({'history_id':dataset.history_id,'ext':'bam','name':'bam alignments','info':'Alignments file','dbkey':dataset.dbkey}) - dataset.peek = "Binary bam alignments file" - dataset.blurb = data.nice_size( dataset.get_size() ) - else: - dataset.peek = 'file does not exist' - dataset.blurb = 'file purged from disk' - def display_peek(self, dataset): - try: - return dataset.peek - except: - return "Binary bam alignments file (%s)" % ( data.nice_size( dataset.get_size() ) ) - def get_mime(self): - """Returns the mime type of the datatype""" - return 'application/octet-stream' diff --git a/lib/galaxy/datatypes/registry.py b/lib/galaxy/datatypes/registry.py index 76d8939f7ff..a49444e6cae 100644 --- a/lib/galaxy/datatypes/registry.py +++ b/lib/galaxy/datatypes/registry.py @@ -3,7 +3,7 @@ Provides mapping between extensions and datatypes, mime-types, etc. """ import os, tempfile import logging -import data, tabular, interval, images, sequence, qualityscore, genetics, xml, coverage, tracks, chrominfo +import data, tabular, interval, images, sequence, qualityscore, genetics, xml, coverage, tracks, chrominfo, binary import galaxy.util from galaxy.util.odict import odict @@ -109,11 +109,11 @@ class Registry( object ): #default values if len(self.datatypes_by_extension) < 1: self.datatypes_by_extension = { - 'ab1' : images.Ab1(), + 'ab1' : binary.Ab1(), 'axt' : sequence.Axt(), - 'bam' : images.Bam(), + 'bam' : binary.Bam(), 'bed' : interval.Bed(), - 'binseq.zip' : images.Binseq(), + 'binseq.zip' : binary.Binseq(), 'blastxml' : xml.BlastXml(), 'coverage' : coverage.LastzCoverage(), 'customtrack' : interval.CustomTrack(), @@ -132,12 +132,12 @@ class Registry( object ): 'qualsolexa' : qualityscore.QualityScoreSolexa(), 'qual454' : qualityscore.QualityScore454(), 'sam' : tabular.Sam(), - 'scf' : images.Scf(), - 'sff' : data.Sff(), + 'scf' : binary.Scf(), + 'sff' : binary.Sff(), 'tabular' : tabular.Tabular(), 'taxonomy' : tabular.Taxonomy(), 'txt' : data.Text(), - 'txtseq.zip' : images.Txtseq(), + 'txtseq.zip' : data.Txtseq(), 'wig' : interval.Wiggle() } self.mimetypes_by_extension = { @@ -174,7 +174,7 @@ class Registry( object ): # because some formats are much more flexibly defined than others. if len(self.sniff_order) < 1: self.sniff_order = [ - data.Sff(), + binary.Sff(), xml.BlastXml(), sequence.Maf(), sequence.Lav(), diff --git a/lib/galaxy/datatypes/tracks.py b/lib/galaxy/datatypes/tracks.py index ed5b534aba6..28eda6c4ea7 100644 --- a/lib/galaxy/datatypes/tracks.py +++ b/lib/galaxy/datatypes/tracks.py @@ -2,11 +2,7 @@ Datatype classes for tracks/track views within galaxy. """ -import data -import logging -import re -import binascii -from cgi import escape +import tabular, binascii, logging from galaxy.datatypes.metadata import MetadataElement from galaxy.datatypes import metadata import galaxy.model @@ -17,7 +13,7 @@ from galaxy.util.hash_util import * log = logging.getLogger(__name__) -class GeneTrack( data.Binary ): +class GeneTrack( tabular.Tabular ): file_ext = "genetrack" MetadataElement( name="genetrack", default="data.genetrack", desc="HDF index", readonly=True, visible=True, no_value=0 ) diff --git a/test/functional/test_get_data.py b/test/functional/test_get_data.py index 196d07d8b4b..36a9b0fac26 100644 --- a/test/functional/test_get_data.py +++ b/test/functional/test_get_data.py @@ -4,128 +4,538 @@ from galaxy.model.mapping import context as sa_session from base.twilltestcase import TwillTestCase class UploadData( TwillTestCase ): - def test_000_upload_files_from_disk( self ): - """Test uploading data files from disk""" + def test_0005_upload_file( self ): + """Test uploading 1.bed, NOT setting the file format""" self.logout() self.login( email='test@bx.psu.edu' ) global admin_user admin_user = sa_session.query( galaxy.model.User ) \ .filter( galaxy.model.User.table.c.email=='test@bx.psu.edu' ) \ .one() - history1 = sa_session.query( galaxy.model.History ) \ - .filter( and_( galaxy.model.History.table.c.deleted==False, - galaxy.model.History.table.c.user_id==admin_user.id ) ) \ - .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ - .first() + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() self.upload_file( '1.bed' ) - hda1 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda1 is not None, "Problem retrieving hda1 from database" - self.verify_dataset_correctness( '1.bed', hid=str( hda1.hid ) ) - self.upload_file( '2.bed', dbkey='hg17' ) - hda2 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda2 is not None, "Problem retrieving hda2 from database" - self.verify_dataset_correctness( '2.bed', hid=str( hda2.hid ) ) - self.upload_file( '3.bed', dbkey='hg17', ftype='bed' ) - hda3 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda3 is not None, "Problem retrieving hda3 from database" - self.verify_dataset_correctness( '3.bed', hid=str( hda3.hid ) ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.bed', hid=str( hda.hid ) ) + self.check_history_for_string( "" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0010_upload_file( self ): + """Test uploading 4.bed.gz, manually setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() self.upload_file( '4.bed.gz', dbkey='hg17', ftype='bed' ) - hda4 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda4 is not None, "Problem retrieving hda4 from database" - self.verify_dataset_correctness( '4.bed', hid=str( hda4.hid ) ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '4.bed', hid=str( hda.hid ) ) + self.check_history_for_string( "" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0015_upload_file( self ): + """Test uploading 1.scf, manually setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() self.upload_file( '1.scf', ftype='scf' ) - hda5 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda5 is not None, "Problem retrieving hda5 from database" - self.verify_dataset_correctness( '1.scf', hid=str( hda5.hid ) ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.scf', hid=str( hda.hid ) ) + self.check_history_for_string( "Binary scf sequence file" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0020_upload_file( self ): + """Test uploading 1.scf, NOT setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '1.scf' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.check_history_for_string( "File Format' to 'Scf' when uploading scf files" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0025_upload_file( self ): + """Test uploading 1.scf.zip, manually setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() self.upload_file( '1.scf.zip', ftype='binseq.zip' ) - hda6 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda6 is not None, "Problem retrieving hda6 from database" - self.verify_dataset_correctness( '1.scf.zip', hid=str( hda6.hid ) ) - self.delete_history( id=self.security.encode_id( history1.id ) ) - def test_005_url_paste( self ): + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.scf.zip', hid=str( hda.hid ) ) + self.check_history_for_string( "Archive of 1 binary sequence files" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0030_upload_file( self ): + """Test uploading 1.scf.zip, NOT setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '1.scf.zip' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.check_history_for_string( "'File Format' for archive consisting of binary files - use 'Binseq.zip'" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0035_upload_file( self ): + """Test uploading 1.sam NOT setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '1.sam' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.sam', hid=str( hda.hid ) ) + self.check_history_for_string( "" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0040_upload_file( self ): + """Test uploading 1.sff, NOT setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '1.sff' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.sff', hid=str( hda.hid ) ) + self.check_history_for_string( 'format: sff' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0045_upload_file( self ): + """Test uploading 454Score.pdf, NOT setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '454Score.pdf' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.check_history_for_string( "The uploaded file contains inappropriate content" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0050_upload_file( self ): + """Test uploading 454Score.png, NOT setting the file format""" + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '454Score.png' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.check_history_for_string( "The uploaded file contains inappropriate content" ) + def test_0055_upload_file( self ): + """Test uploading lped composite datatype file, manually setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + # lped data types include a ped_file and a map_file ( which is binary ) + self.upload_composite_datatype_file( 'lped', ped_file='tinywga.ped', map_file='tinywga.map', base_name='rgenetics' ) + # Get the latest hid for testing + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + # We'll test against the resulting ped file and map file for correctness + self.verify_composite_datatype_file_content( 'rgenetics.ped', str( hda.id ) ) + self.verify_composite_datatype_file_content( 'rgenetics.map', str( hda.id ) ) + self.check_history_for_string( "Uploaded Composite Dataset (lped)" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0060_upload_file( self ): + """Test uploading pbed composite datatype file, manually setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + # pbed data types include a bim_file, a bed_file and a fam_file + self.upload_composite_datatype_file( 'pbed', bim_file='tinywga.bim', bed_file='tinywga.bed', fam_file='tinywga.fam', base_name='rgenetics' ) + # Get the latest hid for testing + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + # We'll test against the resulting ped file and map file for correctness + self.verify_composite_datatype_file_content( 'rgenetics.bim', str( hda.id ) ) + self.verify_composite_datatype_file_content( 'rgenetics.bed', str( hda.id ) ) + self.verify_composite_datatype_file_content( 'rgenetics.fam', str( hda.id ) ) + self.check_history_for_string( "Uploaded Composite Dataset (pbed)" ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0065_upload_file( self ): + """Test uploading asian_chars_1.txt, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( 'asian_chars_1.txt' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( 'asian_chars_1.txt', hid=str( hda.hid ) ) + self.check_history_for_string( 'uploaded multi-byte char file' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0070_upload_file( self ): + """Test uploading 2gen.fastq, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '2gen.fastq' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '2gen.fastq', hid=str( hda.hid ) ) + self.check_history_for_string( '2gen.fastq format: fastq, database: \? Info: uploaded fastq file' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0075_upload_file( self ): + """Test uploading 1.wig, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '1.wig' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.wig', hid=str( hda.hid ) ) + self.check_history_for_string( '1.wig format: wig, database: \? Info: uploaded file' ) + self.check_metadata_for_string( 'value="1.wig" value="\?"' ) + self.check_metadata_for_string( 'Change data type selected value="wig" selected="yes"' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0080_upload_file( self ): + """Test uploading 1.tabular, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '1.tabular' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.tabular', hid=str( hda.hid ) ) + self.check_history_for_string( '1.tabular format: tabular, database: \? Info: uploaded file' ) + self.check_metadata_for_string( 'value="1.tabular" value="\?"' ) + self.check_metadata_for_string( 'Change data type selected value="tabular" selected="yes"' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0085_upload_file( self ): + """Test uploading qualscores.qualsolid, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( 'qualscores.qualsolid' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( 'qualscores.qualsolid', hid=str( hda.hid ) ) + self.check_history_for_string( '2.5 Kb, format: qualsolid, database: \? Info: uploaded file' ) + self.check_metadata_for_string( 'Change data type value="qualsolid" selected="yes">qualsolid' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0090_upload_file( self ): + """Test uploading qualscores.qual454, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( 'qualscores.qual454' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( 'qualscores.qual454', hid=str( hda.hid ) ) + self.check_history_for_string( '5.6 Kb, format: qual454, database: \?' ) + self.check_metadata_for_string( 'Change data type value="qual454" selected="yes">qual454' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0095_upload_file( self ): + """Test uploading 3.maf, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '3.maf' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '3.maf', hid=str( hda.hid ) ) + self.check_history_for_string( '3.maf format: maf, database: \? Info: uploaded file' ) + self.check_metadata_for_string( 'value="3.maf" value="\?"' ) + self.check_metadata_for_string( 'Convert to new format ' ) + self.check_metadata_for_string( 'value="shrimp_cs_test1.csfasta" value="\?" Change data type value="csfasta" selected="yes"' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0140_upload_file( self ): + """Test uploading megablast_xml_parser_test1.gz, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( 'megablast_xml_parser_test1.gz' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.check_history_for_string( 'NCBI Blast XML data format: blastxml' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0145_upload_file( self ): + """Test uploading 1.axt, NOT setting the file format""" + # Logged in as admin_user + self.check_history_for_string( 'Your history is empty' ) + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() + self.upload_file( '1.axt' ) + hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ + .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ + .first() + assert hda is not None, "Problem retrieving hda from database" + self.verify_dataset_correctness( '1.axt', hid=str( hda.hid ) ) + self.check_history_for_string( '1.axt format: axt, database: \? Info: uploaded file' ) + self.check_metadata_for_string( 'value="1.axt" value="\?" Change data type selected value="axt" selected="yes"' ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_0150_url_paste( self ): """Test url paste behavior""" # Logged in as admin_user # Deleting the current history should have created a new history self.check_history_for_string( 'Your history is empty' ) - history2 = sa_session.query( galaxy.model.History ) \ - .filter( and_( galaxy.model.History.table.c.deleted==False, - galaxy.model.History.table.c.user_id==admin_user.id ) ) \ - .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ - .first() + history = sa_session.query( galaxy.model.History ) \ + .filter( and_( galaxy.model.History.table.c.deleted==False, + galaxy.model.History.table.c.user_id==admin_user.id ) ) \ + .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ + .first() self.upload_url_paste( 'hello world' ) self.check_history_for_string( 'Pasted Entry' ) self.check_history_for_string( 'hello world' ) self.upload_url_paste( u'hello world' ) self.check_history_for_string( 'Pasted Entry' ) self.check_history_for_string( 'hello world' ) - self.delete_history( id=self.security.encode_id( history2.id ) ) - def test_010_upload_lped_composite_datatype_files( self ): - """Test uploading lped composite datatype files""" - # Logged in as admin_user - self.check_history_for_string( 'Your history is empty' ) - history3 = sa_session.query( galaxy.model.History ) \ - .filter( and_( galaxy.model.History.table.c.deleted==False, - galaxy.model.History.table.c.user_id==admin_user.id ) ) \ - .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ - .first() - # lped data types include a ped_file and a map_file ( which is binary ) - self.upload_composite_datatype_file( 'lped', ped_file='tinywga.ped', map_file='tinywga.map', base_name='rgenetics' ) - # Get the latest hid for testing - hda1 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda1 is not None, "Problem retrieving hda1 from database" - # We'll test against the resulting ped file and map file for correctness - self.verify_composite_datatype_file_content( 'rgenetics.ped', str( hda1.id ) ) - self.verify_composite_datatype_file_content( 'rgenetics.map', str( hda1.id ) ) - self.delete_history( id=self.security.encode_id( history3.id ) ) - def test_015_upload_pbed_composite_datatype_files( self ): - """Test uploading pbed composite datatype files""" - # Logged in as admin_user - self.check_history_for_string( 'Your history is empty' ) - history4 = sa_session.query( galaxy.model.History ) \ - .filter( and_( galaxy.model.History.table.c.deleted==False, - galaxy.model.History.table.c.user_id==admin_user.id ) ) \ - .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ - .first() - # pbed data types include a bim_file, a bed_file and a fam_file - self.upload_composite_datatype_file( 'pbed', bim_file='tinywga.bim', bed_file='tinywga.bed', fam_file='tinywga.fam', base_name='rgenetics' ) - # Get the latest hid for testing - hda1 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda1 is not None, "Problem retrieving hda1 from database" - # We'll test against the resulting ped file and map file for correctness - self.verify_composite_datatype_file_content( 'rgenetics.bim', str( hda1.id ) ) - self.verify_composite_datatype_file_content( 'rgenetics.bed', str( hda1.id ) ) - self.verify_composite_datatype_file_content( 'rgenetics.fam', str( hda1.id ) ) - self.delete_history( id=self.security.encode_id( history4.id ) ) - def test_020_upload_multibyte_character_file( self ): - """Test uploading multi-byte character file""" - # Logged in as admin_user - self.check_history_for_string( 'Your history is empty' ) - history5 = sa_session.query( galaxy.model.History ) \ - .filter( and_( galaxy.model.History.table.c.deleted==False, - galaxy.model.History.table.c.user_id==admin_user.id ) ) \ - .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ - .first() - self.upload_file( 'asian_chars_1.txt' ) - hda1 = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert hda1 is not None, "Problem retrieving hda1 from database" - self.verify_dataset_correctness( 'asian_chars_1.txt', hid=str( hda1.hid ) ) - self.check_history_for_string( 'uploaded multi-byte char file' ) - self.delete_history( id=self.security.encode_id( history5.id ) ) + self.delete_history( id=self.security.encode_id( history.id ) ) + def test_9999_clean_up( self ): + self.logout() \ No newline at end of file diff --git a/test/functional/test_sniffing_and_metadata_settings.py b/test/functional/test_sniffing_and_metadata_settings.py deleted file mode 100644 index 4db99b13924..00000000000 --- a/test/functional/test_sniffing_and_metadata_settings.py +++ /dev/null @@ -1,262 +0,0 @@ -import galaxy.model -from galaxy.model.orm import * -from galaxy.model.mapping import context as sa_session -from base.twilltestcase import TwillTestCase - -class SniffingAndMetaDataSettings( TwillTestCase ): - def test_000_axt_datatype( self ): - """Testing correctly sniffing axt data type upon upload""" - self.logout() - self.login( email='test@bx.psu.edu' ) - global admin_user - admin_user = sa_session.query( galaxy.model.User ).filter( galaxy.model.User.table.c.email=='test@bx.psu.edu' ).one() - self.new_history( name='history1' ) - global history1 - history1 = sa_session.query( galaxy.model.History ) \ - .filter( and_( galaxy.model.History.table.c.deleted==False, - galaxy.model.History.table.c.user_id==admin_user.id ) ) \ - .order_by( desc( galaxy.model.History.table.c.create_time ) ) \ - .first() - assert history1 is not None, "Problem retrieving history1 from database" - self.upload_file( '1.axt' ) - self.verify_dataset_correctness( '1.axt' ) - self.check_history_for_string( '1.axt format: axt, database: \? Info: uploaded file' ) - self.check_metadata_for_string( 'value="1.axt" value="\?" Change data type selected value="axt" selected="yes"' ) - latest_hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert latest_hda is not None, "Problem retrieving axt hda from the database" - if not latest_hda.name == '1.axt' and not latest_hda.extension == 'axt': - raise AssertionError, "axt data type was not correctly sniffed." - def test_005_bed_datatype( self ): - """Testing correctly sniffing bed data type upon upload""" - self.upload_file( '1.bed' ) - self.verify_dataset_correctness( '1.bed' ) - self.check_history_for_string( '1.bed format: bed, database: \? Info: uploaded file') - self.check_metadata_for_string( 'value="1.bed" value="\?"' ) - self.check_metadata_for_string( 'Chrom column: ' ) - self.check_metadata_for_string( 'value="shrimp_cs_test1.csfasta" value="\?" Change data type value="csfasta" selected="yes"' ) - latest_hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert latest_hda is not None, "Problem retrieving csfasta hda from the database" - if not latest_hda.name == 'shrimp_cs_test1.csfasta' and not latest_hda.extension == 'csfasta': - raise AssertionError, "csfasta data type was not correctly sniffed." - def test_020_customtrack_datatype( self ): - """Testing correctly sniffing customtrack data type upon upload""" - self.upload_file( '1.customtrack' ) - self.verify_dataset_correctness( '1.customtrack' ) - self.check_history_for_string( '1.customtrack format: customtrack, database: \? Info: uploaded file' ) - self.check_metadata_for_string( 'value="1.customtrack" value="\?" Change data type selected value="customtrack" selected="yes"' ) - latest_hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert latest_hda is not None, "Problem retrieving customtrack hda from the database" - if not latest_hda.name == '1.customtrack' and not latest_hda.extension == 'customtrack': - raise AssertionError, "customtrack data type was not correctly sniffed." - def test_025_fasta_datatype( self ): - """Testing correctly sniffing fasta data type upon upload""" - self.upload_file( '1.fasta' ) - self.verify_dataset_correctness( '1.fasta' ) - self.check_history_for_string( '1.fasta format: fasta, database: \? Info: uploaded file' ) - self.check_metadata_for_string( 'value="1.fasta" value="\?" Change data type selected value="fasta" selected="yes"' ) - latest_hda = sa_session.query( galaxy.model.HistoryDatasetAssociation ) \ - .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \ - .first() - assert latest_hda is not None, "Problem retrieving fasta hda from the database" - if not latest_hda.name == '1.fasta' and not latest_hda.extension == 'fasta': - raise AssertionError, "fasta data type was not correctly sniffed." - def test_035_gff_datatype( self ): - """Testing correctly sniffing gff data type upon upload""" - self.upload_file( '5.gff' ) - self.verify_dataset_correctness( '5.gff' ) - self.check_history_for_string( '5.gff format: gff, database: \? Info: uploaded file' ) - self.check_metadata_for_string( 'value="5.gff" value="\?"' ) - self.check_metadata_for_string( 'Convert to new format
1.Chrom2.Start3.End1.Chrom2.Start3.End1.QNAME2.FLAG3.RNAME4.POS>2_14_26_F3,-1282216.0>2_14_26_F3,-1282216.0