BLAST (blastp) functionnal test

>Unknow sequence for GBIO Web Services test ( C Blanchet, 14 jan 2009)
MKKITIYDLAELSGVSASAVSAILNGNWKKRRISAKLAEKVTRIAEEQGYAINRQASMLR
SKKSHVIGMIIPKYDNRYFGSIAERFEEMARERGLLPIITCTRRRPELEIEAVKAMLSWQ
VDWVVATGATNPDKISALCQQAGVPTVNLDLPGSLSPSVISDNYGGAKALTHKILANSAR
RRGELAPLTFIGGRRATITPASVYAASTMRIASWGLACRRRIFWLPAIRKATLRTACRSG
LAARRRCCRGYLLTRRYPWKGLCAGCRRWV

##################################################
# file: Blast_Service_client.py
# 
# client stubs generated by "ZSI.generate.wsdl2python.WriteServiceModule"
#     /usr/local/bin/wsdl2py http://gbio-pbil.ibcp.fr/ws/BlastWS.wsdl
# 
##################################################


import urlparse, types
from ZSI.TCcompound import ComplexType, Struct
from ZSI import client
from ZSI.schema import GED, GTD
import ZSI

# Locator
class Blast_ServiceLocator:
    BlastPort_address = "http://gbio.ibcp.fr:8090/Blast_Service"
    def getBlastPortAddress(self):
        return Blast_ServiceLocator.BlastPort_address
    def getBlastPort(self, url=None, **kw):
        return BlastBindingSOAP(url or Blast_ServiceLocator.BlastPort_address, **kw)

# Methods
class BlastBindingSOAP:
    def __init__(self, url, **kw):
        kw.setdefault("readerclass", None)
        kw.setdefault("writerclass", None)
        # no resource properties
        self.binding = client.Binding(url=url, **kw)
        # no ws-addressing

    # op: submitBlast
    def submitBlast(self, request, **kw):
        if isinstance(request, submitBlastRequest) is False:
            raise TypeError, "%s incorrect request type" % (request.__class__)
        # no input wsaction
        self.binding.Send(None, None, request, soapaction="urn:BlastWS.wsdl#submitBlast", encodingStyle="http://schemas.xmlsoap.org/soap/encoding/", **kw)
        # no output wsaction
        typecode = Struct(pname=None, ofwhat=submitBlastResponse.typecode.ofwhat, pyclass=submitBlastResponse.typecode.pyclass)
        response = self.binding.Receive(typecode)
        return response

    # op: checkStatusBlast
    def checkStatusBlast(self, request, **kw):
        if isinstance(request, checkStatusBlastRequest) is False:
            raise TypeError, "%s incorrect request type" % (request.__class__)
        # no input wsaction
        self.binding.Send(None, None, request, soapaction="urn:BlastWS.wsdl#checkStatusBlast", encodingStyle="http://schemas.xmlsoap.org/soap/encoding/", **kw)
        # no output wsaction
        typecode = Struct(pname=None, ofwhat=checkStatusBlastResponse.typecode.ofwhat, pyclass=checkStatusBlastResponse.typecode.pyclass)
        response = self.binding.Receive(typecode)
        return response

    # op: getResultsBlast
    def getResultsBlast(self, request, **kw):
        if isinstance(request, getResultsBlastRequest) is False:
            raise TypeError, "%s incorrect request type" % (request.__class__)
        # no input wsaction
        self.binding.Send(None, None, request, soapaction="urn:BlastWS.wsdl#getResultsBlast", encodingStyle="http://schemas.xmlsoap.org/soap/encoding/", **kw)
        # no output wsaction
        typecode = Struct(pname=None, ofwhat=getResultsBlastResponse.typecode.ofwhat, pyclass=getResultsBlastResponse.typecode.pyclass)
        response = self.binding.Receive(typecode)
        return response

    # op: cancelResultsBlast
    def cancelResultsBlast(self, request, **kw):
        if isinstance(request, cancelBlastRequest) is False:
            raise TypeError, "%s incorrect request type" % (request.__class__)
        # no input wsaction
        self.binding.Send(None, None, request, soapaction="urn:BlastWS.wsdl#cancelResultsBlast", encodingStyle="http://schemas.xmlsoap.org/soap/encoding/", **kw)
        # no output wsaction
        typecode = Struct(pname=None, ofwhat=cancelBlastResponse.typecode.ofwhat, pyclass=cancelBlastResponse.typecode.pyclass)
        response = self.binding.Receive(typecode)
        return response

class submitBlastRequest:
    def __init__(self, **kw):
        """Keyword parameters:
        Sequence -- part Sequence
        SeqDatabase -- part SeqDatabase
        """
        self._Sequence =  kw.get("Sequence")
        self._SeqDatabase =  kw.get("SeqDatabase")
submitBlastRequest.typecode = Struct(pname=("urn:Blast_Service","submitBlast"), ofwhat=[ZSI.TC.String(pname="Sequence", aname="_Sequence", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True), ZSI.TC.String(pname="SeqDatabase", aname="_SeqDatabase", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=submitBlastRequest, encoded="urn:Blast_Service")

class submitBlastResponse:
    def __init__(self, **kw):
        """Keyword parameters:
        JobId -- part JobId
        """
        self._JobId =  kw.get("JobId")
submitBlastResponse.typecode = Struct(pname=("urn:Blast_Service","submitBlastResponse"), ofwhat=[ZSI.TC.String(pname="JobId", aname="_JobId", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=submitBlastResponse, encoded="urn:Blast_Service")

class checkStatusBlastRequest:
    def __init__(self, **kw):
        """Keyword parameters:
        JobId -- part JobId
        """
        self._JobId =  kw.get("JobId")
checkStatusBlastRequest.typecode = Struct(pname=("urn:Blast_Service","checkStatusBlast"), ofwhat=[ZSI.TC.String(pname="JobId", aname="_JobId", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=checkStatusBlastRequest, encoded="urn:Blast_Service")

class checkStatusBlastResponse:
    def __init__(self, **kw):
        """Keyword parameters:
        Status -- part Status
        """
        self._Status =  kw.get("Status")
checkStatusBlastResponse.typecode = Struct(pname=("urn:Blast_Service","checkStatusBlastResponse"), ofwhat=[ZSI.TC.String(pname="Status", aname="_Status", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=checkStatusBlastResponse, encoded="urn:Blast_Service")

class getResultsBlastRequest:
    def __init__(self, **kw):
        """Keyword parameters:
        JobId -- part JobId
        """
        self._JobId =  kw.get("JobId")
getResultsBlastRequest.typecode = Struct(pname=("urn:Blast_Service","getResultsBlast"), ofwhat=[ZSI.TC.String(pname="JobId", aname="_JobId", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=getResultsBlastRequest, encoded="urn:Blast_Service")

class getResultsBlastResponse:
    def __init__(self, **kw):
        """Keyword parameters:
        SeqSimilar -- part SeqSimilar
        Logout -- part Logout
        """
        self._SeqSimilar =  kw.get("SeqSimilar")
        self._Logout =  kw.get("Logout")
getResultsBlastResponse.typecode = Struct(pname=("urn:Blast_Service","getResultsBlastResponse"), ofwhat=[ZSI.TC.String(pname="SeqSimilar", aname="_SeqSimilar", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True), ZSI.TC.String(pname="Logout", aname="_Logout", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=getResultsBlastResponse, encoded="urn:Blast_Service")

class cancelBlastRequest:
    def __init__(self, **kw):
        """Keyword parameters:
        JobId -- part JobId
        """
        self._JobId =  kw.get("JobId")
cancelBlastRequest.typecode = Struct(pname=("urn:Blast_Service","cancelResultsBlast"), ofwhat=[ZSI.TC.String(pname="JobId", aname="_JobId", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=cancelBlastRequest, encoded="urn:Blast_Service")

class cancelBlastResponse:
    def __init__(self, **kw):
        """Keyword parameters:
        Status -- part Status
        """
        self._Status =  kw.get("Status")
cancelBlastResponse.typecode = Struct(pname=("urn:Blast_Service","cancelResultsBlastResponse"), ofwhat=[ZSI.TC.String(pname="Status", aname="_Status", typed=False, encoded=None, minOccurs=1, maxOccurs=1, nillable=True)], pyclass=cancelBlastResponse, encoded="urn:Blast_Service")

##################################################
# file: Blast_Service_types.py
#
# schema types generated by "ZSI.generate.wsdl2python.WriteServiceModule"
#    /usr/local/bin/wsdl2py http://gbio-pbil.ibcp.fr/ws/BlastWS.wsdl
#
##################################################

import ZSI
import ZSI.TCcompound
from ZSI.schema import LocalElementDeclaration, ElementDeclaration, TypeDefinition, GTD, GED

#!/usr/bin/env python
# ================================================================================
#
#                             COPYRIGHT NOTICE
#
#                Centre National de la Recherche Scientifique
#                                   CNRS
#
#                Institut de Biologie et Chimie des Proteines
#                             IBCP CNRS UMR 5086
#                            7 passage du Vercors
#                             69007 Lyon, FRANCE
#
#  This software/database is covered by copyright of CNRS IBCP
#  All rights reserved. Reproduction and usage of this software/database are 
#  strictly restricted, and need a prior written permission from CNRS IBCP
#  (see "Contact" below).
#
#  CNRS IBCP do not and cannot warrant the performance or results that
#  may be obtained by using this software or data. CNRS IBCP disclaim all
#  warranties, express or implied, including warranties of performance,
#  merchantability or fitness for any particular purpose.
#
# ================================================================================
#
# Author: Christophe Blanchet
#
# Contact: Christophe.Blanchet@ibcp.fr
#
# First Version Creation Date:   18/11/2008
#
# $Revision: 1.0 $
#
# File Description: 
#
#
# Modifications:
# --------------------------------------------------------------------------
# Date     Name        Description of modification
# -------  ----------  -----------------------------------------------------
#
# ================================================================================
import sys
import time
from Blast_Service_client import *

fn_input = 'benchy_seq001.seq'
fn_control = 'control-seq001.blast'
ret = 0

# get a port proxy instance
loc = Blast_ServiceLocator()
port = loc.getBlastPort()

###
# submit a new request
# create
req = submitBlastRequest()
fp_input = open (fn_input, 'r')
req._Sequence = fp_input.read()
req._SeqDatabase = "swiss-prot"


# call the remote method
resp = port.submitBlast(req)
jobid = resp._JobId
#print jobid
if not jobid:
	ret = 1

###
# check status
# while status different of 'done' or 'aborted', check again
status = 'init'
req = checkStatusBlastRequest()
req._JobId = jobid
while status not in ('done','aborted'):
	time.sleep(2) 
	resp = port.checkStatusBlast(req)
	status = resp._Status
	#print status
if status != 'done':
	ret = 1

###
# get results
req = getResultsBlastRequest()
req._JobId = jobid
resp = port.getResultsBlast(req)

# write result
#open(fn_control,'w').write(resp._SeqSimilar)

# check result
if 'RBTR_KLEAE' not in resp._SeqSimilar:
	ret = 1
#	print "wooohh"

#print ret
sys.exit (ret)