From cab19f5814c43ade1bcac7a9ab4c29c83a6cf35d Mon Sep 17 00:00:00 2001 From: Benjamin Dauvergne Date: Tue, 27 May 2008 15:27:36 +0000 Subject: remove obsolete swig binding csharp, php5, python. java stays as long as alain coetmeur needs it php(4) will be superseded by the new bindings/php4 later. --- python/.cvsignore | 9 - python/Makefile.am | 45 - python/doc/.cvsignore | 3 - python/doc/tutorial/.cvsignore | 3 - python/tests/.cvsignore | 4 - python/tests/Makefile.am | 2 - python/tests/XmlTestRunner.py | 74 -- python/tests/binding_tests.py | 324 ------- python/tests/errorchecking_tests.py | 99 -- python/tests/idwsf1_tests.py | 257 ----- python/tests/idwsf2_tests.py | 1753 ----------------------------------- python/tests/profiles_tests.py | 356 ------- python/tests/tests.py | 104 --- 13 files changed, 3033 deletions(-) delete mode 100644 python/.cvsignore delete mode 100644 python/Makefile.am delete mode 100644 python/doc/.cvsignore delete mode 100644 python/doc/tutorial/.cvsignore delete mode 100644 python/tests/.cvsignore delete mode 100644 python/tests/Makefile.am delete mode 100644 python/tests/XmlTestRunner.py delete mode 100755 python/tests/binding_tests.py delete mode 100644 python/tests/errorchecking_tests.py delete mode 100755 python/tests/idwsf1_tests.py delete mode 100755 python/tests/idwsf2_tests.py delete mode 100755 python/tests/profiles_tests.py delete mode 100755 python/tests/tests.py (limited to 'python') diff --git a/python/.cvsignore b/python/.cvsignore deleted file mode 100644 index 3129eda0..00000000 --- a/python/.cvsignore +++ /dev/null @@ -1,9 +0,0 @@ -lasso.py -lasso_wrap.c -*.pyc -.deps -.libs -Makefile -Makefile.in -*.la -*.lo diff --git a/python/Makefile.am b/python/Makefile.am deleted file mode 100644 index e99b9bf5..00000000 --- a/python/Makefile.am +++ /dev/null @@ -1,45 +0,0 @@ -INCLUDES = \ - -I$(top_builddir) \ - -I$(top_srcdir) \ - $(SASL_CFLAGS) - -pythondir= $(PY_SITE_PACKAGES) -python_DATA = lasso.py -python_LTLIBRARIES = _lasso.la - -_lasso_la_SOURCES = lasso_wrap.c -_lasso_la_CFLAGS = $(LASSO_CORE_CFLAGS) -DSWIG_COBJECT_TYPES $(PY_CFLAGS) -_lasso_la_LIBADD = $(top_builddir)/lasso/liblasso.la $(LASSO_LIBS) -if DARWIN -_lasso_la_LDFLAGS = -no-undefined -module -avoid-version -Wl,-F. -Wl,-F. -bundle -framework Python -else -if MINGW -_lasso_la_LDFLAGS = -no-undefined -module -avoid-version -Wl,--add-stdcall-alias -else -_lasso_la_LDFLAGS = -no-undefined -module -avoid-version -endif -endif -SWIG_I_FILES=$(shell find $(top_srcdir)/swig/ -name '*.[ih]') -lasso_wrap.c lasso.py: $(SWIG_I_FILES) - $(SWIG) -v -python -module lasso -o lasso_wrap.c $(top_srcdir)/swig/Lasso.i - cp lasso_wrap.c lasso_wrap.c.bak - sed -e 's/^ int res = 0;$$//' \ - < lasso_wrap.c.bak > lasso_wrap.c - cp lasso_wrap.c lasso_wrap.c.bak - sed -e 's/\/\* if (val) \*val = PyObject_IsTrue(obj); return 1; \*\//int res = 0;/' \ - < lasso_wrap.c.bak > lasso_wrap.c - cp lasso_wrap.c lasso_wrap.c.bak - grep -q "int res =" lasso_wrap.c || \ - sed -e 's/if (obj == Py_True) /int res = 0; if (obj == Py_True) /' \ - < lasso_wrap.c.bak > lasso_wrap.c - -SWIG_OUTPUTS = lasso_wrap.c lasso.py - -clean-local: - -rm lasso_wrap.c.bak - -# In distribution, swig generated files are present. -EXTRA_DIST = $(SWIG_OUTPUTS) - -# Maintainer clean the swig generated files, because he/she maintains them. -MAINTAINERCLEANFILES = Makefile.in $(SWIG_OUTPUTS) diff --git a/python/doc/.cvsignore b/python/doc/.cvsignore deleted file mode 100644 index 22a4e729..00000000 --- a/python/doc/.cvsignore +++ /dev/null @@ -1,3 +0,0 @@ -Makefile -Makefile.in - diff --git a/python/doc/tutorial/.cvsignore b/python/doc/tutorial/.cvsignore deleted file mode 100644 index 22a4e729..00000000 --- a/python/doc/tutorial/.cvsignore +++ /dev/null @@ -1,3 +0,0 @@ -Makefile -Makefile.in - diff --git a/python/tests/.cvsignore b/python/tests/.cvsignore deleted file mode 100644 index 3bfe07c0..00000000 --- a/python/tests/.cvsignore +++ /dev/null @@ -1,4 +0,0 @@ -*.pyc -Makefile -Makefile.in - diff --git a/python/tests/Makefile.am b/python/tests/Makefile.am deleted file mode 100644 index eb7dd41f..00000000 --- a/python/tests/Makefile.am +++ /dev/null @@ -1,2 +0,0 @@ -MAINTAINERCLEANFILES = Makefile.in -EXTRA_DIST = tests.py login_tests.py idwsf2_tests.py diff --git a/python/tests/XmlTestRunner.py b/python/tests/XmlTestRunner.py deleted file mode 100644 index 3739a8a9..00000000 --- a/python/tests/XmlTestRunner.py +++ /dev/null @@ -1,74 +0,0 @@ -# -*- coding: UTF-8 -*- -# -# $Id$ -# -# XmlTestRunner -# -# Copyright (C) 2004-2007 Entr'ouvert -# -# Authors: Frederic Peters -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 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 General Public License for more details. -# -# You should have received a copy of the GNU 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 - - -import unittest -import time -import sys - -def xml(text): - if not text: - return "" - return text.replace('&', '&').replace('<', '<').replace('>', '>') - -class XmlTestResult(unittest.TestResult): - def addSuccess(self, test): - print """ - %s - %s - """ % (test.id(), xml(test.shortDescription())) - - def addError(self, test, err): - unittest.TestResult.addError(self, test, err) - print """ - %s - %s - """ % (test.id(), xml(test.shortDescription())) - # TODO: add err - - def addFailure(self, test, err): - unittest.TestResult.addFailure(self, test, err) - print """ - %s - %s - """ % (test.id(), xml(test.shortDescription())) - # TODO: add err - - -class XmlTestRunner: - def _makeResult(self): - return XmlTestResult() - - def run(self, test): - print "" - result = self._makeResult() - startTime = time.time() - test(result) - stopTime = time.time() - timeTaken = float(stopTime - startTime) - print " %s" % timeTaken - print "" - - return result - diff --git a/python/tests/binding_tests.py b/python/tests/binding_tests.py deleted file mode 100755 index cc718ff9..00000000 --- a/python/tests/binding_tests.py +++ /dev/null @@ -1,324 +0,0 @@ -#! /usr/bin/env python -# -*- coding: UTF-8 -*- -# -# $Id$ -# -# Python unit tests for Lasso library -# -# Copyright (C) 2004-2007 Entr'ouvert -# http://lasso.entrouvert.org -# -# Authors: See AUTHORS file in top-level directory. -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 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 General Public License for more details. -# -# You should have received a copy of the GNU 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 - - -import unittest -import sys - -if not '..' in sys.path: - sys.path.insert(0, '..') -if not '../.libs' in sys.path: - sys.path.insert(0, '../.libs') - -import lasso - - -class BindingTestCase(unittest.TestCase): - def test01(self): - """Create and delete nodes.""" - - authnRequest = lasso.LibAuthnRequest() - del authnRequest - - def test02(self): - """Get & set simple attributes of nodes.""" - - authnRequest = lasso.LibAuthnRequest() - - # Test a string attribute. - self.failUnlessEqual(authnRequest.consent, None) - authnRequest.consent = lasso.LIB_CONSENT_OBTAINED - self.failUnlessEqual(authnRequest.consent, lasso.LIB_CONSENT_OBTAINED) - authnRequest.consent = None - self.failUnlessEqual(authnRequest.consent, None) - - # Test a renamed string attribute. - self.failUnlessEqual(authnRequest.relayState, None) - authnRequest.relayState = 'Hello World!' - self.failUnlessEqual(authnRequest.relayState, 'Hello World!') - authnRequest.relayState = None - self.failUnlessEqual(authnRequest.relayState, None) - - # Test an integer attribute. - self.failUnlessEqual(authnRequest.majorVersion, 0) - authnRequest.majorVersion = 314 - self.failUnlessEqual(authnRequest.majorVersion, 314) - - del authnRequest - - def test03(self): - """Get & set attributes of nodes of type string list.""" - - authnRequest = lasso.LibAuthnRequest() - - self.failUnlessEqual(authnRequest.respondWith, None) - - respondWith = lasso.StringList() - self.failUnlessEqual(len(respondWith), 0) - respondWith.append('first string') - self.failUnlessEqual(len(respondWith), 1) - self.failUnlessEqual(respondWith[0], 'first string') - respondWith.append('second string') - self.failUnlessEqual(len(respondWith), 2) - self.failUnlessEqual(respondWith[0], 'first string') - self.failUnlessEqual(respondWith[1], 'second string') - respondWith.append('third string') - self.failUnlessEqual(len(respondWith), 3) - self.failUnlessEqual(respondWith[0], 'first string') - self.failUnlessEqual(respondWith[1], 'second string') - self.failUnlessEqual(respondWith[2], 'third string') - authnRequest.respondWith = respondWith - self.failUnlessEqual(authnRequest.respondWith[0], 'first string') - self.failUnlessEqual(authnRequest.respondWith[1], 'second string') - self.failUnlessEqual(authnRequest.respondWith[2], 'third string') - self.failUnlessEqual(respondWith[0], 'first string') - self.failUnlessEqual(respondWith[1], 'second string') - self.failUnlessEqual(respondWith[2], 'third string') - del respondWith - self.failUnlessEqual(authnRequest.respondWith[0], 'first string') - self.failUnlessEqual(authnRequest.respondWith[1], 'second string') - self.failUnlessEqual(authnRequest.respondWith[2], 'third string') - respondWith = authnRequest.respondWith - self.failUnlessEqual(respondWith[0], 'first string') - self.failUnlessEqual(respondWith[1], 'second string') - self.failUnlessEqual(respondWith[2], 'third string') - del respondWith - self.failUnlessEqual(authnRequest.respondWith[0], 'first string') - self.failUnlessEqual(authnRequest.respondWith[1], 'second string') - self.failUnlessEqual(authnRequest.respondWith[2], 'third string') - authnRequest.respondWith = None - self.failUnlessEqual(authnRequest.respondWith, None) - - del authnRequest - - def test04(self): - """Get & set attributes of nodes of type node list.""" - - response = lasso.SamlpResponse() - - self.failUnlessEqual(response.assertion, None) - - assertions = lasso.NodeList() - self.failUnlessEqual(len(assertions), 0) - assertion1 = lasso.SamlAssertion() - assertion1.assertionId = 'assertion 1' - assertions.append(assertion1) - self.failUnlessEqual(len(assertions), 1) - self.failUnlessEqual(assertions[0].assertionId, 'assertion 1') - self.failUnlessEqual(assertions[0].assertionId, 'assertion 1') - assertion2 = lasso.SamlAssertion() - assertion2.assertionId = 'assertion 2' - assertions.append(assertion2) - self.failUnlessEqual(len(assertions), 2) - self.failUnlessEqual(assertions[0].assertionId, 'assertion 1') - self.failUnlessEqual(assertions[1].assertionId, 'assertion 2') - assertion3 = lasso.SamlAssertion() - assertion3.assertionId = 'assertion 3' - assertions.append(assertion3) - self.failUnlessEqual(len(assertions), 3) - self.failUnlessEqual(assertions[0].assertionId, 'assertion 1') - self.failUnlessEqual(assertions[1].assertionId, 'assertion 2') - self.failUnlessEqual(assertions[2].assertionId, 'assertion 3') - response.assertion = assertions - self.failUnlessEqual(response.assertion[0].assertionId, 'assertion 1') - self.failUnlessEqual(response.assertion[1].assertionId, 'assertion 2') - self.failUnlessEqual(response.assertion[2].assertionId, 'assertion 3') - self.failUnlessEqual(assertions[0].assertionId, 'assertion 1') - self.failUnlessEqual(assertions[1].assertionId, 'assertion 2') - self.failUnlessEqual(assertions[2].assertionId, 'assertion 3') - del assertions - self.failUnlessEqual(response.assertion[0].assertionId, 'assertion 1') - self.failUnlessEqual(response.assertion[1].assertionId, 'assertion 2') - self.failUnlessEqual(response.assertion[2].assertionId, 'assertion 3') - assertions = response.assertion - self.failUnlessEqual(assertions[0].assertionId, 'assertion 1') - self.failUnlessEqual(assertions[1].assertionId, 'assertion 2') - self.failUnlessEqual(assertions[2].assertionId, 'assertion 3') - del assertions - self.failUnlessEqual(response.assertion[0].assertionId, 'assertion 1') - self.failUnlessEqual(response.assertion[1].assertionId, 'assertion 2') - self.failUnlessEqual(response.assertion[2].assertionId, 'assertion 3') - response.assertion = None - self.failUnlessEqual(response.assertion, None) - - del response - - def test05(self): - """Get & set attributes of nodes of type XML list.""" - - authnRequest = lasso.LibAuthnRequest() - - self.failUnlessEqual(authnRequest.extension, None) - - actionString1 = """\ - - do 1 -""" - actionString2 = """\ - - do 2 -""" - actionString3 = """\ - - do 3 -""" - extension = lasso.StringList() - self.failUnlessEqual(len(extension), 0) - extension.append(actionString1) - self.failUnlessEqual(len(extension), 1) - self.failUnlessEqual(extension[0], actionString1) - self.failUnlessEqual(extension[0], actionString1) - extension.append(actionString2) - self.failUnlessEqual(len(extension), 2) - self.failUnlessEqual(extension[0], actionString1) - self.failUnlessEqual(extension[1], actionString2) - extension.append(actionString3) - self.failUnlessEqual(len(extension), 3) - self.failUnlessEqual(extension[0], actionString1) - self.failUnlessEqual(extension[1], actionString2) - self.failUnlessEqual(extension[2], actionString3) - authnRequest.extension = extension - self.failUnlessEqual(authnRequest.extension[0], actionString1) - self.failUnlessEqual(authnRequest.extension[1], actionString2) - self.failUnlessEqual(authnRequest.extension[2], actionString3) - self.failUnlessEqual(extension[0], actionString1) - self.failUnlessEqual(extension[1], actionString2) - self.failUnlessEqual(extension[2], actionString3) - del extension - self.failUnlessEqual(authnRequest.extension[0], actionString1) - self.failUnlessEqual(authnRequest.extension[1], actionString2) - self.failUnlessEqual(authnRequest.extension[2], actionString3) - extension = authnRequest.extension - self.failUnlessEqual(extension[0], actionString1) - self.failUnlessEqual(extension[1], actionString2) - self.failUnlessEqual(extension[2], actionString3) - del extension - self.failUnlessEqual(authnRequest.extension[0], actionString1) - self.failUnlessEqual(authnRequest.extension[1], actionString2) - self.failUnlessEqual(authnRequest.extension[2], actionString3) - authnRequest.extension = None - self.failUnlessEqual(authnRequest.extension, None) - - del authnRequest - - def test06(self): - """Get & set attributes of nodes of type node.""" - - login = lasso.Login(lasso.Server()) - - self.failUnlessEqual(login.request, None) - login.request = lasso.LibAuthnRequest() - login.request.consent = lasso.LIB_CONSENT_OBTAINED - self.failUnlessEqual(login.request.consent, lasso.LIB_CONSENT_OBTAINED) - login.request = None - self.failUnlessEqual(login.request, None) - - del login - - def test07(self): - """Get & set SAML 2.0 assertion attribute values""" - - attribute1_name = 'first attribute' - attribute1_string = 'first string' - attribute2_name = 'second attribute' - attribute2_string = 'second string' - attribute3_string = 'third string' - - expected_assertion_dump = '''\ - - - - - first string - - - - - second string - - - third string - - - -''' - - text_node1 = lasso.MiscTextNode() - text_node1.content = attribute1_string - any1 = lasso.NodeList() - any1.append(text_node1) - attribute_value1 = lasso.Saml2AttributeValue() - attribute_value1.any = any1 - attribute_values1 = lasso.NodeList() - attribute_values1.append(attribute_value1) - attribute1 = lasso.Saml2Attribute() - attribute1.name = attribute1_name - attribute1.attributeValue = attribute_values1 - - text_node2 = lasso.MiscTextNode() - text_node2.content = attribute2_string - any2 = lasso.NodeList() - any2.append(text_node2) - attribute_value2 = lasso.Saml2AttributeValue() - attribute_value2.any = any2 - - text_node3 = lasso.MiscTextNode() - text_node3.content = attribute3_string - any3 = lasso.NodeList() - any3.append(text_node3) - attribute_value3 = lasso.Saml2AttributeValue() - attribute_value3.any = any3 - - attribute_values2 = lasso.NodeList() - attribute_values2.append(attribute_value2) - attribute_values2.append(attribute_value3) - - attribute2 = lasso.Saml2Attribute() - attribute2.name = attribute2_name - attribute2.attributeValue = attribute_values2 - - attributes = lasso.NodeList() - attributes.append(attribute1) - attributes.append(attribute2) - - attributeStatement = lasso.Saml2AttributeStatement() - attributeStatement.attribute = attributes - attributeStatements = lasso.NodeList() - attributeStatements.append(attributeStatement) - - assertion = lasso.Saml2Assertion() - assertion.attributeStatement = attributeStatements - - self.failUnlessEqual(assertion.dump(), expected_assertion_dump, 'resulting assertion dump is not as expected') - - -bindingSuite = unittest.makeSuite(BindingTestCase, 'test') - -allTests = unittest.TestSuite((bindingSuite, )) - -if __name__ == '__main__': - sys.exit(not unittest.TextTestRunner(verbosity = 2).run(allTests).wasSuccessful()) - diff --git a/python/tests/errorchecking_tests.py b/python/tests/errorchecking_tests.py deleted file mode 100644 index 14b6e045..00000000 --- a/python/tests/errorchecking_tests.py +++ /dev/null @@ -1,99 +0,0 @@ -#! /usr/bin/env python -# -*- coding: UTF-8 -*- -# -# $Id$ -# -# Python unit tests for Lasso library -# -# Copyright (C) 2004-2007 Entr'ouvert -# http://lasso.entrouvert.org -# -# Authors: See AUTHORS file in top-level directory. -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 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 General Public License for more details. -# -# You should have received a copy of the GNU 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 - - -import os -import unittest -import sys - -if not '..' in sys.path: - sys.path.insert(0, '..') -if not '../.libs' in sys.path: - sys.path.insert(0, '../.libs') - -import lasso - - -try: - dataDir -except NameError: - dataDir = '../../tests/data' - - -class ErrorCheckingTestCase(unittest.TestCase): - def test01(self): - """Instanciate Login with None as Server""" - try: - lasso.Login(None).msgUrl - except: - pass - - def test02(self): - """Instanciate Logout with None as Server""" - # Same as test01; replace Login by Logout - try: - lasso.Logout(None, lasso.providerTypeSp).msgUrl - except: - pass - - def test03(self): - """Process empty string as authnrequest msg""" - # This time; we got something wrong as query string; we pass it to - # initFromAuthnRequestMsg; surely it shouldn't segfault - server = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - login = lasso.Login(server) - try: - login.processAuthnRequestMsg("") - except lasso.Error, error: - if error[0] != -407: - raise - - def test04(self): - server = lasso.Server( - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'sp1-la/certificate.pem')) - logout = lasso.Logout(server) - try: - logout.initRequest(None, lasso.HTTP_METHOD_REDIRECT) - except lasso.Error, error: - if error[0] != -418: - raise - - - -suite1 = unittest.makeSuite(ErrorCheckingTestCase, 'test') - -allTests = unittest.TestSuite((suite1,)) - -if __name__ == '__main__': - sys.exit(not unittest.TextTestRunner(verbosity=2).run(allTests).wasSuccessful()) - diff --git a/python/tests/idwsf1_tests.py b/python/tests/idwsf1_tests.py deleted file mode 100755 index 5a0214fd..00000000 --- a/python/tests/idwsf1_tests.py +++ /dev/null @@ -1,257 +0,0 @@ -#! /usr/bin/env python -# -*- coding: UTF-8 -*- -# -# Python unit tests for Lasso library -# -# Copyright (C) 2004-2007 Entr'ouvert -# http://lasso.entrouvert.org -# -# Authors: See AUTHORS file in top-level directory. -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 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 General Public License for more details. -# -# You should have received a copy of the GNU 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 - -import os -import unittest -import sys - -if not '..' in sys.path: - sys.path.insert(0, '..') -if not '../.libs' in sys.path: - sys.path.insert(0, '../.libs') - -import lasso - -try: - dataDir -except NameError: - dataDir = '../../tests/data' - -wsp_metadata = os.path.join(dataDir, 'sp1-la/metadata.xml') -wsp_private_key = os.path.join(dataDir, 'sp1-la/private-key-raw.pem') -wsp_public_key = os.path.join(dataDir, 'sp1-la/public-key.pem') -wsc_metadata = os.path.join(dataDir, 'sp2-la/metadata.xml') -wsc_private_key = os.path.join(dataDir, 'sp2-la/private-key-raw.pem') -wsc_public_key = os.path.join(dataDir, 'sp2-la/public-key.pem') -idp_metadata = os.path.join(dataDir, 'idp1-la/metadata.xml') -idp_private_key = os.path.join(dataDir, 'idp1-la/private-key-raw.pem') -idp_public_key = os.path.join(dataDir, 'idp1-la/public-key.pem') - -abstract_description = "Personal Profile Resource" -resource_id = "http://idp/user/resources/1" - -class IdWsf1TestCase(unittest.TestCase): - def get_wsp_server(self): - server = lasso.Server(wsp_metadata, wsp_private_key, None, None) - server.addProvider(lasso.PROVIDER_ROLE_IDP, idp_metadata, idp_public_key, None) - return server - - def get_wsc_server(self): - server = lasso.Server(wsc_metadata, wsc_private_key, None, None) - server.addProvider(lasso.PROVIDER_ROLE_IDP, idp_metadata, idp_public_key, None) - return server - - def get_idp_server(self): - server = lasso.Server(idp_metadata, idp_private_key, None, None) - server.addProvider(lasso.PROVIDER_ROLE_SP, wsp_metadata, wsp_public_key, None) - server.addProvider(lasso.PROVIDER_ROLE_SP, wsc_metadata, wsc_public_key, None) - return server - - def add_services(self, idp): - # Add Discovery service - disco_description = lasso.DiscoDescription.newWithBriefSoapHttpDescription( - lasso.SECURITY_MECH_NULL, - "http://idp/discovery/soapEndpoint", - "Discovery SOAP Endpoint description"); - disco_service_instance = lasso.DiscoServiceInstance( - lasso.DISCO_HREF, - "http://idp/providerId", - disco_description); - idp.addService(disco_service_instance); - - # Add Personal Profile service - pp_description = lasso.DiscoDescription.newWithBriefSoapHttpDescription( - lasso.SECURITY_MECH_NULL, - "http://idp/pp/soapEndpoint", - "Discovery SOAP Endpoint description"); - pp_service_instance = lasso.DiscoServiceInstance( - lasso.PP_HREF, - "http://idp/providerId", - pp_description); - idp.addService(pp_service_instance); - return idp - - def login(self, sp, idp): - sp_login = lasso.Login(sp) - sp_login.initAuthnRequest(sp.providerIds[0], lasso.HTTP_METHOD_POST) - sp_login.request.nameIdPolicy = lasso.LIB_NAMEID_POLICY_TYPE_FEDERATED - sp_login.request.protocolProfile = lasso.LIB_PROTOCOL_PROFILE_BRWS_POST - sp_login.buildAuthnRequestMsg() - - idp_login = lasso.Login(idp) - idp_login.processAuthnRequestMsg(sp_login.msgBody) - idp_login.validateRequestMsg(True, True) - - # Set a resource offering in the assertion - discovery_resource_id = "http://idp/discovery/resources/1" - idp_login.setResourceId(discovery_resource_id) - idp_login.buildAssertion(lasso.SAML_AUTHENTICATION_METHOD_PASSWORD, None, None, None, None) - idp_login.buildAuthnResponseMsg() - - sp_login = lasso.Login(sp) - sp_login.processAuthnResponseMsg(idp_login.msgBody) - sp_login.acceptSso() - - return sp_login.identity.dump(), sp_login.session.dump(), idp_login.identity.dump(), idp_login.session.dump() - - def get_pp_service(self): - self.wsc = self.get_wsc_server() - self.idp = self.get_idp_server() - self.idp = self.add_services(self.idp) - - # Login from WSC - sp_identity_dump, sp_session_dump, idp_identity_dump, idp_session_dump = self.login(self.wsc, self.idp) - - # Init discovery query - wsc_disco = lasso.Discovery(self.wsc) - wsc_disco.setSessionFromDump(sp_session_dump) - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType(lasso.PP_HREF) - wsc_disco.buildRequestMsg(); - - # Process query - idp_disco = lasso.Discovery(self.idp) - idp_disco.processQueryMsg(wsc_disco.msgBody) - idp_disco.setIdentityFromDump(idp_identity_dump) - - # Build resource offering - service_instance = lasso.DiscoServiceInstance( - lasso.PP_HREF, - self.idp.providerId, - lasso.DiscoDescription_newWithBriefSoapHttpDescription( - lasso.SECURITY_MECH_NULL, - 'http://idp/pp/soapEndpoint')) - resource_offering = lasso.DiscoResourceOffering(service_instance); - resource_offering.resourceId = lasso.DiscoResourceID(resource_id) - resource_offering.abstract = abstract_description - idp_disco.identity.addResourceOffering(resource_offering) - idp_disco.buildResponseMsg() - - # Process response - wsc_disco.processQueryResponseMsg(idp_disco.msgBody); - return wsc_disco.getService() - - def test01(self): - '''Test a discovery query''' - service = self.get_pp_service() - # Check service attributes - self.failUnless(service.resourceId is not None) - self.failUnless(service.resourceId.content == resource_id) - self.failUnless(service.providerId == self.wsc.providerIds[0]) - self.failUnless(service.abstractDescription == abstract_description) - - def test02(self): - '''Test a data service query''' - wsc_service = self.get_pp_service() - wsc_service.initQuery('/pp:PP/pp:InformalName', 'name') - wsc_service.buildRequestMsg() - self.failUnless(lasso.getRequestTypeFromSoapMsg(wsc_service.msgBody) - == lasso.REQUEST_TYPE_DST_QUERY) - - self.wsp = self.get_wsp_server() - wsp_service = lasso.DataService(self.wsp) - wsp_service.processQueryMsg(wsc_service.msgBody) - wsp_service.resourceData = ''' - - Damien - ''' - wsp_service.buildResponseMsg() - - wsc_service.processQueryResponseMsg(wsp_service.msgBody) - self.failUnless(wsc_service.getAnswer('/pp:PP/pp:InformalName') == - 'Damien') - - def test03(self): - '''Test a data service modify''' - wsc_service = self.get_pp_service() - - xpath = '/pp:PP/pp:InformalName' - old_data = ''' - - Damien - ''' - new_data = 'Alain' - - new_full_data = ''' - Alain - ''' - - wsc_service.initModify(xpath, new_data) - wsc_service.buildRequestMsg() - self.failUnless(lasso.getRequestTypeFromSoapMsg(wsc_service.msgBody) - == lasso.REQUEST_TYPE_DST_MODIFY) - - self.wsp = self.get_wsp_server() - wsp_service = lasso.DataService(self.wsp) - wsp_service.processModifyMsg(wsc_service.msgBody) - wsp_service.resourceData = old_data - wsp_service.buildModifyResponseMsg() - # Save the new wsp_service.resourceData here - - self.failUnless(wsp_service.resourceData == new_full_data) - - wsc_service.processModifyResponseMsg(wsp_service.msgBody) - - def test10(self): - '''Test a data service modify - root element''' - wsc_service = self.get_pp_service() - - xpath = '/pp:PP' - old_data = ''' - - Damien - ''' - new_data = ''' - Alain - ''' - - new_full_data = ''' - Alain - ''' - - wsc_service.initModify(xpath, new_data) - wsc_service.buildRequestMsg() - - self.failUnless(lasso.getRequestTypeFromSoapMsg(wsc_service.msgBody) - == lasso.REQUEST_TYPE_DST_MODIFY) - - self.wsp = self.get_wsp_server() - wsp_service = lasso.DataService(self.wsp) - wsp_service.processModifyMsg(wsc_service.msgBody) - wsp_service.resourceData = old_data - wsp_service.buildModifyResponseMsg() - # Save the new wsp_service.resourceData here - - self.failUnless(wsp_service.resourceData == new_full_data) - - wsc_service.processModifyResponseMsg(wsp_service.msgBody) - - -idWsf1Suite = unittest.makeSuite(IdWsf1TestCase, 'test') - -allTests = unittest.TestSuite((idWsf1Suite)) - -if __name__ == '__main__': - sys.exit(not unittest.TextTestRunner(verbosity = 2).run(allTests).wasSuccessful()) - diff --git a/python/tests/idwsf2_tests.py b/python/tests/idwsf2_tests.py deleted file mode 100755 index ce889b2e..00000000 --- a/python/tests/idwsf2_tests.py +++ /dev/null @@ -1,1753 +0,0 @@ -#! /usr/bin/env python -# -*- coding: UTF-8 -*- -# -# $Id: idwsf2_tests.py 3254 2007-06-05 21:23:57Z dlaniel $ -# -# Python unit tests for Lasso library -# -# Copyright (C) 2004-2007 Entr'ouvert -# http://lasso.entrouvert.org -# -# Authors: See AUTHORS file in top-level directory. -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 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 General Public License for more details. -# -# You should have received a copy of the GNU 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 - - -import os -import unittest -import sys - -if not '..' in sys.path: - sys.path.insert(0, '..') -if not '../.libs' in sys.path: - sys.path.insert(0, '../.libs') - -import lasso - - -try: - dataDir -except NameError: - dataDir = '../../tests/data' - - -class IdWsf2TestCase(unittest.TestCase): - def getWspServer(self): - wsp_metadata = os.path.join(dataDir, 'sp5-saml2/metadata.xml') - wsp_private_key = os.path.join(dataDir, 'sp5-saml2/private-key.pem') - idp_metadata = os.path.join(dataDir, 'idp5-saml2/metadata.xml') - - server = lasso.Server(wsp_metadata, wsp_private_key, None, None) - server.addProvider(lasso.PROVIDER_ROLE_IDP, idp_metadata, None, None) - server.setEncryptionPrivateKey(wsp_private_key); - - return server; - - def getWscServer(self): - wsc_metadata = os.path.join(dataDir, 'sp6-saml2/metadata.xml') - wsc_private_key = os.path.join(dataDir, 'sp6-saml2/private-key.pem') - idp_metadata = os.path.join(dataDir, 'idp5-saml2/metadata.xml') - - server = lasso.Server(wsc_metadata, wsc_private_key, None, None) - server.addProvider(lasso.PROVIDER_ROLE_IDP, idp_metadata, None, None) - - return server; - - def getIdpServer(self): - if hasattr(self, 'idp_server_dump') and self.idp_server_dump is not None: - server = lasso.Server.newFromDump(self.idp_server_dump) - else: - idp_metadata = os.path.join(dataDir, 'idp5-saml2/metadata.xml') - idp_private_key = os.path.join(dataDir, 'idp5-saml2/private-key.pem') - wsp_metadata = os.path.join(dataDir, 'sp5-saml2/metadata.xml') - wsc_metadata = os.path.join(dataDir, 'sp6-saml2/metadata.xml') - - server = lasso.Server(idp_metadata, idp_private_key, None, None) - server.addProvider(lasso.PROVIDER_ROLE_SP, wsp_metadata, None, None) - server.getProvider(server.providerIds[0]).setEncryptionMode(lasso.ENCRYPTION_MODE_NAMEID); - server.addProvider(lasso.PROVIDER_ROLE_SP, wsc_metadata, None, None) - self.idp_server_dump = server.dump() - - return server - - def idpRegisterSelf(self, idp_server): - disco = lasso.IdWsf2Discovery(idp_server) - service_type = lasso.IDWSF2_DISCO_HREF - abstract = 'Disco service' - soapEndpoint = 'http://idp1/soapEndpoint' - disco.metadataRegisterSelf(service_type, abstract, soapEndpoint) - - return idp_server - - def metadataRegister(self, wsp, idp): - wsp_disco = lasso.IdWsf2Discovery(wsp) - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataRegisterMsg(wsp_disco.msgBody) - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataRegisterResponseMsg(idp_disco.msgBody) - return idp, wsp_disco.svcMDID - - def login(self, sp, idp, sp_identity_dump=None, sp_session_dump=None, - idp_identity_dump=None, idp_session_dump=None): - sp_login = lasso.Login(sp) - idp_provider_id = 'http://idp5/metadata' - sp_login.initAuthnRequest(idp_provider_id, lasso.HTTP_METHOD_REDIRECT) - sp_login.request.nameIDPolicy.format = lasso.SAML2_NAME_IDENTIFIER_FORMAT_PERSISTENT - sp_login.request.nameIDPolicy.allowCreate = True - sp_login.buildAuthnRequestMsg() - - idp_login = lasso.Login(idp) - query = sp_login.msgUrl.split('?')[1] - if idp_identity_dump is not None: - idp_login.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_login.setSessionFromDump(idp_session_dump) - idp_login.processAuthnRequestMsg(query) - idp_login.validateRequestMsg(True, True) - idp_login.buildAssertion(lasso.SAML_AUTHENTICATION_METHOD_PASSWORD, None, None, None, None) - idp_login.buildArtifactMsg(lasso.HTTP_METHOD_ARTIFACT_GET) - artifact_message = idp_login.artifactMessage - - if idp_login.isIdentityDirty: - idp_identity_dump = idp_login.identity.dump() - if idp_login.isSessionDirty: - idp_session_dump = idp_login.session.dump() - - sp_login = lasso.Login(sp) - query = idp_login.msgUrl.split('?')[1] - query = query.replace("%3D", "=") - sp_login.initRequest(query, lasso.HTTP_METHOD_ARTIFACT_GET) - sp_login.buildRequestMsg() - - idp_login = lasso.Login(idp) - idp_login.processRequestMsg(sp_login.msgBody) - idp_login.artifactMessage = artifact_message - idp_login.buildResponseMsg(None) - - sp_login.processResponseMsg(idp_login.msgBody) - sp_login.acceptSso() - if sp_login.isIdentityDirty: - sp_identity_dump = sp_login.identity.dump() - if sp_login.isSessionDirty: - sp_session_dump = sp_login.session.dump() - - return sp_identity_dump, sp_session_dump, idp_identity_dump, idp_session_dump - - -class IdpSelfRegistrationTestCase(IdWsf2TestCase): - def test01(self): - """Register IdP as Dicovery Service and get a random svcMDID""" - - disco = lasso.IdWsf2Discovery(self.getIdpServer()) - - service_type = lasso.IDWSF2_DISCO_HREF - abstract = 'Disco service' - soapEndpoint = 'http://idp1/soapEndpoint' - - svcMDID = disco.metadataRegisterSelf(service_type, abstract, soapEndpoint) - # In real use, store the server dump here - - self.failUnless(svcMDID, 'missing svcMDID') - - def test02(self): - """Register IdP as Dicovery Service with a given svcMDID""" - - disco = lasso.IdWsf2Discovery(self.getIdpServer()) - - service_type = lasso.IDWSF2_DISCO_HREF - abstract = 'Disco service' - soapEndpoint = 'http://idp1/soapEndpoint' - mySvcMDID = 'RaNdOm StRiNg' - - svcMDID = disco.metadataRegisterSelf(service_type, abstract, soapEndpoint, mySvcMDID) - # In real use, store the server dump here - - self.failUnless(svcMDID, 'missing svcMDID') - self.failUnlessEqual(svcMDID, mySvcMDID, 'wrong svcMDID') - - def test03(self): - """Register IdP as Dicovery Service with wrong parameters""" - - disco = lasso.IdWsf2Discovery(self.getIdpServer()) - - service_type = None - abstract = 'Disco service' - soapEndpoint = 'http://idp1/soapEndpoint' - - svcMDID = disco.metadataRegisterSelf(service_type, abstract, soapEndpoint) - - self.failIf(svcMDID, 'svcMDID should not be set') - - -class MetadataRegisterTestCase(IdWsf2TestCase): - def test01(self): - """Init metadata registration request""" - - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - try: - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - except lasso.Error, e: - self.fail(e) - - - def test02(self): - """Build metadata registration request""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - self.failUnless(wsp_disco.msgBody, 'missing soap request') - - def test03(self): - """Check metadata registration request type""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - request_type = lasso.getRequestTypeFromSoapMsg(wsp_disco.msgBody) - self.failUnlessEqual(request_type, lasso.REQUEST_TYPE_IDWSF2_DISCO_SVCMD_REGISTER, - 'wrong request type in metadata_register : %s' % request_type) - - def test04(self): - """Process metadata registration request""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - try: - idp_disco.processMetadataRegisterMsg(wsp_disco.msgBody) - except lasso.Error, e: - self.fail(e) - - def test05(self): - """Check metadata registration on the Discovery service""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataRegisterMsg(wsp_disco.msgBody) - - self.failUnless(idp_disco.metadata.dump(), 'missing registered metadata') - - def test06(self): - """Build metadata registration response""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataRegisterMsg(wsp_disco.msgBody) - idp_disco.buildResponseMsg() - - self.failUnless(idp_disco.msgBody, 'missing soap answer') - - def test07(self): - """Process metadata registration response""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataRegisterMsg(wsp_disco.msgBody) - idp_disco.buildResponseMsg() - - try: - wsp_disco.processMetadataRegisterResponseMsg(idp_disco.msgBody) - except lasso.Error, e: - self.fail(e) - - def test08(self): - """Check metadata registration on the WSP""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - wsp_disco = lasso.IdWsf2Discovery(wsp) - - abstract = 'Personal Profile service' - soapEndpoint = 'http://idp1/soapEndpoint' - wsp_disco.initMetadataRegister( - 'urn:liberty:id-sis-pp:2005-05', abstract, wsp.providerIds[0], soapEndpoint) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataRegisterMsg(wsp_disco.msgBody) - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataRegisterResponseMsg(idp_disco.msgBody) - - self.failUnless(wsp_disco.svcMDID, 'missing svcMDID') - -class MetadataAssociationAddTestCase(IdWsf2TestCase): - def test01(self): - """Init metadata association add request""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - - try: - wsp_disco.initMetadataAssociationAdd(svcMDID) - except lasso.Error, e: - self.fail(e) - - def test02(self): - """Init metadata association add request without login""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - - try: - wsp_disco.initMetadataAssociationAdd(svcMDID) - except lasso.Error, e: - if e[0] != lasso.PROFILE_ERROR_SESSION_NOT_FOUND: - self.fail(e) - else: - self.fail('Should have a "session not found" exception') - - def test03(self): - """Init metadata association add request - msgUrl construction""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - - self.failUnless(wsp_disco.msgUrl, 'missing msgUrl') - - def test04(self): - """Build metadata association add request""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - self.failUnless(wsp_disco.msgBody, 'missing msgBody') - - def test05(self): - """Process metadata association add request""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - - try: - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - except lasso.Error, e: - self.fail(e) - - def test06(self): - """Register metadata association""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - try: - idp_disco.registerMetadata() - except lasso.Error, e: - self.fail(e) - - def test07(self): - """Check metadata association registration""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - idp_disco.registerMetadata() - - self.failUnless(idp_disco.isIdentityDirty, 'identity has not changed, it should contain a svcMDID') - self.failUnless(idp_disco.identity.dump() != idp_identity_dump, - 'identity dump has not changed, it should contain a svcMDID') - - def test08(self): - """Build metadata association add response""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - - idp_disco.buildResponseMsg() - - self.failUnless(idp_disco.msgBody) - - def test09(self): - """Process metadata association add response""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - try: - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - except lasso.Error, e: - self.fail(e) - - -class DiscoveryQueryTestCase(IdWsf2TestCase): - def test01(self): - """Init discovery query""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - try: - wsc_disco.initQuery() - except lasso.Error, e: - self.fail(e) - - def test02(self): - """Init discovery query without login""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - - wsc_disco = lasso.IdWsf2Discovery(wsc) - - try: - wsc_disco.initQuery() - except lasso.Error, e: - if e[0] != lasso.PROFILE_ERROR_SESSION_NOT_FOUND: - self.fail(e) - else: - self.fail('Should have a "session not found" exception') - - def test03(self): - """Init discovery query - check msg url""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - - self.failUnless(wsc_disco.msgUrl, 'missing msgUrl') - - def test04(self): - """Add requested service type to discovery query""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - - try: - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - except lasso.Error, e: - self.fail(e) - - def test05(self): - """Build discovery query""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - self.failUnless(wsc_disco.msgBody) - - def test06(self): - """Process discovery query""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - try: - idp_disco.processQueryMsg(wsc_disco.msgBody) - except lasso.Error, e: - self.fail(e) - - def test07(self): - """Process discovery query and check name identifier""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - idp_disco.processQueryMsg(wsc_disco.msgBody) - - self.failUnless(idp_disco.nameIdentifier and idp_disco.nameIdentifier.content, - 'missing name identifier') - - def test08(self): - """Build discovery query response EPRs""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - idp_disco.processQueryMsg(wsc_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - try: - idp_disco.buildQueryResponseEprs() - except lasso.Error, e: - self.fail(e) - - def test09(self): - """Build discovery query response""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - idp_disco.processQueryMsg(wsc_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - idp_disco.buildQueryResponseEprs() - idp_disco.buildResponseMsg() - - self.failUnless(idp_disco.msgBody, 'missing msgBody') - - def test10(self): - """Process discovery query response""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - idp_disco.processQueryMsg(wsc_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - idp_disco.buildQueryResponseEprs() - idp_disco.buildResponseMsg() - - try: - wsc_disco.processQueryResponseMsg(idp_disco.msgBody) - except lasso.Error, e: - self.fail(e) - - def test11(self): - """Check discovery query result""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - idp_disco.processQueryMsg(wsc_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - idp_disco.buildQueryResponseEprs() - idp_disco.buildResponseMsg() - - wsc_disco.processQueryResponseMsg(idp_disco.msgBody) - - self.failUnless(wsc_disco.getService(), 'missing service after discovery query') - - -class DataServiceQueryTestCase(IdWsf2TestCase): - def getProfileService(self): - """Check discovery query result""" - idp = self.getIdpServer() - idp = self.idpRegisterSelf(idp) - wsp = self.getWspServer() - idp, svcMDID = self.metadataRegister(wsp, idp) - wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump = self.login(wsp, idp) - - wsp_disco = lasso.IdWsf2Discovery(wsp) - if wsp_identity_dump is not None: - wsp_disco.setIdentityFromDump(wsp_identity_dump) - if wsp_session_dump is not None: - wsp_disco.setSessionFromDump(wsp_session_dump) - wsp_disco.initMetadataAssociationAdd(svcMDID) - wsp_disco.buildRequestMsg() - - idp_disco = lasso.IdWsf2Discovery(idp) - idp_disco.processMetadataAssociationAddMsg(wsp_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - - idp_disco.registerMetadata() - if idp_disco.isIdentityDirty: - idp_identity_dump = idp_disco.identity.dump() - if idp_disco.isSessionDirty: - idp_session_dump = idp_disco.session.dump() - idp_disco.buildResponseMsg() - - wsp_disco.processMetadataAssociationAddResponseMsg(idp_disco.msgBody) - - wsc = self.getWscServer() - wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump = \ - self.login(wsc, idp, None, None, idp_identity_dump, idp_session_dump) - - wsc_disco = lasso.IdWsf2Discovery(wsc) - if wsc_identity_dump is not None: - wsc_disco.setIdentityFromDump(wsc_identity_dump) - if wsc_session_dump is not None: - wsc_disco.setSessionFromDump(wsc_session_dump) - - wsc_disco.initQuery() - wsc_disco.addRequestedServiceType('urn:liberty:id-sis-pp:2005-05') - wsc_disco.buildRequestMsg() - - idp_disco.processQueryMsg(wsc_disco.msgBody) - if idp_identity_dump is not None: - idp_disco.setIdentityFromDump(idp_identity_dump) - if idp_session_dump is not None: - idp_disco.setSessionFromDump(idp_session_dump) - idp_disco.buildQueryResponseEprs() - idp_disco.buildResponseMsg() - - wsc_disco.processQueryResponseMsg(idp_disco.msgBody) - - return wsc_disco.getService(), wsp - - def test01(self): - """Data service init query""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - - try: - service.initQuery() - except lasso.Error, e: - self.fail(e) - - def test02(self): - """Data service init query - msgUrl construction""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - - self.failUnless(service.msgUrl, 'missing msgUrl') - - def test03(self): - """Data service add query item""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - - try: - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - except lasso.Error, e: - self.fail(e) - - def test04(self): - """Data service build query""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.buildRequestMsg() - - self.failUnless(service.msgBody, 'missing msgBody') - - def test05(self): - """Data service build query with multiple items""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - self.failUnless(service.msgBody, 'missing msgBody') - - def test06(self): - """Data service process query""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - - try: - wsp_service.processQueryMsg(service.msgBody) - except lasso.Error, e: - self.fail(e) - - def test07(self): - """Data service check service type""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - self.failUnless(wsp_service.type, 'service type is not set') - self.failUnless(wsp_service.type == 'urn:liberty:id-sis-pp:2005-05', 'wrong service type') - - def test08(self): - """Data service get query items""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - self.failUnless(wsp_service.queryItems, 'queryItems list is None or empty') - - def test09(self): - """Data service check query items""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - items = [ '/pp2:PP/pp2:InformalName', 'not existing attribute', '/pp2:PP/pp2:MsgContact' ] - for i in range(3): - self.failUnless(wsp_service.queryItems[i] == items[i], - "query items don't match : %s != %s" % (wsp_service.queryItems[i], items[i])) - - def test10(self): - """Data service check name identifier""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - self.failUnless(wsp_service.nameIdentifier and wsp_service.nameIdentifier.content, - 'missing name identifier') - - def test11(self): - """Data service parse query items - success""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - try: - wsp_service.parseQueryItems() - except lasso.Error, e: - self.fail(e) - - def test12(self): - """Data service parse query items - failure - no item""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - try: - wsp_service.parseQueryItems() - except lasso.Error, e: - if e[0] != lasso.DST_ERROR_QUERY_FAILED: - self.fail(e) - else: - self.fail('query items parsing should have failed because no item was requested') - - def test13(self): - """Data service parse query items - failure - wrong item""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('not existing attribute', 'not existing attribute') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - try: - wsp_service.parseQueryItems() - except lasso.Error, e: - if e[0] != lasso.DST_ERROR_QUERY_FAILED: - self.fail(e) - else: - self.fail('query items parsing should have failed because a wrong query item was requested') - - def test14(self): - """Data service parse query items - failure - no data""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - try: - wsp_service.parseQueryItems() - except lasso.Error, e: - if e[0] != lasso.DST_ERROR_MISSING_SERVICE_DATA: - self.fail(e) - else: - self.fail('query items parsing should have failed because no data was provided') - - - def test15(self): - """Data service parse query items - partial failure""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - """ - - # No exception should be raised here but one will be raised on the WSC - # when parsing response status - try: - wsp_service.parseQueryItems() - except lasso.Error, e: - self.fail(e) - - def test16(self): - """Data service build query response""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - wsp_service.parseQueryItems() - wsp_service.buildResponseMsg() - - self.failUnless(wsp_service.msgBody, 'missing msgBody') - - def test17(self): - """Data service process query response - success""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - wsp_service.parseQueryItems() - wsp_service.buildResponseMsg() - - try: - service.processQueryResponseMsg(wsp_service.msgBody) - except lasso.Error, e: - self.fail(e) - - def test18(self): - """Data service process query response - failure""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('not existing attribute', 'not existing attribute') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - try: - wsp_service.parseQueryItems() - except lasso.Error, e: - pass - wsp_service.buildResponseMsg() - - try: - service.processQueryResponseMsg(wsp_service.msgBody) - except lasso.Error, e: - if e[0] != lasso.DST_ERROR_QUERY_FAILED: - self.fail(e) - else: - self.fail('response should have a "failed" status because a wrong query item was requested') - - def test19(self): - """Data service process query response - partial failure""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - wsp_service.parseQueryItems() - wsp_service.buildResponseMsg() - - try: - service.processQueryResponseMsg(wsp_service.msgBody) - except lasso.Error, e: - if e[0] != lasso.DST_ERROR_QUERY_PARTIALLY_FAILED: - self.fail(e) - else: - self.fail('response should have a "partially failed" status because a wrong query item was requested') - - def test20(self): - """Data service get first attribute""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - wsp_service.parseQueryItems() - wsp_service.buildResponseMsg() - - service.processQueryResponseMsg(wsp_service.msgBody) - informal_name = service.getAttributeNode() - - self.failUnlessEqual(informal_name, """User name""", 'first attribute node is wrong') - - def test21(self): - """Data service get attribute string""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - wsp_service.parseQueryItems() - wsp_service.buildResponseMsg() - - service.processQueryResponseMsg(wsp_service.msgBody) - informal_name = service.getAttributeString('name') - - self.failUnlessEqual(informal_name, 'User name', 'attribute string is wrong') - - def test22(self): - """Data service get attribute node""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - wsp_service.parseQueryItems() - wsp_service.buildResponseMsg() - - service.processQueryResponseMsg(wsp_service.msgBody) - email = service.getAttributeNode('email') - - expected_result = """.*?Email account.*?Email server.*?""" - - import re - result = re.findall(expected_result, email, re.DOTALL) - - self.failUnless(len(result) == 1, 'attribute node is wrong') - - def test23(self): - """Data service get attribute node - partial failure""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - wsp_service.data = """ - User name - - Email account - Email server - - """ - - wsp_service.parseQueryItems() - wsp_service.buildResponseMsg() - - try: - service.processQueryResponseMsg(wsp_service.msgBody) - except lasso.Error, e: - if e[0] == lasso.DST_ERROR_QUERY_PARTIALLY_FAILED: - pass - informal_name = service.getAttributeString('name') - email = service.getAttributeNode('email') - - self.failUnlessEqual(informal_name, 'User name', 'attribute string is wrong') - self.failUnlessEqual(email, None, 'attribute node should be None') - - def test24(self): - """Data service redirect request""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - if '/pp2:PP/pp2:MsgContact' in wsp_service.queryItems: - wsp_service.initRedirectUserForConsent('http://sp5/consent'); - wsp_service.buildResponseMsg() - - try: - service.processQueryResponseMsg(wsp_service.msgBody) - except lasso.Error, e: - if e[0] != lasso.SOAP_FAULT_REDIRECT_REQUEST: - self.fail(e) - else: - self.fail('a "soap fault redirect request" exception should have been raised') - - def test25(self): - """Data service redirect request - check redirectUrl""" - service, wsp = self.getProfileService() - lasso.registerIdWsf2DstService('pp2', 'urn:liberty:id-sis-pp:2005-05') - service.initQuery() - service.addQueryItem('/pp2:PP/pp2:InformalName', 'name') - service.addQueryItem('not existing attribute', 'not existing attribute') - service.addQueryItem('/pp2:PP/pp2:MsgContact', 'email') - service.buildRequestMsg() - - wsp_service = lasso.IdWsf2DataService(wsp) - wsp_service.processQueryMsg(service.msgBody) - - if '/pp2:PP/pp2:MsgContact' in wsp_service.queryItems: - wsp_service.initRedirectUserForConsent('http://sp5/consent'); - wsp_service.buildResponseMsg() - - try: - service.processQueryResponseMsg(wsp_service.msgBody) - except lasso.Error, e: - if e[0] == lasso.SOAP_FAULT_REDIRECT_REQUEST: - pass - - self.failUnlessEqual(service.redirectUrl, 'http://sp5/consent', 'redirectUrl is not set or wrong') - - -idpSelfRegistrationSuite = unittest.makeSuite(IdpSelfRegistrationTestCase, 'test') -metadataRegisterSuite = unittest.makeSuite(MetadataRegisterTestCase, 'test') -metadataAssociationAddSuite = unittest.makeSuite(MetadataAssociationAddTestCase, 'test') -discoveryQuerySuite = unittest.makeSuite(DiscoveryQueryTestCase, 'test') -dataServiceQuerySuite = unittest.makeSuite(DataServiceQueryTestCase, 'test') - -allTests = unittest.TestSuite((idpSelfRegistrationSuite, metadataRegisterSuite, - metadataAssociationAddSuite, discoveryQuerySuite, dataServiceQuerySuite)) - -if __name__ == '__main__': - sys.exit(not unittest.TextTestRunner(verbosity = 2).run(allTests).wasSuccessful()) - diff --git a/python/tests/profiles_tests.py b/python/tests/profiles_tests.py deleted file mode 100755 index 60da6fe7..00000000 --- a/python/tests/profiles_tests.py +++ /dev/null @@ -1,356 +0,0 @@ -#! /usr/bin/env python -# -*- coding: UTF-8 -*- -# -# $Id$ -# -# Python unit tests for Lasso library -# -# Copyright (C) 2004-2007 Entr'ouvert -# http://lasso.entrouvert.org -# -# Authors: See AUTHORS file in top-level directory. -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 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 General Public License for more details. -# -# You should have received a copy of the GNU 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 - - -import os -import unittest -import sys - -if not '..' in sys.path: - sys.path.insert(0, '..') -if not '../.libs' in sys.path: - sys.path.insert(0, '../.libs') - -import lasso - - -try: - dataDir -except NameError: - dataDir = '../../tests/data' - - -class ServerTestCase(unittest.TestCase): - def test01(self): - """Server construction, dump & newFromDump.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'sp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_IDP, - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/public-key.pem'), - os.path.join(dataDir, 'idp1-la/certificate.pem')) - dump = lassoServer.dump() - lassoServer2 = lassoServer.newFromDump(dump) - dump2 = lassoServer2.dump() - self.failUnlessEqual(dump, dump2) - - def test02(self): - """Server construction without argument, dump & newFromDump.""" - - lassoServer = lasso.Server() - lassoServer.addProvider( - lasso.PROVIDER_ROLE_IDP, - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/public-key.pem')) - dump = lassoServer.dump() - lassoServer2 = lassoServer.newFromDump(dump) - dump2 = lassoServer2.dump() - self.failUnlessEqual(dump, dump2) - - -class LoginTestCase(unittest.TestCase): - def test01(self): - """SP login; testing access to authentication request.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'sp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_IDP, - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/public-key.pem'), - os.path.join(dataDir, 'idp1-la/certificate.pem')) - login = lasso.Login(lassoServer) - login.initAuthnRequest() - login.request.protocolProfile = lasso.LIB_PROTOCOL_PROFILE_BRWS_ART - self.failUnlessEqual(login.request.protocolProfile, lasso.LIB_PROTOCOL_PROFILE_BRWS_ART) - - def test02(self): - """SP login; testing processing of an empty Response.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'sp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_IDP, - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/public-key.pem'), - os.path.join(dataDir, 'idp1-la/certificate.pem')) - login = lasso.Login(lassoServer) - try: - login.processResponseMsg('') - except lasso.Error, error: - if error[0] != lasso.PROFILE_ERROR_INVALID_MSG: - raise - - def test03(self): - """Conversion of a lib:AuthnRequest with an AuthnContext into a query and back.""" - - sp = lasso.Server( - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'sp1-la/certificate.pem')) - sp.addProvider( - lasso.PROVIDER_ROLE_IDP, - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/public-key.pem'), - os.path.join(dataDir, 'idp1-la/certificate.pem')) - spLogin = lasso.Login(sp) - spLogin.initAuthnRequest() - requestAuthnContext = lasso.LibRequestAuthnContext() - authnContextClassRefsList = lasso.StringList() - authnContextClassRefsList.append( - lasso.LIB_AUTHN_CONTEXT_CLASS_REF_PASSWORD) - requestAuthnContext.authnContextClassRef = authnContextClassRefsList - spLogin.request.requestAuthnContext = requestAuthnContext - spLogin.request.protocolProfile = lasso.LIB_PROTOCOL_PROFILE_BRWS_ART - spLogin.buildAuthnRequestMsg() - authnRequestUrl = spLogin.msgUrl - authnRequestQuery = spLogin.msgUrl[spLogin.msgUrl.index('?') + 1:] - idp = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - idp.addProvider( - lasso.PROVIDER_ROLE_SP, - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/public-key.pem'), - os.path.join(dataDir, 'sp1-la/certificate.pem')) - idpLogin = lasso.Login(idp) - idpLogin.processAuthnRequestMsg(authnRequestQuery) - self.failUnless(idpLogin.request.requestAuthnContext) - authnContextClassRefsList = idpLogin.request.requestAuthnContext.authnContextClassRef - self.failUnlessEqual(len(authnContextClassRefsList), 1) - self.failUnlessEqual(authnContextClassRefsList[0], - lasso.LIB_AUTHN_CONTEXT_CLASS_REF_PASSWORD) - - def test04(self): - """Conversion of a lib:AuthnRequest with extensions into a query and back.""" - - sp = lasso.Server( - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'sp1-la/certificate.pem')) - sp.addProvider( - lasso.PROVIDER_ROLE_IDP, - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/public-key.pem'), - os.path.join(dataDir, 'idp1-la/certificate.pem')) - spLogin = lasso.Login(sp) - spLogin.initAuthnRequest() - requestAuthnContext = lasso.LibRequestAuthnContext() - extensionList = lasso.StringList() - for extension in ( - 'do', - 'do action 2do action 3'): - extensionList.append( - '%s' - % extension) - spLogin.request.extension = extensionList - spLogin.request.protocolProfile = lasso.LIB_PROTOCOL_PROFILE_BRWS_ART - spLogin.buildAuthnRequestMsg() - authnRequestUrl = spLogin.msgUrl - authnRequestQuery = spLogin.msgUrl[spLogin.msgUrl.index('?') + 1:] - idp = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - idp.addProvider( - lasso.PROVIDER_ROLE_SP, - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/public-key.pem'), - os.path.join(dataDir, 'sp1-la/certificate.pem')) - idpLogin = lasso.Login(idp) - idpLogin.processAuthnRequestMsg(authnRequestQuery) - self.failUnless(idpLogin.request.extension) - extensionsList = idpLogin.request.extension - self.failUnlessEqual(len(extensionsList), 1) - self.failUnless('do' in extensionsList[0]) - self.failUnless('do action 2' in extensionsList[0]) - self.failUnless('do action 3' in extensionsList[0]) - - -class LogoutTestCase(unittest.TestCase): - def test01(self): - """SP logout without session and identity; testing initRequest.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'sp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_IDP, - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/public-key.pem'), - os.path.join(dataDir, 'idp1-la/certificate.pem')) - logout = lasso.Logout(lassoServer) - try: - logout.initRequest() - except lasso.Error, error: - if error[0] != lasso.PROFILE_ERROR_SESSION_NOT_FOUND: - raise - else: - self.fail('logout.initRequest without having set identity before should fail') - - def test02(self): - """IDP logout without session and identity; testing logout.getNextProviderId.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_SP, - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/public-key.pem'), - os.path.join(dataDir, 'sp1-la/certificate.pem')) - logout = lasso.Logout(lassoServer) - self.failIf(logout.getNextProviderId()) - - def test03(self): - """IDP logout; testing processRequestMsg with non Liberty query.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_SP, - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/public-key.pem'), - os.path.join(dataDir, 'sp1-la/certificate.pem')) - logout = lasso.Logout(lassoServer) - # The processRequestMsg should fail but not abort. - try: - logout.processRequestMsg('passport=0&lasso=1') - except lasso.Error, error: - if error[0] != lasso.PROFILE_ERROR_INVALID_MSG: - raise - else: - self.fail('Logout processRequestMsg should have failed.') - - def test04(self): - """IDP logout; testing processResponseMsg with non Liberty query.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_SP, - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/public-key.pem'), - os.path.join(dataDir, 'sp1-la/certificate.pem')) - logout = lasso.Logout(lassoServer) - # The processResponseMsg should fail but not abort. - try: - logout.processResponseMsg('liberty=&alliance') - except lasso.Error, error: - if error[0] != lasso.PROFILE_ERROR_INVALID_MSG: - raise - else: - self.fail('Logout processResponseMsg should have failed.') - - def test05(self): - """IDP logout; testing logout dump & newFromDump().""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_SP, - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/public-key.pem'), - os.path.join(dataDir, 'sp1-la/certificate.pem')) - - -class DefederationTestCase(unittest.TestCase): - def test01(self): - """IDP initiated defederation; testing processNotificationMsg with non Liberty query.""" - - lassoServer = lasso.Server( - os.path.join(dataDir, 'idp1-la/metadata.xml'), - os.path.join(dataDir, 'idp1-la/private-key-raw.pem'), - None, - os.path.join(dataDir, 'idp1-la/certificate.pem')) - lassoServer.addProvider( - lasso.PROVIDER_ROLE_SP, - os.path.join(dataDir, 'sp1-la/metadata.xml'), - os.path.join(dataDir, 'sp1-la/public-key.pem'), - os.path.join(dataDir, 'sp1-la/certificate.pem')) - defederation = lasso.Defederation(lassoServer) - # The processNotificationMsg should fail but not abort. - try: - defederation.processNotificationMsg('nonLibertyQuery=1') - except lasso.Error, error: - if error[0] != lasso.PROFILE_ERROR_INVALID_MSG: - raise - else: - self.fail('Defederation processNotificationMsg should have failed.') - - -class IdentityTestCase(unittest.TestCase): - def test01(self): - """Identity newFromDump & dump.""" - return - # test disabled since dump format changed - identityDump = """_CD739B41C602EAEA93626EBD1751CB46_11EA77A4FED32C41824AC5DE87298E65""" - identity = lasso.Identity.newFromDump(identityDump) - newIdentityDump = identity.dump() - self.failUnlessEqual(identityDump, newIdentityDump) - - -serverSuite = unittest.makeSuite(ServerTestCase, 'test') -loginSuite = unittest.makeSuite(LoginTestCase, 'test') -logoutSuite = unittest.makeSuite(LogoutTestCase, 'test') -defederationSuite = unittest.makeSuite(DefederationTestCase, 'test') -identitySuite = unittest.makeSuite(IdentityTestCase, 'test') - -allTests = unittest.TestSuite((serverSuite, loginSuite, logoutSuite, defederationSuite, - identitySuite)) - -if __name__ == '__main__': - sys.exit(not unittest.TextTestRunner(verbosity = 2).run(allTests).wasSuccessful()) - diff --git a/python/tests/tests.py b/python/tests/tests.py deleted file mode 100755 index d4bd42dd..00000000 --- a/python/tests/tests.py +++ /dev/null @@ -1,104 +0,0 @@ -#! /usr/bin/env python -# -*- coding: UTF-8 -*- -# -# $Id$ -# -# Python unit tests for Lasso library -# -# Copyright (C) 2004-2007 Entr'ouvert -# http://lasso.entrouvert.org -# -# Authors: See AUTHORS file in top-level directory. -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 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 General Public License for more details. -# -# You should have received a copy of the GNU 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 - - -import __builtin__ -import imp -from optparse import OptionParser -import os -import sys -import time -import unittest - -from XmlTestRunner import XmlTestRunner - -if not '..' in sys.path: - sys.path.insert(0, '..') -if not '../.libs' in sys.path: - sys.path.insert(0, '../.libs') - - -testSuites = [ - 'binding_tests', - 'profiles_tests', - 'errorchecking_tests', - ] - -import lasso -if lasso.WSF_SUPPORT: - testSuites.append('idwsf1_tests') - testSuites.append('idwsf2_tests') - - -# Parse command line options. -parser = OptionParser() -parser.add_option( - '-x', '--xml', dest = 'xmlMode', help = 'enable XML output', - action = 'store_true', default = False) -parser.add_option( - '-s', '--source-dir', dest = 'srcDir', help = 'path of source directory', - metavar = 'DIR', default = os.getcwd()) -(options, args) = parser.parse_args() -__builtin__.__dict__['dataDir'] = os.path.join(options.srcDir, '../../tests/data') - -if options.xmlMode: - print """""" - print """""" - print """ Python Bindings""" - print """ %s""" % time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) - -success = True -for testSuite in testSuites: - fp, pathname, description = imp.find_module(testSuite) - try: - module = imp.load_module(testSuite, fp, pathname, description) - finally: - if fp: - fp.close() - if not module: - print >> sys.stderr, 'Unable to load test suite:', testSuite - continue - - if module.__doc__: - doc = module.__doc__ - else: - doc = testSuite - - if options.xmlMode: - runner = XmlTestRunner() - else: - runner = unittest.TextTestRunner(verbosity=2) - print - print '-' * len(doc) - print doc - print '-' * len(doc) - result = runner.run(module.allTests) - success = success and result.wasSuccessful() - -if options.xmlMode: - print """""" - -sys.exit(not success) -- cgit