mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-25 00:12:08 +08:00
More moving things around.
This commit is contained in:
James Taylor
@@ -0,0 +1,274 @@
|
||||
import pkg_resources
|
||||
pkg_resources.require('twill')
|
||||
|
||||
import StringIO
|
||||
import os, sys, random, filecmp, time, unittest, urllib, logging, difflib
|
||||
from itertools import *
|
||||
|
||||
import twill
|
||||
import twill.commands as tc
|
||||
from elementtree import ElementTree
|
||||
|
||||
buffer = StringIO.StringIO()
|
||||
|
||||
# Force twill to log to a buffer -- FIXME: Should this go to stdout and be
|
||||
# captured by nose?
|
||||
twill.set_output(buffer)
|
||||
tc.config('use_tidy', 0)
|
||||
|
||||
# Dial ClientCookie logging down (very noisy)
|
||||
logging.getLogger( "ClientCookie.cookies" ).setLevel( logging.WARNING )
|
||||
|
||||
class TwillTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.history_id = os.environ.get( 'GALAXY_TEST_HISTORY_ID', None )
|
||||
self.host = os.environ.get( 'GALAXY_TEST_HOST' )
|
||||
self.port = os.environ.get( 'GALAXY_TEST_PORT' )
|
||||
self.url = "http://%s:%s" % ( self.host, self.port )
|
||||
self.file_dir = os.environ.get( 'GALAXY_TEST_FILE_DIR' )
|
||||
self.home()
|
||||
self.set_history()
|
||||
|
||||
def get_fname(self, fname):
|
||||
full = os.path.join( self.file_dir, fname)
|
||||
return os.path.abspath(full)
|
||||
|
||||
def home(self):
|
||||
tc.go("%s" % self.url)
|
||||
tc.code(200)
|
||||
|
||||
def go2myurl(self, myurl):
|
||||
tc.go("%s" % myurl)
|
||||
# print "+++++++++++++++++++++++++++++"
|
||||
# print tc.show()
|
||||
# print "-----------------------------"
|
||||
tc.code(200)
|
||||
|
||||
def set_history(self):
|
||||
"""Sets the history (stores the cookies for this run)"""
|
||||
if self.history_id:
|
||||
tc.go( "./history?id=%s" % self.history_id )
|
||||
else:
|
||||
tc.go( "./history" )
|
||||
tc.code(200)
|
||||
|
||||
def clear_history(self):
|
||||
tc.go("./history_new")
|
||||
tc.go("./history")
|
||||
tc.find('Your history is empty')
|
||||
|
||||
def upload_file(self, fname, ftype='auto', dbkey='hg17'):
|
||||
"""Uploads a file"""
|
||||
fname = self.get_fname(fname)
|
||||
tc.go("./tool_runner/index?tool_id=upload1")
|
||||
tc.fv("1","file_type", ftype)
|
||||
tc.fv("1","dbkey", dbkey)
|
||||
tc.formfile("1","file_data", fname)
|
||||
tc.submit("runtool_btn")
|
||||
self.home()
|
||||
|
||||
def edit_data(self, hid, check_patt=None, **kwd):
|
||||
"""
|
||||
Edits data and sets parameters accordig to the keyword arguments
|
||||
"""
|
||||
tc.go('./edit?hid=%d' % hid )
|
||||
if check_patt:
|
||||
tc.find(check_patt)
|
||||
if kwd:
|
||||
self.submit_form(form=1, button="edit_genome_btn", **kwd)
|
||||
|
||||
def get_xml_history(self):
|
||||
"""Returns a parsed xml object corresponding to the history"""
|
||||
self.home()
|
||||
tc.go('./history?template=history.xml' )
|
||||
xml = self.last_page()
|
||||
tree = ElementTree.fromstring(xml)
|
||||
return tree
|
||||
|
||||
def get_data_list(self):
|
||||
"""Returns the data at history id 'hid'"""
|
||||
tree = self.get_xml_history()
|
||||
data_list = [ elem for elem in tree.findall("data") ]
|
||||
return data_list
|
||||
|
||||
def delete_data(self, hid):
|
||||
"""Deletes data at a certain history id"""
|
||||
hid = str(hid)
|
||||
data_list = self.get_data_list()
|
||||
self.assertTrue( data_list )
|
||||
elems = [ elem for elem in data_list if elem.get('hid') == hid ]
|
||||
self.assertEqual(len(elems), 1)
|
||||
tc.go("/delete?id=%s" % elems[0].get('id') )
|
||||
tc.code(200)
|
||||
|
||||
|
||||
def get_xml_history_ids(self):
|
||||
"""Returns a parsed xml object corresponding to the history"""
|
||||
self.home()
|
||||
tc.go('./history?template=history_ids.xml' )
|
||||
xml = self.last_page()
|
||||
tree = ElementTree.fromstring(xml)
|
||||
# print "xml=", xml
|
||||
# print "tree=", tree
|
||||
return tree
|
||||
|
||||
def get_history_ids(self):
|
||||
"""Returns the data at history id 'hid'"""
|
||||
tree = self.get_xml_history_ids()
|
||||
data_list = [ elem for elem in tree.findall("data") ]
|
||||
# print "data_list=", data_list
|
||||
return data_list
|
||||
|
||||
def delete_history(self, hid):
|
||||
"""Deletes a history at a certain id"""
|
||||
data_list = self.get_history_ids()
|
||||
self.assertTrue( data_list )
|
||||
if hid < 0:
|
||||
hid = len(data_list) + hid +1
|
||||
print hid
|
||||
hid = str(hid)
|
||||
elems = [ elem for elem in data_list if elem.get('hid') == hid ]
|
||||
self.assertEqual(len(elems), 1)
|
||||
tc.go("/history_delete?id=%s" % elems[0].get('id') )
|
||||
tc.code(200)
|
||||
|
||||
|
||||
def historyid(self):
|
||||
data_list = self.get_data_list()
|
||||
hid_old = -2
|
||||
hids = []
|
||||
same = 0
|
||||
for elem in data_list:
|
||||
hid = elem.get('hid')
|
||||
hids.append(hid)
|
||||
if int(hid) == hid_old + 1:
|
||||
if same == 0 :
|
||||
print "-",
|
||||
same =1
|
||||
else :
|
||||
if hid_old > 0 :
|
||||
print "%d," % hid_old,
|
||||
print "%s" % hid,
|
||||
same = 0
|
||||
hid_old = int(hid)
|
||||
if same == 1 :
|
||||
print "%s" % hid
|
||||
return hids
|
||||
|
||||
def _assert_dataset_state( self, elem, state ):
|
||||
assert elem.get( 'state' ) == state, \
|
||||
"Expecting dataset state '%s' but is '%s'. Dataset blurb: %s" % ( state, elem.get('state'), elem.text.strip() )
|
||||
|
||||
def check_data(self, fname, hid=None, wait=True):
|
||||
"""
|
||||
Verifies that a data at a history id is indentical to
|
||||
the contents of a file
|
||||
"""
|
||||
|
||||
if wait: # wait for tools to finish
|
||||
self.wait()
|
||||
|
||||
data_list = self.get_data_list()
|
||||
self.assertTrue( data_list )
|
||||
|
||||
if hid is None: # take last hid
|
||||
elem = data_list[-1]
|
||||
else:
|
||||
hid = str(hid)
|
||||
elems = [ elem for elem in data_list if elem.get('hid') == hid ]
|
||||
self.assertTrue( len(elems) == 1 )
|
||||
elem = elems[0]
|
||||
|
||||
if elem.get('state') != 'ok':
|
||||
tc.go("./history")
|
||||
tc.code(200)
|
||||
# print tc.show()
|
||||
|
||||
hid = elem.get('hid')
|
||||
self.assertTrue( hid )
|
||||
self._assert_dataset_state( elem, 'ok' )
|
||||
|
||||
local_name = self.get_fname(fname)
|
||||
temp_name = self.get_fname('temp_%s' % fname)
|
||||
tc.go("./display?hid=" + str(hid) )
|
||||
|
||||
data = self.last_page()
|
||||
file(temp_name, 'wb').write(data)
|
||||
|
||||
if not filecmp.cmp(local_name, temp_name):
|
||||
#maybe it is just the line endings
|
||||
lc = 0
|
||||
for line1, line2 in zip(file(local_name), file(temp_name)):
|
||||
lc += 1
|
||||
line1 = line1.strip()
|
||||
line2 = line2.strip()
|
||||
if line1 != line2:
|
||||
# nicer message
|
||||
fromlines = open( local_name, 'U').readlines()
|
||||
tolines = open( temp_name, 'U').readlines()
|
||||
diff = difflib.unified_diff( fromlines, tolines, "local_file", "history_data" )
|
||||
diff_slice = list( islice( diff, 40 ) )
|
||||
if len( diff_slice ) == 40:
|
||||
errmsg = [ 'Data at history id %s does not match expected, first 40 lines of diff:\n' % hid ]
|
||||
else:
|
||||
errmsg = [ 'Data at history id %s does not match expected, diff:\n' % hid ]
|
||||
errmsg += diff_slice
|
||||
raise AssertionError( "".join( errmsg ) )
|
||||
os.remove(temp_name)
|
||||
|
||||
def submit_form(self, form=1, button="runtool_btn", **kwd):
|
||||
"""Populates and submits a form from the keyword arguments"""
|
||||
for key, value in kwd.items():
|
||||
# needs to be able to handle multiple values per key
|
||||
if type(value) != type([]):
|
||||
value = [ value ]
|
||||
for elem in value:
|
||||
tc.fv(str(form), str(key), str(elem) )
|
||||
tc.submit(button)
|
||||
|
||||
def last_page(self):
|
||||
return tc.browser.get_html()
|
||||
|
||||
def run_tool(self, tool_id, **kwd):
|
||||
tool_id = tool_id.replace(" ", "+")
|
||||
"""Runs the tool 'tool_id' and pass it the key/values from the *kwd"""
|
||||
tc.go("%s/tool_runner/index?tool_id=%s" % (self.url, tool_id) )
|
||||
tc.code(200)
|
||||
tc.find('runtool_btn')
|
||||
self.submit_form(**kwd)
|
||||
tc.code(200)
|
||||
|
||||
def save_log(*path):
|
||||
"""Saves the log to a file"""
|
||||
fname = os.path.join( *path )
|
||||
file(fname, 'wt').write(buffer.getvalue())
|
||||
|
||||
def wait(self, maxiter=20):
|
||||
"""Waits for the tools to finish"""
|
||||
count = 0
|
||||
sleep_amount = 1
|
||||
self.home()
|
||||
while count < maxiter:
|
||||
count += 1
|
||||
tc.go("./history")
|
||||
page = tc.browser.get_html()
|
||||
if page.find('add_refresh( 9 );') > -1:
|
||||
time.sleep( sleep_amount )
|
||||
sleep_amount += 1
|
||||
else:
|
||||
break
|
||||
self.assertNotEqual(count, maxiter)
|
||||
|
||||
def check_history(self, patt):
|
||||
"""Checks history for a pattern"""
|
||||
tc.go("./history")
|
||||
for subpatt in patt.split():
|
||||
tc.find(subpatt)
|
||||
|
||||
def check_errors(self):
|
||||
"""Waits for the tools to finish"""
|
||||
tc.go("./history")
|
||||
page = self.last_page()
|
||||
if page.find('error') > -1:
|
||||
raise AssertionError('Errors in the history for user %s' % self.user )
|
||||
@@ -0,0 +1,111 @@
|
||||
import atexit, logging, os, sys, tempfile
|
||||
|
||||
import pkg_resources
|
||||
|
||||
pkg_resources.require( "twill==0.8.3" )
|
||||
pkg_resources.require( "Paste" )
|
||||
pkg_resources.require( "PasteDeploy" )
|
||||
pkg_resources.require( "Cheetah" )
|
||||
|
||||
import universe_wsgi, twill, unittest, time
|
||||
import os, os.path, subprocess, sys, threading
|
||||
import galaxy.web.server
|
||||
from galaxy.app import UniverseApplication
|
||||
|
||||
log = logging.getLogger( __name__ )
|
||||
|
||||
# TODO: should be able to override these, as well as prevent starting th
|
||||
# server (for running the tests against a running instance)
|
||||
|
||||
default_galaxy_test_host = "localhost"
|
||||
default_galaxy_test_port = "9999"
|
||||
|
||||
galaxy_test_file_dir = "test-data"
|
||||
|
||||
server = None
|
||||
app = None
|
||||
|
||||
def setup():
|
||||
"""Start the web server for the tests"""
|
||||
global server, app
|
||||
|
||||
galaxy_test_host = os.environ.get( 'GALAXY_TEST_HOST', default_galaxy_test_host )
|
||||
galaxy_test_port = os.environ.get( 'GALAXY_TEST_PORT', default_galaxy_test_port )
|
||||
|
||||
start_server = 'GALAXY_TEST_EXTERNAL' not in os.environ
|
||||
|
||||
if start_server:
|
||||
|
||||
tempdir = tempfile.mkdtemp()
|
||||
file_path = os.path.join( tempdir, 'database', 'files' )
|
||||
os.makedirs( file_path )
|
||||
if 'GALAXY_TEST_DBURI' in os.environ:
|
||||
database_connection = os.environ['GALAXY_TEST_DBURI']
|
||||
else:
|
||||
database_connection = 'sqlite:///' + os.path.join( tempdir, 'database', 'universe.sqlite' )
|
||||
|
||||
app = UniverseApplication( job_queue_workers = 5,
|
||||
template_path = "templates",
|
||||
database_connection = database_connection,
|
||||
file_path = file_path,
|
||||
tool_config_file = "tool_conf.xml",
|
||||
tool_path = "tools",
|
||||
test_conf = "test.conf",
|
||||
log_destination = "stdout",
|
||||
use_heartbeat=True )
|
||||
|
||||
log.info( "Embedded Universe application started" )
|
||||
|
||||
webapp = universe_wsgi.app_factory( dict(),
|
||||
use_translogger = False,
|
||||
app=app )
|
||||
|
||||
server = galaxy.web.server.serve( webapp, dict(),
|
||||
host=galaxy_test_host,
|
||||
port=galaxy_test_port,
|
||||
start_loop=False )
|
||||
|
||||
atexit.register( teardown )
|
||||
|
||||
import threading
|
||||
t = threading.Thread( target=server.serve_forever )
|
||||
t.start()
|
||||
|
||||
time.sleep( 2 )
|
||||
|
||||
log.info( "Embedded web server started" )
|
||||
|
||||
if app:
|
||||
# TODO: provisions for loading toolbox from file when using external server
|
||||
import test_toolbox
|
||||
test_toolbox.toolbox = app.toolbox
|
||||
else:
|
||||
from galaxy import tools
|
||||
import test_toolbox
|
||||
test_toolbox.toolbox = tools.ToolBox( 'tool_conf.xml', 'tools' )
|
||||
|
||||
# Test if the server is up
|
||||
import httplib
|
||||
conn = httplib.HTTPConnection( galaxy_test_host, galaxy_test_port )
|
||||
conn.request( "GET", "/" )
|
||||
assert conn.getresponse().status == 200, "Test HTTP server did not return '200 OK'"
|
||||
|
||||
os.environ['GALAXY_TEST_HOST'] = galaxy_test_host
|
||||
os.environ['GALAXY_TEST_PORT'] = galaxy_test_port
|
||||
os.environ['GALAXY_TEST_FILE_DIR'] = galaxy_test_file_dir
|
||||
|
||||
def teardown():
|
||||
global server, app
|
||||
if server:
|
||||
server.shutdown()
|
||||
server = None
|
||||
log.info( "Embedded web server stopped" )
|
||||
if app:
|
||||
app.shutdown()
|
||||
app = None
|
||||
log.info( "Embedded Universe application stopped" )
|
||||
time.sleep(2)
|
||||
|
||||
# def test_toolbox():
|
||||
# for test in app.toolbox.get_tests():
|
||||
# yield ( test, )
|
||||
@@ -0,0 +1,51 @@
|
||||
from galaxy.test.base.twilltestcase import TwillTestCase
|
||||
#from twilltestcase import TwillTestCase
|
||||
|
||||
class UcscTests(TwillTestCase):
|
||||
|
||||
def test_00_first(self): # will run first due to its name
|
||||
"""3A_GetData: Clearing history"""
|
||||
self.clear_history()
|
||||
|
||||
def test_10_upload(self):
|
||||
"""3A_GetData: Testing single upload"""
|
||||
self.clear_history()
|
||||
|
||||
self.upload_file('MyData.bed')
|
||||
self.check_data('MyData.bed', hid=1)
|
||||
|
||||
def test_20_UCSC_bed(self):
|
||||
"""3A_GetData: Getting bed files from UCSC"""
|
||||
self.clear_history()
|
||||
|
||||
self.run_tool('ucsc_proxy')
|
||||
params = dict(
|
||||
hgta_regionType="range",
|
||||
hgta_outputType="bed",
|
||||
)
|
||||
self.submit_form(1, button="hgta_doTopSubmit", **params)
|
||||
self.submit_form(1, button="hgta_doGetBed" )
|
||||
self.wait()
|
||||
self.check_data('3A_GetData_ucsc_bed_range.dat', hid=1)
|
||||
|
||||
def test_30_Biomart_Uniprot(self):
|
||||
"""3A_GetData: Connection to Biomart"""
|
||||
|
||||
# some button indices are hardcoded (twill limitation)
|
||||
self.run_tool('biomart')
|
||||
|
||||
self.submit_form(form='settings', button='stage_filter',
|
||||
database='UNIPROTPROTOTYPE4-5(EBI)__default', dataset='uniprot'
|
||||
)
|
||||
|
||||
self.submit_form(form='settings', button=6,
|
||||
uniprot_collection_start=1, uniprot_collection_end=1,
|
||||
uniprot_component_start=1000, uniprot_component_end=2000
|
||||
)
|
||||
|
||||
#self.submit_form(form='settings', button=5,
|
||||
# outtype='Features', outformat='tsv'
|
||||
#)
|
||||
|
||||
#self.check_data('biomart_uniprot.dat')
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
from galaxy.test.base.twilltestcase import TwillTestCase
|
||||
#from twilltestcase import TwillTestCase
|
||||
|
||||
class EncodeTests(TwillTestCase):
|
||||
|
||||
def test_00_first(self): # will run first due to its name
|
||||
"""3B_GetEncodeData: Clearing history"""
|
||||
self.clear_history()
|
||||
|
||||
def test_10_Encode_Data(self):
|
||||
"""3B_GetEncodeData: Getting encode data"""
|
||||
|
||||
self.run_tool('encode_import_chromatin_and_chromosomes1', hg17=['cc.EarlyRepSeg.20051216.bed'] )
|
||||
# hg17=[ "cc.EarlyRepSeg.20051216.bed", "cc.EarlyRepSeg.20051216.gencode_partitioned.bed", "cc.LateRepSeg.20051216.bed", "cc.LateRepSeg.20051216.gencode_partitioned.bed", "cc.MidRepSeg.20051216.bed", "cc.MidRepSeg.20051216.gencode_partitioned.bed" ] )
|
||||
|
||||
self.wait()
|
||||
self.check_data('cc.EarlyRepSeg.20051216.bed', hid=1)
|
||||
# self.check_data('cc.EarlyRepSeg.20051216.gencode_partitioned.bed', hid=2)
|
||||
# self.check_data('cc.LateRepSeg.20051216.bed', hid=3)
|
||||
# self.check_data('cc.LateRepSeg.20051216.gencode_partitioned.bed', hid=4)
|
||||
# self.check_data('cc.MidRepSeg.20051216.bed', hid=5)
|
||||
# self.check_data('cc.MidRepSeg.20051216.gencode_partitioned.bed', hid=6)
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
from galaxy.test.base.twilltestcase import TwillTestCase
|
||||
|
||||
class BasicFunctions(TwillTestCase):
|
||||
|
||||
def test_init(self):
|
||||
"""Testing home page"""
|
||||
self.home()
|
||||
|
||||
def test_single_upload(self):
|
||||
"""Testing single upload"""
|
||||
self.clear_history()
|
||||
self.upload_file('1.bed')
|
||||
|
||||
def test_multi_upload(self):
|
||||
"""Checking data for multiple uploads"""
|
||||
self.clear_history()
|
||||
self.upload_file('1.bed')
|
||||
self.check_data('1.bed')
|
||||
|
||||
self.upload_file('2.bed', dbkey='hg17')
|
||||
self.check_data('2.bed')
|
||||
|
||||
self.upload_file('3.bed', dbkey='hg17', ftype='bed')
|
||||
self.check_data('3.bed')
|
||||
self.clear_history()
|
||||
|
||||
def test_edit(self):
|
||||
"""Testing metadata editing"""
|
||||
self.clear_history()
|
||||
self.upload_file('1.bed', dbkey='hg15')
|
||||
self.check_history('hg15 bed')
|
||||
|
||||
self.edit_data(hid=1, dbkey='hg16', name='Testdata')
|
||||
self.check_history('hg16 Testdata bed')
|
||||
|
||||
# test switching types
|
||||
self.edit_data(hid=1, dbkey='hg16', chromCol='')
|
||||
self.check_history('hg16 Testdata tabular')
|
||||
|
||||
self.edit_data(hid=1, dbkey='hg15', chromCol='1', info='foo')
|
||||
self.check_history('hg15 Testdata interval foo')
|
||||
|
||||
def test_delete(self):
|
||||
"""Testing individual deletes"""
|
||||
self.clear_history()
|
||||
self.upload_file('1.bed', dbkey='hg15')
|
||||
self.check_history('hg15 bed')
|
||||
self.assertEqual ( len(self.get_data_list()), 1)
|
||||
self.delete_data(1)
|
||||
self.check_history("Your history is empty")
|
||||
self.assertEqual ( len(self.get_data_list()), 0)
|
||||
@@ -0,0 +1,56 @@
|
||||
from galaxy.test.base.twilltestcase import TwillTestCase
|
||||
|
||||
class UcscTests(TwillTestCase):
|
||||
|
||||
def test_00_first(self): # will run first due to its name
|
||||
"""Clearing history"""
|
||||
self.clear_history()
|
||||
|
||||
def test_Biomart_Uniprot(self):
|
||||
"""Connection to Biomart"""
|
||||
|
||||
# some button indices are hardcoded (twill limitation)
|
||||
self.run_tool('biomart')
|
||||
|
||||
self.submit_form(form='settings', button='stage_filter',
|
||||
database='UNIPROTPROTOTYPE4-5(EBI)__default', dataset='uniprot'
|
||||
)
|
||||
|
||||
self.submit_form(form='settings', button=6,
|
||||
uniprot_collection_start=1, uniprot_collection_end=1,
|
||||
uniprot_component_start=1000, uniprot_component_end=2000
|
||||
)
|
||||
|
||||
#self.submit_form(form='settings', button=5,
|
||||
# outtype='Features', outformat='tsv'
|
||||
#)
|
||||
|
||||
#self.check_data('biomart_uniprot.dat')
|
||||
|
||||
|
||||
def test_UCSC_bed(self):
|
||||
"""Getting bed files from UCSC"""
|
||||
|
||||
self.clear_history()
|
||||
|
||||
# in range
|
||||
self.run_tool('ucsc_proxy')
|
||||
params = dict(
|
||||
hgta_regionType="range",
|
||||
hgta_outputType="primaryTable",
|
||||
)
|
||||
self.submit_form(1, button="hgta_doTopSubmit", **params)
|
||||
|
||||
# encode
|
||||
self.run_tool('ucsc_proxy')
|
||||
params = dict(
|
||||
hgta_regionType="encode",
|
||||
hgta_outputType="primaryTable",
|
||||
)
|
||||
self.submit_form(1, button="hgta_doTopSubmit", **params)
|
||||
|
||||
self.wait()
|
||||
|
||||
self.check_data('ucsc_proxy_range.dat', hid=1)
|
||||
self.check_data('ucsc_proxy_encode.dat', hid=2)
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
from galaxy.test.base.twilltestcase import TwillTestCase
|
||||
import sys, filecmp, string, os.path
|
||||
from twill.commands import *
|
||||
|
||||
#
|
||||
# the numbering of the tests is essential as they
|
||||
# will be executed in order, sorted by name
|
||||
#
|
||||
|
||||
class ToolTests(TwillTestCase):
|
||||
|
||||
def test_00_upload(self): # this must run first
|
||||
"Clear history"
|
||||
self.clear_history()
|
||||
|
||||
|
||||
def test_history(self):
|
||||
"""Test history """
|
||||
self.clear_history()
|
||||
|
||||
# in range
|
||||
self.run_tool('ucsc_proxy')
|
||||
params = dict(
|
||||
hgta_regionType="range",
|
||||
hgta_outputType="primaryTable",
|
||||
)
|
||||
self.submit_form(1, button="hgta_doTopSubmit", **params)
|
||||
|
||||
self.wait()
|
||||
self.check_data('ucsc_proxy_range.dat', hid=1)
|
||||
|
||||
self.clear_history()
|
||||
if len(self.historyid()) > 0:
|
||||
raise AssertionError("Failure - clear history.")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
from galaxy.test.base.twilltestcase import TwillTestCase
|
||||
|
||||
# the numbering of the tests is essential as they
|
||||
# will be executed in order, sorted by name
|
||||
class Operations(TwillTestCase):
|
||||
|
||||
def test_00_first(self): # must be first
|
||||
"Uploading data"
|
||||
self.clear_history()
|
||||
self.upload_file('1.bed', dbkey='hg17')
|
||||
self.upload_file('2.bed', dbkey='hg17')
|
||||
|
||||
def test_intersect(self):
|
||||
"Intersect"
|
||||
self.run_tool('Intersect1', input1='1.bed', input2="2.bed", min=1)
|
||||
self.check_data('op-intersect.dat')
|
||||
|
||||
def test_join(self):
|
||||
"Join"
|
||||
self.run_tool('Join1', input1='1.bed', input2="2.bed", min=1)
|
||||
self.check_data('op-join.dat')
|
||||
|
||||
def test_merge(self):
|
||||
"Merge"
|
||||
self.run_tool('Merge1', input='1.bed')
|
||||
self.check_data('op-merge.dat')
|
||||
|
||||
def test_overlap(self):
|
||||
"Overlap"
|
||||
self.run_tool('Overlap1', input1='1.bed', input2="2.bed", min=1)
|
||||
self.check_data('op-overlap.dat')
|
||||
|
||||
def test_proximity(self):
|
||||
"Proximity"
|
||||
self.run_tool('Proximity1', input1='1.bed', input2="2.bed", within="-within", up=0, down=0)
|
||||
self.check_data('op-proximity.dat')
|
||||
|
||||
def test_subtract(self):
|
||||
"Subtract"
|
||||
self.run_tool('Subtract1', input1='1.bed', input2="2.bed")
|
||||
self.check_data('op-subtract.dat')
|
||||
|
||||
def test_union(self):
|
||||
"Union"
|
||||
self.run_tool('Union2', input1='1.bed', input2="2.bed")
|
||||
self.check_data('op-union.dat')
|
||||
|
||||
def test_cluster(self):
|
||||
"Cluster"
|
||||
self.run_tool('Cluster1', input='1.bed', num=2, size=10)
|
||||
self.check_data('op-cluster.dat')
|
||||
|
||||
def test_clustermerge(self):
|
||||
"Cluster merge"
|
||||
self.run_tool('ClusterMerge1', input='1.bed', num=2, size=10)
|
||||
self.check_data( 'op-clustermerge.dat')
|
||||
|
||||
def test_complement(self):
|
||||
"Complement"
|
||||
self.run_tool('Complement1', input='1.bed')
|
||||
self.check_data('op-complement.dat')
|
||||
|
||||
def test_interval_complement(self):
|
||||
"Interval Complement"
|
||||
self.run_tool('interval_complement1', input='1.bed')
|
||||
self.check_data('op-interval-complement.dat')
|
||||
|
||||
def test_covdens(self):
|
||||
"covDensity"
|
||||
self.run_tool('covDensity1', input1='1.bed', input2="2.bed")
|
||||
self.check_data('op-covdensity.dat')
|
||||
|
||||
def test_interval_covdens(self):
|
||||
"Interval covDensity"
|
||||
self.run_tool('interval_coverage', input1='1.bed')
|
||||
self.check_data('op-interval-covdensity.dat')
|
||||
|
||||
def test_difference(self):
|
||||
"Difference"
|
||||
self.run_tool('Difference1', input1='1.bed', input2="2.bed")
|
||||
self.check_data('op-difference.dat')
|
||||
|
||||
def test_extend(self):
|
||||
"Extend"
|
||||
self.run_tool('Extend1', input='1.bed', len=100, where="ud")
|
||||
self.check_data('op-extend.dat')
|
||||
|
||||
def test_zz_fill_the_queue(self): # must be last (when sorted on the name)
|
||||
"Stress test, running all operations"
|
||||
|
||||
self.test_00_first()
|
||||
|
||||
self.run_tool('Intersect1', input1='1.bed', input2="2.bed", min=1)
|
||||
self.run_tool('Join1', input1='1.bed', input2="2.bed", min=1)
|
||||
self.run_tool('Merge1', input='1.bed')
|
||||
self.run_tool('Overlap1', input1='1.bed', input2="2.bed", min=1)
|
||||
|
||||
self.run_tool('Proximity1', input1='1.bed', input2="2.bed", within="-within", up=0, down=0)
|
||||
self.run_tool('Subtract1', input1='1.bed', input2="2.bed")
|
||||
self.run_tool('Union2', input1='1.bed', input2="2.bed")
|
||||
self.run_tool('Cluster1', input='1.bed', num=2, size=10)
|
||||
|
||||
self.run_tool('ClusterMerge1', input='1.bed', num=2, size=10)
|
||||
self.run_tool('Complement1', input='1.bed')
|
||||
self.run_tool('covDensity1', input1='1.bed', input2="2.bed")
|
||||
self.run_tool('Difference1', input1='1.bed', input2="2.bed")
|
||||
self.run_tool('interval_complement1', input='1.bed')
|
||||
self.run_tool('interval_coverage', input1='1.bed')
|
||||
self.run_tool('Extend1', input='1.bed', len=100, where="ud")
|
||||
|
||||
self.wait()
|
||||
names = [
|
||||
'op-intersect.dat','op-join.dat', 'op-merge.dat', 'op-overlap.dat',
|
||||
'op-proximity.dat', 'op-subtract.dat', 'op-union.dat', 'op-cluster.dat',
|
||||
'op-clustermerge.dat', 'op-complement.dat', 'op-covdensity.dat', 'op-difference.dat',
|
||||
'op-interval-complement.dat', 'op-interval-covdensity.dat', 'op-extend.dat'
|
||||
]
|
||||
for idx, name in enumerate(names):
|
||||
self.check_data(name, hid=idx+3, wait=False)
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
import sys
|
||||
import new
|
||||
from galaxy.test.base.twilltestcase import TwillTestCase
|
||||
|
||||
toolbox = None
|
||||
|
||||
class ToolTestCase( TwillTestCase ):
|
||||
"""
|
||||
Abstract test case that runs tests based on a `galaxy.tools.test.ToolTest`
|
||||
"""
|
||||
def do_it( self ):
|
||||
# Start with an empty history
|
||||
self.clear_history()
|
||||
# Upload any needed files
|
||||
for fname, extra in self.testdef.required_files:
|
||||
self.upload_file( fname, ftype=extra.get( 'ftype', 'auto' ), dbkey=extra.get( 'dbkey', 'hg17' ) )
|
||||
# Run the tool
|
||||
all_inputs = dict( ( name, value ) for ( name, value, _ ) in self.testdef.inputs )
|
||||
# Do the first page
|
||||
page_inputs = dict( ( key, all_inputs[key] )
|
||||
for key in self.testdef.tool.param_map_by_page[0].keys() )
|
||||
self.run_tool( self.testdef.tool.id, **page_inputs )
|
||||
# Do other pages if they exist
|
||||
for i in range( 1, self.testdef.tool.npages ):
|
||||
page_inputs = dict( ( key, all_inputs[key] )
|
||||
for key in self.testdef.tool.param_map_by_page[i].keys() )
|
||||
self.submit_form( **page_inputs )
|
||||
# Check the result
|
||||
assert len( self.testdef.outputs ) == 1, "ToolTestCase does not deal with multiple outputs properly yet."
|
||||
for name, file in self.testdef.outputs:
|
||||
self.check_data( file )
|
||||
def shortDescription( self ):
|
||||
return self.name
|
||||
|
||||
def get_testcase( testdef, name ):
|
||||
"""
|
||||
Dynamically generate a `ToolTestCase` for `testdef`
|
||||
"""
|
||||
n = "GeneratedToolTestCase_" + testdef.tool.id.replace( ' ', '_' )
|
||||
s = ( ToolTestCase, )
|
||||
def test_tool( self ):
|
||||
self.do_it()
|
||||
d = dict( testdef=testdef, test_tool=test_tool, name=name )
|
||||
return new.classobj( n, s, d )
|
||||
|
||||
def setup():
|
||||
"""
|
||||
If the module level variable `toolbox` is set, generate `ToolTestCase`
|
||||
classes for all of its tests and put them into this modules globals() so
|
||||
they can be discovered by nose.
|
||||
"""
|
||||
if toolbox is None:
|
||||
return
|
||||
# Push all the toolbox tests to module level
|
||||
G = globals()
|
||||
for i, section in enumerate( toolbox.sections ):
|
||||
for j, tool in enumerate( section.tools ):
|
||||
if tool.tests:
|
||||
for k, testdef in enumerate( tool.tests ):
|
||||
name = "%s > %s > %s" % ( section.name, tool.name, testdef.name )
|
||||
testcase = get_testcase( testdef, name )
|
||||
G[ 'testcase_%d_%d_%d' % ( i, j, k ) ] = testcase
|
||||
Reference in New Issue
Block a user