path: root/tests/axis1.py

# This program is free software; you can redistribute it and/or modify
# it under the terms of the (LGPL) GNU Lesser General Public License as
# published by the Free Software Foundation; either version 3 of the 
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Library Lesser General Public License for more details at
# ( http://www.gnu.org/licenses/lgpl.html ).
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
# written by: Jeff Ortel ( jortel@redhat.com )

#
# This test requires installation or visability to my local axis(1) server.
#

import sys
sys.path.append('../')

import logging
import traceback as tb
import suds.metrics as metrics
from tests import *
from suds import WebFault
from suds.client import Client
from suds.sudsobject import Object
from suds.transport.https import HttpAuthenticated
from suds.plugin import Plugin

errors = 0

credentials = dict(username='jortel', password='abc123')

setup_logging()


class TestPlugin(Plugin):
    
    def initialized(self, context):
        print 'initialized: ctx=%s' % context.__dict__
        
    def loaded(self, context):
        print 'loaded: ctx=%s' % context.__dict__
    
    def sending(self, context):
        print 'sending: ctx=%s' % context.__dict__

    def received(self, context):
        print 'received: ctx=%s' % context.__dict__


#logging.getLogger('suds.client').setLevel(logging.DEBUG)

def start(url):
    global errors
    print '\n________________________________________________________________\n' 
    print 'Test @ ( %s )\nerrors = %d\n' % (url, errors)

try:
    url = 'http://localhost:8081/axis/services/basic-rpc-encoded?wsdl'
    start(url)
    t = HttpAuthenticated(**credentials)
    client = Client(url, transport=t, cache=None, plugins=[TestPlugin()])
    print client
    #
    # create a name object using the wsdl
    #
    print 'create name'
    name = client.factory.create('ns0:Name')
    name.first = u'jeff'+unichr(1234)
    name.last = 'ortel'
    print name
    #
    # create a phone object using the wsdl
    #
    print 'create phone'
    phoneA = client.factory.create('ns0:Phone')
    phoneA.npa = 410
    phoneA.nxx = 555
    phoneA.number = 5138
    phoneB = client.factory.create('ns0:Phone')
    phoneB.npa = 919
    phoneB.nxx = 555
    phoneB.number = 4406
    phoneC = {
        'npa':205,
        'nxx':777,
        'number':1212
    }
    #
    # create a dog
    #
    dog = client.factory.create('ns0:Dog')
    dog.name = 'Chance'
    dog.trained = True
    #
    # create a person object using the wsdl
    #
    person = client.factory.create('ns0:Person')
    print '{empty} person=\n%s' % person
    person.name = name
    person.age = 43
    person.phone = [phoneA,phoneB,phoneC]
    person.pets = [dog]
    print 'person=\n%s' % person
    #
    # add the person (using the webservice)
    #
    print 'addPersion()'
    result = client.service.addPerson(person)
    print '\nreply(\n%s\n)\n' % str(result)
    #
    # create a new name object used to update the person
    #
    newname = client.factory.create('ns0:Name')
    newname.first = 'Todd'
    newname.last = None
    #
    # create AnotherPerson using Person
    #
    ap = client.factory.create('ns0:AnotherPerson')
    ap.name = person.name
    ap.age = person.age
    ap.phone = person.phone
    ap.pets = person.pets
    print 'AnotherPerson\n%s' % ap
    #
    # update the person's name (using the webservice)
    #
    print 'updatePersion()'
    result = client.service.updatePerson(ap, newname)
    print '\nreply(\n%s\n)\n' % str(result)
    result = client.service.updatePerson(ap, None)
    print '\nreply(\n%s\n)\n' % str(result)
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()
    
try:
    url = 'http://localhost:8081/axis/services/basic-rpc-encoded?wsdl'
    start(url)
    t = HttpAuthenticated(**credentials)
    client = Client(url, transport=t, cache=None)
    print client
    #
    # create a name object as dict
    #
    print 'create name'
    name = {}
    name['first'] = 'Elmer'
    name['last'] = 'Fudd'
    print name
    #
    # create a phone as dict
    #
    print 'create phone'
    phoneA = {}
    phoneA['npa'] = 410
    phoneA['nxx'] = 555
    phoneA['number'] = 5138
    phoneB = {}
    phoneB['npa'] = 919
    phoneB['nxx'] = 555
    phoneB['number'] = 4406
    phoneC = {
        'npa':205,
        'nxx':777,
        'number':1212
    }
    #
    # create a dog
    #
    dog = {
        'name':'Chance',
        'trained':True,
    }
    #
    # create a person as dict
    #
    person = {}
    print '{empty} person=\n%s' % person
    person['name'] = name
    person['age'] = 43
    person['phone'] = [phoneA,phoneB, phoneC]
    person['pets'] = [dog]
    print 'person=\n%s' % person
    #
    # add the person (using the webservice)
    #
    print 'addPersion()'
    result = client.service.addPerson(person)
    print '\nreply(\n%s\n)\n' % str(result)
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()
    
try:
    print "echo(' this is cool ')"
    result = client.service.echo('this is cool')
    print '\nreply( "%s" )\n' % str(result)
    print 'echo(None)'
    result = client.service.echo(None)
    print '\nreply( "%s" )\n' % str(result)
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()
    
try:
    print 'hello()'
    result = client.service.hello()
    print '\nreply( %s )\n' % str(result)
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()

try:
    print 'testVoid()'
    result = client.service.getVoid()
    print '\nreply( %s )\n' % str(result)
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()

try:
    print '** new style arrays **'
    words = ['my', 'dog', 'likes', 'steak']
    result = client.service.printList(words)
    print '\nreply( %s )\n' % str(result)

    print '** old style arrays **'
    array = client.factory.create('ArrayOf_xsd_string')
    print 'ArrayOf_xsd_string=\n%s' % array
    array.item = ['my', 'dog', 'likes', 'steak']
    result = client.service.printList(array)
    print '\nreply( %s )\n' % str(result)
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()

try:
    s = 'hello'
    for n in range(0, 3):
        print 'getList(%s, %d)' % (s, n)
        result = client.service.getList(s, n)
        print '\nreply( %s )\n' % str(result)
        assert ( isinstance(result, list) and len(result) == n )
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()

try:
    print 'testExceptions()' 
    result = client.service.throwException()
    print '\nreply( %s )\n' % tostr(result)
    raise Exception('Fault expected and not raised')
except WebFault, f:
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()

try:
    url = 'http://localhost:8081/axis/services/basic-rpc-encoded?wsdl'
    start(url)
    client = Client(url, faults=False, **credentials)
    print 'testExceptions()'
    result = client.service.throwException()
    print '\nreply( %s )\n' % str(result)
except WebFault, f:
    errors += 1
    print f
    print f.fault
except Exception, e:
    errors += 1
    print e
    tb.print_exc()
    
print '\nFinished: errors=%d' % errors