from phaser.pipeline import proxies
import phaser
import iotbx.pdb
import libtbx.load_env
import unittest
import tempfile
import urllib2
import StringIO
import gzip
import time
import xml.etree.ElementTree
import os
class TestDownloaderBase(unittest.TestCase):
def setUp(self):
self.downloader = proxies.DownloaderBase()
self.downloader.ENCODING = "encoding"
self.data = "Some random data"
self.data_stream = StringIO.StringIO( self.data )
self.data_stream.headers = {}
def testHandles(self):
self.data_stream.headers[ proxies.SimpleDownloader.STREAM_HEADER ] = ""
self.assert_( not self.downloader.handles( self.data_stream ) )
self.data_stream.headers[ proxies.SimpleDownloader.STREAM_HEADER ] = "gzip"
self.assert_( not self.downloader.handles( self.data_stream ) )
self.data_stream.headers[ proxies.SimpleDownloader.STREAM_HEADER ] = self.downloader.ENCODING
self.assert_( self.downloader.handles( self.data_stream ) )
def testInterpretableEncoding(self):
self.assert_( not self.downloader.interpretable_encoding( None ) )
self.assert_( not self.downloader.interpretable_encoding( "" ) )
self.assert_( not self.downloader.interpretable_encoding( "unknown" ) )
self.assert_( self.downloader.interpretable_encoding( self.downloader.ENCODING ) )
def testEnableEcoding(self):
request = urllib2.Request( "http://foo/bar" )
self.downloader.enable_encoding( request )
self.assert_(
self.downloader.ENCODING in request.get_header(
self.downloader.REQUEST_HEADER
)
)
class TestSimpleDownloader(unittest.TestCase):
def setUp(self):
self.data = "Some random data"
self.data_stream = StringIO.StringIO( self.data )
self.data_stream.headers = {}
self.downloader = proxies.SimpleDownloader()
def testInterpretableEncoding(self):
self.assert_( self.downloader.interpretable_encoding( "" ) )
self.assert_( self.downloader.interpretable_encoding( "identity" ) )
self.assert_( not self.downloader.interpretable_encoding( "gzip" ) )
def testFetch(self):
self.assertEqual( self.downloader.fetch( self.data_stream ), self.data )
class TestGzipDownloader(unittest.TestCase):
def setUp(self):
self.data = "Some random data"
compressed_data = StringIO.StringIO()
compressor_stream = gzip.GzipFile( fileobj = compressed_data, mode = "wb" )
compressor_stream.write( self.data )
compressor_stream.close()
self.data_stream = StringIO.StringIO( compressed_data.getvalue() )
self.data_stream.headers = {}
self.downloader = proxies.GzipDownloader()
compressed_data.close()
def testFetch(self):
self.assertEqual( self.downloader.fetch( self.data_stream ), self.data )
self.assertRaises(
proxies.ProxyException,
self.downloader.fetch,
StringIO.StringIO( self.data )
)
class TestHTTPInteface(unittest.TestCase):
def testResourceOnline(self):
http_interface = proxies.HTTPInterface()
self.assert_( http_interface.resource_online(
url = "http://dolphin/nightly_builds/index"
) )
self.assert_( not http_interface.resource_online(
url = "http://foo/bar"
) )
self.assert_( not http_interface.resource_online(
url = "http://dolphin/bar"
) )
def testHTTPCall(self):
http_interface = proxies.HTTPInterface()
request = {
"url": "http://dolphin/nightly_builds/build_log",
"data": {
"platform": "perch-default",
"id": "build-2009-03-27",
}
}
data = http_interface.http_call( request )
self.assert_( 'Python: 2.5.1 "/usr/bin/python"' in data )
self.assert_( "scons: done building targets." in data )
request[ "url" ] = "http://dolphin/foo"
self.assertRaises(
proxies.ProxyException,
http_interface.http_call,
request
)
class TestOCAHTTP(unittest.TestCase):
SEQUENCE_3ADK = """Sequence for 3adk
>3adk_ mol:protein-het length:195 Adenylate Kinase (E.C. 2.7.4.3) - Chain
XMEEKLKKSKIIFVVGGPGSGKGTQCEKIVQKYGYTHLSTGDLLRAEVSSGSARGKMLSEIMEKGQLVPLETVLD
MLRDAMVAKVDTSKGFLIDGYPREVKQGEEFERKIGQPTLLLYVDAGPETMTKRLLKRGETSGRVDDNEETIKKR
LETYYKATEPVIAFYEKRGIVRKVNAEGSVDDVFSQVCTHLDTLK
"""
oca = None
def setUp(self):
if not self.oca:
self.oca = proxies.OCAHTTP()
def testName(self):
self.assertEqual( str( self.oca ), "OCA service at EBI" )
def testRetrieve(self):
self.assertEqual(
self.oca.fetch( identifier = "3ADK" ),
self.SEQUENCE_3ADK
)
self.assertEqual(
self.oca.fetch( identifier = "3adK" ),
self.SEQUENCE_3ADK
)
self.assertEqual(
self.oca.fetch( identifier = "4BZQ" ),
"Sequence for 4bzq\n"
)
class TestDbfetchHTTP(unittest.TestCase):
dbfetch = None
def setUp(self):
if not self.dbfetch:
self.dbfetch = proxies.DbfetchHTTP()
def testName(self):
self.assertEqual( str( self.dbfetch ), "Dbfetch service at EBI" )
def testRetrieve(self):
data = self.dbfetch.fetch( identifier = "3ADK" )
self.checkData( data )
data = self.dbfetch.fetch( identifier = "3adK" )
self.checkData( data )
self.assertEqual(
self.dbfetch.fetch( identifier = "4BZQ" ),
"No entries found. \n"
)
def checkData(self, data):
hierarchy = iotbx.pdb.input( lines = data, source_info = "PDB" ).construct_hierarchy()
self.assertEqual( len( hierarchy.models() ), 1 )
self.assertEqual( len( hierarchy.models()[0].chains() ), 2 )
self.assertEqual(
len( hierarchy.models()[0].chains()[0].residue_groups() ),
195
)
self.assertEqual(
len( hierarchy.models()[0].chains()[1].residue_groups() ),
2
)
self.assertEqual( len( hierarchy.atoms() ), 1529 )
class TestRCSBHTTP(unittest.TestCase):
SEQUENCE_3ADK = """>3ADK:A|PDBID|CHAIN|SEQUENCE
XMEEKLKKSKIIFVVGGPGSGKGTQCEKIVQKYGYTHLSTGDLLRAEVSSGSARGKMLSEIMEKGQLVPLETVLDMLRDA
MVAKVDTSKGFLIDGYPREVKQGEEFERKIGQPTLLLYVDAGPETMTKRLLKRGETSGRVDDNEETIKKRLETYYKATEP
VIAFYEKRGIVRKVNAEGSVDDVFSQVCTHLDTLK
"""
pdb = None
def setUp(self):
if not self.pdb:
self.pdb = proxies.RCSBHTTP()
def testName(self):
self.assertEqual( str( self.pdb ), "PDB service at RCSB" )
def testRetrieve(self):
data = self.pdb.fetch( identifier = "3ADK", compression = False )
self.checkData( data )
data = self.pdb.fetch( identifier = "3adK", compression = True )
uncompressed = gzip.GzipFile( fileobj = StringIO.StringIO( data ) ).read()
self.checkData( uncompressed )
self.assertEqual(
self.pdb.fetch( identifier = "3ADK", format = "fasta" ),
self.SEQUENCE_3ADK
)
self.assertEqual(
self.pdb.fetch( identifier = "3adK", format = "FASTA" ),
self.SEQUENCE_3ADK
)
self.assertRaises(
proxies.ProxyException,
self.pdb.fetch,
"4BZQ"
)
def checkData(self, data):
hierarchy = iotbx.pdb.input( lines = data, source_info = "PDB" ).construct_hierarchy()
self.assertEqual( len( hierarchy.models() ), 1 )
self.assertEqual( len( hierarchy.models()[0].chains() ), 2 )
self.assertEqual(
len( hierarchy.models()[0].chains()[0].residue_groups() ),
195
)
self.assertEqual(
len( hierarchy.models()[0].chains()[1].residue_groups() ),
2
)
self.assertEqual( len( hierarchy.atoms() ), 1529 )
class TestNCBIBlastHTTP(unittest.TestCase):
ncbi = None
rid = None
SEQUENCE_3ADK = """>3ADK:A|PDBID|CHAIN|SEQUENCE
XMEEKLKKSKIIFVVGGPGSGKGTQCEKIVQKYGYTHLSTGDLLRAEVSSGSARGKMLSEIMEKGQLVPLETVLDMLRDA
MVAKVDTSKGFLIDGYPREVKQGEEFERKIGQPTLLLYVDAGPETMTKRLLKRGETSGRVDDNEETIKKRLETYYKATEP
VIAFYEKRGIVRKVNAEGSVDDVFSQVCTHLDTLK
"""
def setUp(self):
if not self.ncbi:
self.ncbi = proxies.NCBIBlastHTTP()
if not self.rid:
self.rid = self.ncbi.put(
program = "blastp",
database = "pdb",
query = self.SEQUENCE_3ADK
)
while self.ncbi.status( rid = self.rid ) != "READY":
time.sleep( 3 )
def testName(self):
self.assertEqual( str( self.ncbi ), "BLAST service at NCBI" )
def testPut(self):
self.assertRaises(
proxies.ProxyException,
self.ncbi.put,
program = "blastf",
database = "pdb",
query = self.SEQUENCE_3ADK
)
def testGet(self):
data = self.ncbi.get(
rid = self.rid,
format_type = "XML"
)
root = xml.etree.ElementTree.fromstring( data )
program = root.find( "BlastOutput_program" )
self.assert_( program is not None )
self.assertEqual( program.text, "blastp" )
db = root.find( "BlastOutput_db" )
self.assert_( db is not None )
self.assertEqual( db.text, "pdb" )
def testStatus(self):
self.assertEqual( self.ncbi.status( rid = self.rid ), "READY" )
class TestNCBIBlastStandalone(unittest.TestCase):
blast = None
jobid = None
SEQUENCE_3ADK = """>3ADK:A|PDBID|CHAIN|SEQUENCE
XMEEKLKKSKIIFVVGGPGSGKGTQCEKIVQKYGYTHLSTGDLLRAEVSSGSARGKMLSEIMEKGQLVPLETVLDMLRDA
MVAKVDTSKGFLIDGYPREVKQGEEFERKIGQPTLLLYVDAGPETMTKRLLKRGETSGRVDDNEETIKKRLETYYKATEP
VIAFYEKRGIVRKVNAEGSVDDVFSQVCTHLDTLK
"""
def setUp(self):
if not self.blast:
self.blast = proxies.NCBIBlastStandalone(
command_line = (
"/usr/bin/ssh", "ulysses", "/srs/blast/blast-2.2.20/bin/blastall"
)
)
if not self.jobid:
self.jobid = self.blast.submit(
program = "blastp",
database = "/srs/srs/data/blast/pdbaa",
query = self.SEQUENCE_3ADK
)
def testName(self):
self.assertEqual( str( self.blast ), "BLAST standalone executable" )
def testGet(self):
data = self.blast.results( jobid = self.jobid )
root = xml.etree.ElementTree.fromstring( data )
program = root.find( "BlastOutput_program" )
self.assert_( program is not None )
self.assertEqual( program.text, "blastp" )
db = root.find( "BlastOutput_db" )
self.assert_( db is not None )
self.assertEqual( db.text, "/srs/srs/data/blast/pdbaa" )
def testStatus(self):
self.assertEqual( self.blast.status( jobid = self.jobid ), "FINISHED" )
class TestClustalWLocal(unittest.TestCase):
def setUp(self):
self.job_data_dir = tempfile.mkdtemp()
self.clustalw2 = proxies.ClustalWLocal(
job_data_dir = self.job_data_dir,
executable = "/xtal/crystal/ccp4/ccp4-6.1.1/clustalw-2.0.9/bin/clustalw2"
)
def tearDown(self):
for jobid in self.clustalw2.existing_jobs():
self.clustalw2.cleanup( jobid )
if os.path.exists( self.job_data_dir ):
os.rmdir( self.job_data_dir )
def testCreate(self):
self.assertRaises(
proxies.ProxyException,
proxies.ClustalWLocal,
self.job_data_dir,
"clustalx"
)
non_existent_dir = tempfile.mkdtemp()
os.rmdir( non_existent_dir )
self.assertRaises(
proxies.ProxyException,
proxies.ClustalWLocal,
non_existent_dir
)
def testRun(self):
sequences = """>1vkk
VVCEVDPELKETLRKFRFRKETNNAAIIMKVDKDRQMVVLEDELQNISPEELKLELPERQPRFVVYSYKYV
HDDGRVSYPLCFIFSSPVGCKPEQQMMYAGSKNRLVQTAELTKVFEIRTTDDLTETWLKEKLAFFR
>1ahq
GIAVSDDCVQKFNELKLGHQHRYVTFKMNASNTEVVVEHVGGPNATYEDFKSQLPERDCRYAIFDYEFQVD
GGQRNKITFILWAPDSAPIKSKMMYTSTKDSIKKKLVGIQVEVQATDAAEISEDAVSERAKK
"""
aln = ( "CLUSTAL 2.0.9 multiple sequence alignment\n\n\n1vkk "
+ " VVCEVDPELKETLRKFRFRKETNNAAIIMKVDKDRQMVVLEDELQNISPEELKLELPERQ\n1"
+ "ahq -GIAVSDDCVQKFNELKLGHQHRYVTFKMNAS-NTEVVVEHVGGPNATY"
+ "EDFKSQLPERD\n *. : :.:.:::: :: . .:: *:.. : "
+ "::** . * : *::* :****:\n\n1vkk PRFVVYSYKYVHDDGRVSY"
+ "PLCFIFSSPVGCKPEQQMMYAGSKNRLVQT-AELTKVFEIR\n1ahq CRYAI"
+ "FDYEFQVDGGQRNK-ITFILWAPDSAPIKSKMMYTSTKDSIKKKLVGIQVEVQAT\n "
+ " *:.::.*:: *.*: . : **: :* .. :.:***:.:*: : :. . : "
+ " .: \n\n1vkk TTDDLTETWLKEKLAFFR\n1ahq DAA"
+ "EISEDAVSERAKK--\n : :::* :.*: \n" )
title = "CLUSTAL 2.0.9 Multiple Sequence Alignments"
jobid = self.clustalw2.run( sequences = sequences )
self.assert_(
self.clustalw2.status( jobid = jobid ) in [ "RUNNING", "FINISHED" ]
)
while self.clustalw2.status( jobid = jobid ) == "RUNNING":
time.sleep( 1 )
self.assertEqual( self.clustalw2.status( jobid = jobid ), "FINISHED" )
self.assertEqual( self.clustalw2.fetch_aln( jobid = jobid ), aln )
self.assert_(
self.clustalw2.fetch_log( jobid = jobid ).find( title ) != -1
)
self.clustalw2.cleanup( jobid = jobid )
self.assertEqual( self.clustalw2.status( jobid = jobid ), "UNKNOWN" )
self.assertEqual( self.clustalw2.fetch_aln( jobid = jobid ), "" )
self.assertEqual( self.clustalw2.fetch_log( jobid = jobid ), "" )
class TestPhaserMultiprocessing(unittest.TestCase):
phaser = None
def setUp(self):
self.model1_file = libtbx.env.under_dist(
"phaser",
os.path.join( "tutorial", "1D0D_B.pdb" )
)
self.model2_file = libtbx.env.under_dist(
"phaser",
os.path.join( "tutorial", "1BIK.pdb" )
)
self.hkl_file = libtbx.env.under_dist(
"phaser",
os.path.join( "tutorial", "toxd.mtz" )
)
if not self.phaser:
self.create_proxy()
def tearDown(self):
if self.phaser:
for job_name in self.phaser.existing_jobs():
self.phaser.cleanup( job_name = job_name )
def create_proxy(self):
self.phaser = proxies.PhaserMultiprocessing()
def load_reflection_data(self):
job_name = self.phaser.submit_mr_dat_job(
hkl_file = self.hkl_file,
fp_label = "FTOXD3",
sigfp_label = "SIGFTOXD3"
)
self.wait_for_jobs_to_finish( jobs = [ job_name ] )
result = self.phaser.results( job_name = job_name )
self.assert_( len( result ), 6 )
( success, hkl, fp, sigfp, space_group, unit_cell ) = result
self.assert_( success )
self.assertEqual( len( hkl ), 3161 )
self.assertEqual( len( fp ), 3161 )
self.assertEqual( len( sigfp ), 3161 )
self.assertEqual( space_group, "P 21 21 21" )
for ( obs, exp ) in zip( unit_cell, [ 73.58, 38.73, 23.19, 90.00, 90.00, 90.00 ] ):
self.assertAlmostEqual( obs, exp, 2 )
self.hkl = hkl
self.fp = fp
self.sigfp = sigfp
self.space_group = space_group
self.unit_cell = unit_cell
def start_mr_auto_job(self, jobs = 1):
return self.phaser.submit_mr_auto_job(
hkl = self.hkl,
fp = self.fp,
sigfp = self.sigfp,
space_group = self.space_group,
unit_cell = self.unit_cell,
asu_weight = 7139,
models_for = {
"toxd": [ ( self.model1_file, 0.364 ), ( self.model2_file, 0.377 ) ]
},
search_ensembles = [ ( "toxd", 1 ) ],
high_resolution = 4.0,
jobs = jobs
)
def compare_job_results_with_expected(self, job_name):
self.assertEqual(
self.phaser.status( job_name = job_name ),
self.phaser.STATUS_FINISHED
)
result = self.phaser.results( job_name = job_name )
self.assertEqual( len( result ), 2 )
self.assert_( result[0] )
self.assert_( isinstance( result[1], phaser.mr_solution ) )
self.assertEqual(
result[1][0].ANNOTATION,
" RFZ=3.7 TFZ=7.4 PAK=0 LLG=63 LLG=113"
)
def cleanup_changes_status(self, job_name):
self.assertEqual(
self.phaser.status( job_name = job_name ),
self.phaser.STATUS_FINISHED
)
self.phaser.cleanup( job_name = job_name )
self.assertEqual(
self.phaser.status( job_name = job_name ),
self.phaser.STATUS_UNKNOWN
)
def wait_for_jobs_to_finish(self, jobs):
for job_name in jobs:
self.assert_(
self.phaser.status( job_name = job_name ) in [
self.phaser.STATUS_RUNNING,
self.phaser.STATUS_FINISHED,
self.phaser.STATUS_UNKNOWN,
]
)
while any(
[ self.phaser.status( job_name = j ) == self.phaser.STATUS_RUNNING
for j in jobs ]
):
time.sleep( 1 )
def testRun(self):
self.load_reflection_data()
job_name = self.start_mr_auto_job()
self.wait_for_jobs_to_finish( jobs = [ job_name ] )
self.compare_job_results_with_expected( job_name = job_name )
self.cleanup_changes_status( job_name = job_name )
# Run 3 jobs in parallel
job1 = self.start_mr_auto_job()
job2 = self.start_mr_auto_job()
job3 = self.start_mr_auto_job()
self.wait_for_jobs_to_finish( jobs = [ job1, job2, job3 ] )
for job_name in [ job1, job2, job3 ]:
self.compare_job_results_with_expected( job_name = job_name )
self.cleanup_changes_status( job_name = job_name )
class TestPhaserOpenMP(TestPhaserMultiprocessing):
def create_proxy(self):
self.phaser = proxies.PhaserOpenMP()
def testRun(self):
self.load_reflection_data()
job_name = self.start_mr_auto_job()
self.wait_for_jobs_to_finish( jobs = [ job_name ] )
self.compare_job_results_with_expected( job_name = job_name )
self.cleanup_changes_status( job_name = job_name )
# Run on 3 CPUs
job_more_cpu = self.start_mr_auto_job( jobs = 3 )
self.wait_for_jobs_to_finish( jobs = [ job_more_cpu ] )
self.compare_job_results_with_expected( job_name = job_more_cpu )
self.cleanup_changes_status( job_name = job_more_cpu )
suite_downloader_base = unittest.TestLoader().loadTestsFromTestCase(
TestDownloaderBase
)
suite_simple_downloader = unittest.TestLoader().loadTestsFromTestCase(
TestSimpleDownloader
)
suite_gzip_downloader = unittest.TestLoader().loadTestsFromTestCase(
TestGzipDownloader
)
suite_http_inteface = unittest.TestLoader().loadTestsFromTestCase(
TestHTTPInteface
)
suite_oca_http = unittest.TestLoader().loadTestsFromTestCase(
TestOCAHTTP
)
suite_dbfetch_http = unittest.TestLoader().loadTestsFromTestCase(
TestDbfetchHTTP
)
suite_rcsb_http = unittest.TestLoader().loadTestsFromTestCase(
TestRCSBHTTP
)
suite_ncbi_blast_http = unittest.TestLoader().loadTestsFromTestCase(
TestNCBIBlastHTTP
)
suite_ncbi_blast_standalone = unittest.TestLoader().loadTestsFromTestCase(
TestNCBIBlastStandalone
)
suite_clustalw_local = unittest.TestLoader().loadTestsFromTestCase(
TestClustalWLocal
)
suite_phaser_multiprocessing = unittest.TestLoader().loadTestsFromTestCase(
TestPhaserMultiprocessing
)
suite_phaser_openmp = unittest.TestLoader().loadTestsFromTestCase(
TestPhaserOpenMP
)
alltests = unittest.TestSuite(
[
suite_downloader_base,
suite_simple_downloader,
suite_gzip_downloader,
#suite_http_inteface,
#suite_oca_http,
#suite_dbfetch_http,
#suite_rcsb_http,
#suite_ncbi_blast_http,
#suite_ncbi_blast_standalone,
#suite_clustalw_local,
#suite_phaser_multiprocessing,
#suite_phaser_openmp,
]
)
if __name__ == "__main__":
unittest.TextTestRunner( verbosity = 2 ).run( alltests )