Updating tests

7 files changed, 131 insertions, 49 deletions

diff --git a/tests/test_20_assertion.py b/tests/test_20_assertion.py
index 2a25f048..c1c5e21c 100644
--- a/tests/test_20_assertion.py
+++ b/tests/test_20_assertion.py
@@ -93,8 +93,7 @@ def test_lifetime_2():
     
     assert r.get_lifetime("urn:mace:umu.se:saml:roland:sp") == {"minutes": 5}               
     assert r.get_lifetime("urn:mace:example.se:saml:sp") == {"hours": 1}        
-    
-    
+
     
 def test_ava_filter_1():
     conf = {
diff --git a/tests/test_21_attribute_converter.py b/tests/test_21_attribute_converter.py
index f262216e..7df0c4bd 100644
--- a/tests/test_21_attribute_converter.py
+++ b/tests/test_21_attribute_converter.py
@@ -20,7 +20,6 @@ class TestAC():
         assert len(self.acs) == 3
         assert _eq([a.name_format for a in self.acs],[BASIC_NF, URI_NF, SAML1] )
 
-        
     def test_ava_fro_1(self):
         ats = saml.attribute_statement_from_string(STATEMENT1)
         #print ats
@@ -58,7 +57,7 @@ class TestAC():
         
         statement = attribute_converter.from_local(self.acs, ava, BASIC_NF)
         
-        assert statement != None
+        assert statement is not None
         assert len(statement) == 2
         a0 = statement[0]
         a1 = statement[1]
@@ -140,3 +139,15 @@ class TestAC():
         lan = [attribute_converter.to_local_name(self.acs, a) for a in attr]
 
         assert _eq(lan, ['eduPersonPrimaryOrgUnitDN'])
+
+    def test_to_and_for(self):
+        ava = { "givenName": "Roland", "surname": "Hedberg" }
+
+        basic_ac = [a for a in self.acs if a.name_format == BASIC_NF][0]
+
+        attr_state = saml.AttributeStatement(basic_ac.to_(ava))
+
+        oava = basic_ac.fro(attr_state)
+
+        assert _eq(ava.keys(), oava.keys())
+        
\ No newline at end of file
diff --git a/tests/test_30_metadata.py b/tests/test_30_metadata.py
index 2226b9be..64185266 100644
--- a/tests/test_30_metadata.py
+++ b/tests/test_30_metadata.py
@@ -446,3 +446,20 @@ def test_ui_info():
     ui_info = loc[0][1][0]
     print ui_info
     assert ui_info.description[0].text == "Exempel bolag"
+
+def test_pdp():
+    md = metadata.MetaData(attrconv=ATTRCONV)
+    md.import_metadata(_fix_valid_until(_read_file("pdp_meta.xml")), "-")
+
+    assert md
+
+    pdps = md.pdp_services("http://www.example.org/pysaml2/")
+
+    assert len(pdps) == 1
+    pdp = pdps[0]
+    assert len(pdp.authz_service) == 1
+    assert pdp.authz_service[0].location == "http://www.example.org/pysaml2/authz"
+    assert pdp.authz_service[0].binding == BINDING_SOAP
+    endpoints = md.authz_service_endpoints("http://www.example.org/pysaml2/")
+    assert len(endpoints) == 1
+    assert endpoints[0] == "http://www.example.org/pysaml2/authz"
\ No newline at end of file
diff --git a/tests/test_31_config.py b/tests/test_31_config.py
index 209beba5..0578e15e 100644
--- a/tests/test_31_config.py
+++ b/tests/test_31_config.py
@@ -112,6 +112,33 @@ IDP2 = {
     #"xmlsec_binary" : "/usr/local/bin/xmlsec1",
 }
 
+PDP = {
+    "entityid" : "http://example.org/pysaml2/pdp",
+    "name" : "Rolands PdP",
+    "service": {
+        "pdp": {
+            "endpoints": {
+                "authz_service" : [("http://example.org/pysaml2/pdp/authz",
+                                   BINDING_SOAP)],
+            },
+        }
+    },
+    "key_file" : "test.key",
+    "cert_file" : "test.pem",
+    "organization": {
+        "name": "Exempel AB",
+        "display_name": [("Exempel AB","se"),("Example Co.","en")],
+        "url":"http://www.example.com/roland",
+    },
+    "contact_person": [{
+        "given_name":"John",
+        "sur_name": "Smith",
+        "email_address": ["john.smith@example.com"],
+        "contact_type": "technical",
+        },
+    ],
+}
+
 def _eq(l1,l2):
     return set(l1) == set(l2)
 
diff --git a/tests/test_43_soap.py b/tests/test_43_soap.py
index f11c9b1e..8ab8fd20 100755
--- a/tests/test_43_soap.py
+++ b/tests/test_43_soap.py
@@ -59,4 +59,3 @@ def test_make_soap_envelope():
     assert len(body) == 1
     saml_part = body[0]
     assert saml_part.tag == '{%s}AuthnRequest' % SAMLP_NAMESPACE
-    
\ No newline at end of file
diff --git a/tests/test_50_server.py b/tests/test_50_server.py
index f2e82a05..81891975 100644
--- a/tests/test_50_server.py
+++ b/tests/test_50_server.py
@@ -2,18 +2,15 @@
 # -*- coding: utf-8 -*-
 
 from saml2.server import Server, Identifier
-from saml2 import server, make_instance
 from saml2 import samlp, saml, client, config
 from saml2 import s_utils
 from saml2 import time_util
 from saml2.s_utils import OtherError
 from saml2.s_utils import do_attribute_statement, factory
 from saml2.soap import make_soap_enveloped_saml_thingy
-from saml2 import BINDING_HTTP_POST, BINDING_HTTP_REDIRECT, BINDING_SOAP
+from saml2 import BINDING_HTTP_POST, BINDING_HTTP_REDIRECT
 
 from py.test import raises
-import shelve
-import re
 import os
 
 def _eq(l1,l2):
@@ -22,7 +19,7 @@ def _eq(l1,l2):
 class TestIdentifier():
     def setup_class(self):
         self.ident = Identifier("foobar.db")
-    
+
     def test_persistent_nameid(self):
         sp_id = "urn:mace:umu.se:sp"
         nameid = self.ident.persistent_nameid(sp_id, "abcd0001")
@@ -31,8 +28,8 @@ class TestIdentifier():
         print self.ident.map
         local = self.ident.local_name(sp_id, remote_id)
         assert local == "abcd0001"
-        assert self.ident.local_name(sp_id, "pseudo random string") == None
-        assert self.ident.local_name(sp_id+":x", remote_id) == None
+        assert self.ident.local_name(sp_id, "pseudo random string") is None
+        assert self.ident.local_name(sp_id+":x", remote_id) is None
 
         # Always get the same
         nameid2 = self.ident.persistent_nameid(sp_id, "abcd0001")
@@ -46,21 +43,21 @@ class TestIdentifier():
         print self.ident.map
         local = self.ident.local_name(sp_id, remote_id)
         assert local == "abcd0001"
-        assert self.ident.local_name(sp_id, "pseudo random string") == None
-        assert self.ident.local_name(sp_id+":x", remote_id) == None
+        assert self.ident.local_name(sp_id, "pseudo random string") is None
+        assert self.ident.local_name(sp_id+":x", remote_id) is None
 
         # Getting a new, means really getting a new !
         nameid2 = self.ident.transient_nameid(sp_id, "abcd0001")
         assert nameid.text.strip() != nameid2.text.strip()
-        
+
     def teardown_class(self):
-	if os.path.exists("foobar.db"):
-	    os.unlink("foobar.db")
-        
+        if os.path.exists("foobar.db"):
+            os.unlink("foobar.db")
+
 class TestServer1():
     def setup_class(self):
         self.server = Server("idp_conf")
-        
+
         conf = config.SPConfig()
         conf.load_file("server_conf")
         self.client = client.Saml2Client(conf)
@@ -71,7 +68,7 @@ class TestServer1():
         assert _eq(issuer.keyswv(), ["text","format"])
         assert issuer.format == saml.NAMEID_FORMAT_ENTITY
         assert issuer.text == self.server.conf.entityid
-        
+
 
     def test_assertion(self):
         assertion = s_utils.assertion_factory(
@@ -84,7 +81,7 @@ class TestServer1():
                                 }),
             issuer=self.server.issuer(),
             )
-            
+
         assert _eq(assertion.keyswv(),['attribute_statement', 'issuer', 'id',
                                     'subject', 'issue_instant', 'version'])
         assert assertion.version == "2.0"
@@ -109,7 +106,7 @@ class TestServer1():
         assert _eq(subject.keyswv(),["text", "name_id"])
         assert subject.text == "_aaa"
         assert subject.name_id.format == saml.NAMEID_FORMAT_TRANSIENT
-        
+
     def test_response(self):
         response = s_utils.response_factory(
                 in_response_to="_012345",
@@ -126,7 +123,7 @@ class TestServer1():
                 ),
                 issuer=self.server.issuer(),
             )
-            
+
         print response.keyswv()
         assert _eq(response.keyswv(),['destination', 'assertion','status', 
                                     'in_response_to', 'issue_instant', 
@@ -148,11 +145,11 @@ class TestServer1():
                             spentityid = "urn:mace:example.com:saml:roland:sp",
                             my_name = "My real name",
                         )
-                        
+
         intermed = s_utils.deflate_and_base64_encode(authn_request)
         # should raise an error because faulty spentityid
         raises(OtherError, self.server.parse_authn_request, intermed)
-        
+
     def test_parse_faulty_request_to_err_status(self):
         authn_request = self.client.authn_request(
                             query_id = "id1",
@@ -161,7 +158,7 @@ class TestServer1():
                             spentityid = "urn:mace:example.com:saml:roland:sp",
                             my_name = "My real name",
                         )
-                        
+
         intermed = s_utils.deflate_and_base64_encode(authn_request)
         try:
             self.server.parse_authn_request(intermed)
@@ -169,7 +166,7 @@ class TestServer1():
         except OtherError, oe:
             print oe.args
             status = s_utils.error_status_factory(oe)
-            
+
         assert status
         print status
         assert _eq(status.keyswv(), ["status_code", "status_message"])
@@ -187,7 +184,7 @@ class TestServer1():
                             spentityid = "urn:mace:example.com:saml:roland:sp",
                             my_name = "My real name",
                         )
-                        
+
         print authn_request
         intermed = s_utils.deflate_and_base64_encode(authn_request)
         response = self.server.parse_authn_request(intermed)
@@ -211,7 +208,7 @@ class TestServer1():
                     { "eduPersonEntitlement": "Short stop"}, # identity
                     name_id
                 )
-                
+
         print resp.keyswv()
         assert _eq(resp.keyswv(),['status', 'destination', 'assertion', 
                                     'in_response_to', 'issue_instant', 
@@ -252,7 +249,7 @@ class TestServer1():
                     "http://localhost:8087/",           # consumer_url
                     "urn:mace:example.com:saml:roland:sp", # sp_entity_id
                 )
-                
+
         print resp.keyswv()
         assert _eq(resp.keyswv(),['status', 'destination', 'in_response_to', 
                                   'issue_instant', 'version', 'id', 'issuer'])
@@ -267,7 +264,7 @@ class TestServer1():
         exc = s_utils.MissingValue("eduPersonAffiliation missing")
         resp = self.server.error_response("id12", "http://localhost:8087/", 
                         "urn:mace:example.com:saml:roland:sp", exc )
-                
+
         print resp.keyswv()
         assert _eq(resp.keyswv(),['status', 'destination', 'in_response_to', 
                                   'issue_instant', 'version', 'id', 'issuer'])
@@ -293,7 +290,7 @@ class TestServer1():
                     samlp.NameIDPolicy(format=saml.NAMEID_FORMAT_TRANSIENT,
                                         allow_create="true"),
                     "foba0001@example.com")
-                   
+
         response = samlp.response_from_string("\n".join(resp_str))
         print response.keyswv()
         assert _eq(response.keyswv(),['status', 'destination', 'assertion', 
@@ -309,12 +306,12 @@ class TestServer1():
         astate = assertion.attribute_statement[0]
         print astate
         assert len(astate.attribute) == 3
-        
+
     def test_signed_response(self):
         name_id = self.server.ident.transient_nameid(
                                         "urn:mace:example.com:saml:roland:sp",
                                         "id12")
-                
+
         signed_resp = self.server.do_response(
                     "id12",                                 # in_response_to
                     "http://lingon.catalogix.se:8087/",     # consumer_url
@@ -326,7 +323,7 @@ class TestServer1():
 
         print "%s" % signed_resp
         assert signed_resp
-        
+
         # It's the assertions that are signed not the response per se
         assert len(signed_resp.assertion) == 1
         assertion = signed_resp.assertion[0]
@@ -347,7 +344,7 @@ class TestServer1():
             }
         }
         self.client.users.add_information_about_person(sinfo)
-        
+
         logout_request = self.client.construct_logout_request(
                     subject_id="foba0001",
                     destination = "http://localhost:8088/slo",
@@ -355,11 +352,11 @@ class TestServer1():
                     reason = "I'm tired of this")
 
         intermed = s_utils.deflate_and_base64_encode("%s" % (logout_request,))
-                        
+
         #saml_soap = make_soap_enveloped_saml_thingy(logout_request)
         request = self.server.parse_logout_request(intermed, BINDING_HTTP_POST)
         assert request
-        
+
     def test_slo_soap(self):
         soon = time_util.in_a_while(days=1)
         sinfo = {
@@ -374,19 +371,19 @@ class TestServer1():
 
         sp = client.Saml2Client(config_file="server_conf")
         sp.users.add_information_about_person(sinfo)
-        
+
         logout_request = sp.construct_logout_request(subject_id = "foba0001",
                     destination = "http://localhost:8088/slo",
                     issuer_entity_id = "urn:mace:example.com:saml:roland:idp",
                     reason = "I'm tired of this")
 
         intermed = s_utils.deflate_and_base64_encode("%s" % (logout_request,))
-                        
+
         saml_soap = make_soap_enveloped_saml_thingy(logout_request)
         idp = Server("idp_soap_conf")
         request = idp.parse_logout_request(saml_soap)
         assert request
-        
+
 #------------------------------------------------------------------------
 
 IDENTITY = {"eduPersonAffiliation": ["staff", "member"],
@@ -403,7 +400,7 @@ class TestServer2():
         response = self.server.do_aa_response("aaa", "http://example.com/sp/", 
                         "urn:mace:example.com:sp:1", IDENTITY.copy())
 
-        assert response != None
+        assert response is not None
         assert response.destination == "http://example.com/sp/"
         assert response.in_response_to == "aaa"
         assert response.version == "2.0"
@@ -434,15 +431,15 @@ def _logout_request(conf_file):
         }
     }
     sp.users.add_information_about_person(sinfo)
-    
+
     return sp.construct_logout_request(
                 subject_id = "foba0001",
                 destination = "http://localhost:8088/slo",
                 issuer_entity_id = "urn:mace:example.com:saml:roland:idp",
                 reason = "I'm tired of this")
-    
+
 class TestServerLogout():
-    
+
     def test_1(self):
         server = Server("idp_slo_redirect_conf")
         request = _logout_request("sp_slo_redirect_conf")
@@ -453,7 +450,7 @@ class TestServerLogout():
         assert len(headers) == 1
         assert headers[0][0] == "Location"
         assert message == ['']
-        
+
 # class TestSign():
 #     def test_1(self):
 #         IDP = server.Server("restrictive_idp.config", debug=1)
diff --git a/tests/test_51_client.py b/tests/test_51_client.py
index 1fc38045..52ea71e8 100644
--- a/tests/test_51_client.py
+++ b/tests/test_51_client.py
@@ -8,15 +8,14 @@ from saml2.client import Saml2Client, LogoutError
 from saml2 import samlp, BINDING_HTTP_POST
 from saml2 import BINDING_SOAP
 from saml2 import saml, config, class_name
-#from saml2.sigver import correctly_signed_authn_request
 from saml2.server import Server
 from saml2.s_utils import decode_base64_and_inflate
 from saml2.time_util import in_a_while
-from saml2.sigver import xmlsec_version
+from saml2.assertion import Assertion
+from saml2.assertion import Policy
 
 from py.test import raises
 
-
 def for_me(condition, me ):
     for restriction in condition.audience_restriction:
         audience = restriction.audience
@@ -480,3 +479,36 @@ class TestClient:
         assert state_info["operation"] == "SLO"
         assert state_info["entity_ids"] == entity_ids
         assert state_info["sign"] == False
+
+    def test_authz_decision_query(self):
+        conf = config.SPConfig()
+        conf.load_file("server3_conf")
+        client = Saml2Client(conf)
+
+        AVA = {'mail': u'roland.hedberg@adm.umu.se',
+               'eduPersonTargetedID': '95e9ae91dbe62d35198fbbd5e1fb0976',
+               'displayName': u'Roland Hedberg',
+               'uid': 'http://roland.hedberg.myopenid.com/'}
+
+        sp_entity_id = "sp_entity_id"
+        in_response_to = "1234"
+        consumer_url = "http://example.com/consumer"
+        name_id = saml.NameID(saml.NAMEID_FORMAT_TRANSIENT, text="name_id")
+        policy = Policy(None)
+        ava = Assertion(AVA)
+        assertion = ava.construct(sp_entity_id, in_response_to,
+                                    consumer_url, name_id,
+                                    conf.attribute_converters,
+                                    policy, issuer=client.issuer())
+
+        adq = client.authz_decision_query_using_assertion("entity_id",
+                                                         assertion,
+                                                        "read",
+                                                        "http://example.com/text")
+
+        assert adq
+        print adq
+        assert adq.keyswv() != []
+        assert adq.destination == "entity_id"
+        assert adq.resource == "http://example.com/text"
+        assert adq.action[0].text == "read"