# -*- coding: utf-8 -*- from openid import message from openid import oidutil from openid.extensions import sreg import urllib.request import urllib.parse import urllib.error import unittest def mkGetArgTest(ns, key, expected=None): def test(self): a_default = object() self.assertEqual(self.msg.getArg(ns, key), expected) if expected is None: self.assertEqual( self.msg.getArg(ns, key, a_default), a_default) self.assertRaises( KeyError, self.msg.getArg, ns, key, message.no_default) else: self.assertEqual( self.msg.getArg(ns, key, a_default), expected) self.assertEqual( self.msg.getArg(ns, key, message.no_default), expected) return test class EmptyMessageTest(unittest.TestCase): def setUp(self): self.msg = message.Message() def test_toPostArgs(self): self.assertEqual(self.msg.toPostArgs(), {}) def test_toArgs(self): self.assertEqual(self.msg.toArgs(), {}) def test_toKVForm(self): self.assertEqual(self.msg.toKVForm(), b'') def test_toURLEncoded(self): self.assertEqual(self.msg.toURLEncoded(), '') def test_toURL(self): base_url = 'http://base.url/' self.assertEqual(self.msg.toURL(base_url), base_url) def test_getOpenID(self): self.assertEqual(self.msg.getOpenIDNamespace(), None) def test_getKeyOpenID(self): # Could reasonably return None instead of raising an # exception. I'm not sure which one is more right, since this # case should only happen when you're building a message from # scratch and so have no default namespace. self.assertRaises(message.UndefinedOpenIDNamespace, self.msg.getKey, message.OPENID_NS, 'foo') def test_getKeyBARE(self): self.assertEqual(self.msg.getKey(message.BARE_NS, 'foo'), 'foo') def test_getKeyNS1(self): self.assertEqual(self.msg.getKey(message.OPENID1_NS, 'foo'), None) def test_getKeyNS2(self): self.assertEqual(self.msg.getKey(message.OPENID2_NS, 'foo'), None) def test_getKeyNS3(self): self.assertEqual(self.msg.getKey('urn:nothing-significant', 'foo'), None) def test_hasKey(self): # Could reasonably return False instead of raising an # exception. I'm not sure which one is more right, since this # case should only happen when you're building a message from # scratch and so have no default namespace. self.assertRaises(message.UndefinedOpenIDNamespace, self.msg.hasKey, message.OPENID_NS, 'foo') def test_hasKeyBARE(self): self.assertEqual(self.msg.hasKey(message.BARE_NS, 'foo'), False) def test_hasKeyNS1(self): self.assertEqual(self.msg.hasKey(message.OPENID1_NS, 'foo'), False) def test_hasKeyNS2(self): self.assertEqual(self.msg.hasKey(message.OPENID2_NS, 'foo'), False) def test_hasKeyNS3(self): self.assertEqual(self.msg.hasKey('urn:nothing-significant', 'foo'), False) def test_getAliasedArgSuccess(self): msg = message.Message.fromPostArgs({'openid.ns.test': 'urn://foo', 'openid.test.flub': 'bogus'}) actual_uri = msg.getAliasedArg('ns.test', message.no_default) self.assertEqual("urn://foo", actual_uri) def test_getAliasedArgFailure(self): msg = message.Message.fromPostArgs({'openid.test.flub': 'bogus'}) self.assertRaises(KeyError, msg.getAliasedArg, 'ns.test', message.no_default) def test_getArg(self): # Could reasonably return None instead of raising an # exception. I'm not sure which one is more right, since this # case should only happen when you're building a message from # scratch and so have no default namespace. self.assertRaises(message.UndefinedOpenIDNamespace, self.msg.getArg, message.OPENID_NS, 'foo') test_getArgBARE = mkGetArgTest(message.BARE_NS, 'foo') test_getArgNS1 = mkGetArgTest(message.OPENID1_NS, 'foo') test_getArgNS2 = mkGetArgTest(message.OPENID2_NS, 'foo') test_getArgNS3 = mkGetArgTest('urn:nothing-significant', 'foo') def test_getArgs(self): # Could reasonably return {} instead of raising an # exception. I'm not sure which one is more right, since this # case should only happen when you're building a message from # scratch and so have no default namespace. self.assertRaises(message.UndefinedOpenIDNamespace, self.msg.getArgs, message.OPENID_NS) def test_getArgsBARE(self): self.assertEqual(self.msg.getArgs(message.BARE_NS), {}) def test_getArgsNS1(self): self.assertEqual(self.msg.getArgs(message.OPENID1_NS), {}) def test_getArgsNS2(self): self.assertEqual(self.msg.getArgs(message.OPENID2_NS), {}) def test_getArgsNS3(self): self.assertEqual(self.msg.getArgs('urn:nothing-significant'), {}) def test_updateArgs(self): self.assertRaises(message.UndefinedOpenIDNamespace, self.msg.updateArgs, message.OPENID_NS, {'does not': 'matter'}) def _test_updateArgsNS(self, ns): update_args = { 'Camper van Beethoven': 'David Lowery', 'Magnolia Electric Co.': 'Jason Molina', } self.assertEqual(self.msg.getArgs(ns), {}) self.msg.updateArgs(ns, update_args) self.assertEqual(self.msg.getArgs(ns), update_args) def test_updateArgsBARE(self): self._test_updateArgsNS(message.BARE_NS) def test_updateArgsNS1(self): self._test_updateArgsNS(message.OPENID1_NS) def test_updateArgsNS2(self): self._test_updateArgsNS(message.OPENID2_NS) def test_updateArgsNS3(self): self._test_updateArgsNS('urn:nothing-significant') def test_setArg(self): self.assertRaises(message.UndefinedOpenIDNamespace, self.msg.setArg, message.OPENID_NS, 'does not', 'matter') def _test_setArgNS(self, ns): key = 'Camper van Beethoven' value = 'David Lowery' self.assertEqual(self.msg.getArg(ns, key), None) self.msg.setArg(ns, key, value) self.assertEqual(self.msg.getArg(ns, key), value) def test_setArgBARE(self): self._test_setArgNS(message.BARE_NS) def test_setArgNS1(self): self._test_setArgNS(message.OPENID1_NS) def test_setArgNS2(self): self._test_setArgNS(message.OPENID2_NS) def test_setArgNS3(self): self._test_setArgNS('urn:nothing-significant') def test_setArgToNone(self): self.assertRaises(AssertionError, self.msg.setArg, message.OPENID1_NS, 'op_endpoint', None) def test_delArg(self): # Could reasonably raise KeyError instead of raising # UndefinedOpenIDNamespace. I'm not sure which one is more # right, since this case should only happen when you're # building a message from scratch and so have no default # namespace. self.assertRaises(message.UndefinedOpenIDNamespace, self.msg.delArg, message.OPENID_NS, 'key') def _test_delArgNS(self, ns): key = 'Camper van Beethoven' self.assertRaises(KeyError, self.msg.delArg, ns, key) def test_delArgBARE(self): self._test_delArgNS(message.BARE_NS) def test_delArgNS1(self): self._test_delArgNS(message.OPENID1_NS) def test_delArgNS2(self): self._test_delArgNS(message.OPENID2_NS) def test_delArgNS3(self): self._test_delArgNS('urn:nothing-significant') def test_isOpenID1(self): self.assertFalse(self.msg.isOpenID1()) def test_isOpenID2(self): self.assertFalse(self.msg.isOpenID2()) class OpenID1MessageTest(unittest.TestCase): def setUp(self): self.msg = message.Message.fromPostArgs({ 'openid.mode': 'error', 'openid.error': 'unit test' }) def test_toPostArgs(self): self.assertEqual(self.msg.toPostArgs(), { 'openid.mode': 'error', 'openid.error': 'unit test' }) def test_toArgs(self): self.assertEqual(self.msg.toArgs(), { 'mode': 'error', 'error': 'unit test' }) def test_toKVForm(self): self.assertEqual(self.msg.toKVForm(), b'error:unit test\nmode:error\n') def test_toURLEncoded(self): self.assertEqual(self.msg.toURLEncoded(), 'openid.error=unit+test&openid.mode=error') def test_toURL(self): base_url = 'http://base.url/' actual = self.msg.toURL(base_url) actual_base = actual[:len(base_url)] self.assertEqual(actual_base, base_url) self.assertEqual(actual[len(base_url)], '?') query = actual[len(base_url) + 1:] parsed = urllib.parse.parse_qs(query) self.assertEqual(parsed, { 'openid.mode': ['error'], 'openid.error': ['unit test'] }) def test_getOpenID(self): self.assertEqual(self.msg.getOpenIDNamespace(), message.OPENID1_NS) def test_getKeyOpenID(self): self.assertEqual(self.msg.getKey(message.OPENID_NS, 'mode'), 'openid.mode') def test_getKeyBARE(self): self.assertEqual(self.msg.getKey(message.BARE_NS, 'mode'), 'mode') def test_getKeyNS1(self): self.assertEqual( self.msg.getKey(message.OPENID1_NS, 'mode'), 'openid.mode') def test_getKeyNS2(self): self.assertEqual(self.msg.getKey(message.OPENID2_NS, 'mode'), None) def test_getKeyNS3(self): self.assertEqual( self.msg.getKey('urn:nothing-significant', 'mode'), None) def test_hasKey(self): self.assertEqual(self.msg.hasKey(message.OPENID_NS, 'mode'), True) def test_hasKeyBARE(self): self.assertEqual(self.msg.hasKey(message.BARE_NS, 'mode'), False) def test_hasKeyNS1(self): self.assertEqual(self.msg.hasKey(message.OPENID1_NS, 'mode'), True) def test_hasKeyNS2(self): self.assertEqual( self.msg.hasKey(message.OPENID2_NS, 'mode'), False) def test_hasKeyNS3(self): self.assertEqual( self.msg.hasKey('urn:nothing-significant', 'mode'), False) test_getArgNSBARE = mkGetArgTest(message.BARE_NS, 'mode') test_getArgNS = mkGetArgTest(message.OPENID_NS, 'mode', 'error') test_getArgNS1 = mkGetArgTest(message.OPENID1_NS, 'mode', 'error') test_getArgNS2 = mkGetArgTest(message.OPENID2_NS, 'mode') test_getArgNS3 = mkGetArgTest('urn:nothing-significant', 'mode') def test_getArgs(self): self.assertEqual(self.msg.getArgs(message.OPENID_NS), { 'mode': 'error', 'error': 'unit test', }) def test_getArgsBARE(self): self.assertEqual(self.msg.getArgs(message.BARE_NS), {}) def test_getArgsNS1(self): self.assertEqual(self.msg.getArgs(message.OPENID1_NS), { 'mode': 'error', 'error': 'unit test', }) def test_getArgsNS2(self): self.assertEqual(self.msg.getArgs(message.OPENID2_NS), {}) def test_getArgsNS3(self): self.assertEqual(self.msg.getArgs('urn:nothing-significant'), {}) def _test_updateArgsNS(self, ns, before=None): if before is None: before = {} update_args = { 'Camper van Beethoven': 'David Lowery', 'Magnolia Electric Co.': 'Jason Molina', } self.assertEqual(self.msg.getArgs(ns), before) self.msg.updateArgs(ns, update_args) after = dict(before) after.update(update_args) self.assertEqual(self.msg.getArgs(ns), after) def test_updateArgs(self): self._test_updateArgsNS(message.OPENID_NS, before={'mode': 'error', 'error': 'unit test'}) def test_updateArgsBARE(self): self._test_updateArgsNS(message.BARE_NS) def test_updateArgsNS1(self): self._test_updateArgsNS(message.OPENID1_NS, before={'mode': 'error', 'error': 'unit test'}) def test_updateArgsNS2(self): self._test_updateArgsNS(message.OPENID2_NS) def test_updateArgsNS3(self): self._test_updateArgsNS('urn:nothing-significant') def _test_setArgNS(self, ns): key = 'Camper van Beethoven' value = 'David Lowery' self.assertEqual(self.msg.getArg(ns, key), None) self.msg.setArg(ns, key, value) self.assertEqual(self.msg.getArg(ns, key), value) def test_setArg(self): self._test_setArgNS(message.OPENID_NS) def test_setArgBARE(self): self._test_setArgNS(message.BARE_NS) def test_setArgNS1(self): self._test_setArgNS(message.OPENID1_NS) def test_setArgNS2(self): self._test_setArgNS(message.OPENID2_NS) def test_setArgNS3(self): self._test_setArgNS('urn:nothing-significant') def _test_delArgNS(self, ns): key = 'Camper van Beethoven' value = 'David Lowery' self.assertRaises(KeyError, self.msg.delArg, ns, key) self.msg.setArg(ns, key, value) self.assertEqual(self.msg.getArg(ns, key), value) self.msg.delArg(ns, key) self.assertEqual(self.msg.getArg(ns, key), None) def test_delArg(self): self._test_delArgNS(message.OPENID_NS) def test_delArgBARE(self): self._test_delArgNS(message.BARE_NS) def test_delArgNS1(self): self._test_delArgNS(message.OPENID1_NS) def test_delArgNS2(self): self._test_delArgNS(message.OPENID2_NS) def test_delArgNS3(self): self._test_delArgNS('urn:nothing-significant') def test_isOpenID1(self): self.assertTrue(self.msg.isOpenID1()) def test_isOpenID2(self): self.assertFalse(self.msg.isOpenID2()) class OpenID1ExplicitMessageTest(unittest.TestCase): def setUp(self): self.msg = message.Message.fromPostArgs({ 'openid.mode': 'error', 'openid.error': 'unit test', 'openid.ns': message.OPENID1_NS }) def test_toPostArgs(self): self.assertEqual(self.msg.toPostArgs(), { 'openid.mode': 'error', 'openid.error': 'unit test', 'openid.ns': message.OPENID1_NS, }) def test_toArgs(self): self.assertEqual(self.msg.toArgs(), { 'mode': 'error', 'error': 'unit test', 'ns': message.OPENID1_NS, }) def test_toKVForm(self): self.assertEqual( self.msg.toKVForm(), bytes('error:unit test\nmode:error\nns:%s\n' % message.OPENID1_NS, encoding="utf-8")) def test_toURLEncoded(self): self.assertEqual( self.msg.toURLEncoded(), 'openid.error=unit+test&openid.mode=error&openid.ns=http%3A%2F%2Fopenid.net%2Fsignon%2F1.0') def test_toURL(self): base_url = 'http://base.url/' actual = self.msg.toURL(base_url) actual_base = actual[:len(base_url)] self.assertEqual(actual_base, base_url) self.assertEqual(actual[len(base_url)], '?') query = actual[len(base_url) + 1:] parsed = urllib.parse.parse_qs(query) self.assertEqual(parsed, { 'openid.mode': ['error'], 'openid.error': ['unit test'], 'openid.ns': [message.OPENID1_NS] }) def test_isOpenID1(self): self.assertTrue(self.msg.isOpenID1()) class OpenID2MessageTest(unittest.TestCase): def setUp(self): self.msg = message.Message.fromPostArgs({ 'openid.mode': 'error', 'openid.error': 'unit test', 'openid.ns': message.OPENID2_NS }) self.msg.setArg(message.BARE_NS, "xey", "value") def test_toPostArgs(self): self.assertEqual( self.msg.toPostArgs(), { 'openid.mode': 'error', 'openid.error': 'unit test', 'openid.ns': message.OPENID2_NS, 'xey': 'value', }) def test_toPostArgs_bug_with_utf8_encoded_values(self): msg = message.Message.fromPostArgs({'openid.mode': 'error', 'openid.error': 'unit test', 'openid.ns': message.OPENID2_NS }) msg.setArg(message.BARE_NS, 'ünicöde_key', 'ünicöde_välüe') self.assertEqual(msg.toPostArgs(), {'openid.mode': 'error', 'openid.error': 'unit test', 'openid.ns': message.OPENID2_NS, 'ünicöde_key': 'ünicöde_välüe', }) def test_toArgs(self): # This method can't tolerate BARE_NS. self.msg.delArg(message.BARE_NS, "xey") self.assertEqual(self.msg.toArgs(), { 'mode': 'error', 'error': 'unit test', 'ns': message.OPENID2_NS, }) def test_toKVForm(self): # Can't tolerate BARE_NS in kvform self.msg.delArg(message.BARE_NS, "xey") self.assertEqual( self.msg.toKVForm(), bytes('error:unit test\nmode:error\nns:%s\n' % message.OPENID2_NS, encoding="utf-8")) def _test_urlencoded(self, s): expected = ('openid.error=unit+test&openid.mode=error&' 'openid.ns=%s&xey=value' % ( urllib.parse.quote(message.OPENID2_NS, ''),)) self.assertEqual(s, expected) def test_toURLEncoded(self): self._test_urlencoded(self.msg.toURLEncoded()) def test_toURL(self): base_url = 'http://base.url/' actual = self.msg.toURL(base_url) actual_base = actual[:len(base_url)] self.assertEqual(actual_base, base_url) self.assertEqual(actual[len(base_url)], '?') query = actual[len(base_url) + 1:] self._test_urlencoded(query) def test_getOpenID(self): self.assertEqual(self.msg.getOpenIDNamespace(), message.OPENID2_NS) def test_getKeyOpenID(self): self.assertEqual(self.msg.getKey(message.OPENID_NS, 'mode'), 'openid.mode') def test_getKeyBARE(self): self.assertEqual(self.msg.getKey(message.BARE_NS, 'mode'), 'mode') def test_getKeyNS1(self): self.assertEqual( self.msg.getKey(message.OPENID1_NS, 'mode'), None) def test_getKeyNS2(self): self.assertEqual( self.msg.getKey(message.OPENID2_NS, 'mode'), 'openid.mode') def test_getKeyNS3(self): self.assertEqual( self.msg.getKey('urn:nothing-significant', 'mode'), None) def test_hasKeyOpenID(self): self.assertEqual(self.msg.hasKey(message.OPENID_NS, 'mode'), True) def test_hasKeyBARE(self): self.assertEqual(self.msg.hasKey(message.BARE_NS, 'mode'), False) def test_hasKeyNS1(self): self.assertEqual( self.msg.hasKey(message.OPENID1_NS, 'mode'), False) def test_hasKeyNS2(self): self.assertEqual( self.msg.hasKey(message.OPENID2_NS, 'mode'), True) def test_hasKeyNS3(self): self.assertEqual( self.msg.hasKey('urn:nothing-significant', 'mode'), False) test_getArgBARE = mkGetArgTest(message.BARE_NS, 'mode') test_getArgNS = mkGetArgTest(message.OPENID_NS, 'mode', 'error') test_getArgNS1 = mkGetArgTest(message.OPENID1_NS, 'mode') test_getArgNS2 = mkGetArgTest(message.OPENID2_NS, 'mode', 'error') test_getArgNS3 = mkGetArgTest('urn:nothing-significant', 'mode') def test_getArgsOpenID(self): self.assertEqual(self.msg.getArgs(message.OPENID_NS), { 'mode': 'error', 'error': 'unit test', }) def test_getArgsBARE(self): self.assertEqual(self.msg.getArgs(message.BARE_NS), {'xey': 'value'}) def test_getArgsNS1(self): self.assertEqual(self.msg.getArgs(message.OPENID1_NS), {}) def test_getArgsNS2(self): self.assertEqual(self.msg.getArgs(message.OPENID2_NS), { 'mode': 'error', 'error': 'unit test', }) def test_getArgsNS3(self): self.assertEqual(self.msg.getArgs('urn:nothing-significant'), {}) def _test_updateArgsNS(self, ns, before=None): if before is None: before = {} update_args = { 'Camper van Beethoven': 'David Lowery', 'Magnolia Electric Co.': 'Jason Molina', } self.assertEqual(self.msg.getArgs(ns), before) self.msg.updateArgs(ns, update_args) after = dict(before) after.update(update_args) self.assertEqual(self.msg.getArgs(ns), after) def test_updateArgsOpenID(self): self._test_updateArgsNS(message.OPENID_NS, before={'mode': 'error', 'error': 'unit test'}) def test_updateArgsBARE(self): self._test_updateArgsNS(message.BARE_NS, before={'xey': 'value'}) def test_updateArgsNS1(self): self._test_updateArgsNS(message.OPENID1_NS) def test_updateArgsNS2(self): self._test_updateArgsNS(message.OPENID2_NS, before={'mode': 'error', 'error': 'unit test'}) def test_updateArgsNS3(self): self._test_updateArgsNS('urn:nothing-significant') def _test_setArgNS(self, ns): key = 'Camper van Beethoven' value = 'David Lowery' self.assertEqual(self.msg.getArg(ns, key), None) self.msg.setArg(ns, key, value) self.assertEqual(self.msg.getArg(ns, key), value) def test_setArgOpenID(self): self._test_setArgNS(message.OPENID_NS) def test_setArgBARE(self): self._test_setArgNS(message.BARE_NS) def test_setArgNS1(self): self._test_setArgNS(message.OPENID1_NS) def test_setArgNS2(self): self._test_setArgNS(message.OPENID2_NS) def test_setArgNS3(self): self._test_setArgNS('urn:nothing-significant') def test_badAlias(self): """Make sure dotted aliases and OpenID protocol fields are not allowed as namespace aliases.""" for f in message.OPENID_PROTOCOL_FIELDS + ['dotted.alias']: args = {'openid.ns.%s' % f: 'blah', 'openid.%s.foo' % f: 'test'} # .fromPostArgs covers .fromPostArgs, .fromOpenIDArgs, # ._fromOpenIDArgs, and .fromOpenIDArgs (since it calls # .fromPostArgs). self.assertRaises(AssertionError, self.msg.fromPostArgs, args) def test_mysterious_missing_namespace_bug(self): """A failing test for bug #112""" openid_args = { 'assoc_handle': '{{HMAC-SHA256}{1211477242.29743}{v5cadg==}', 'claimed_id': 'http://nerdbank.org/OPAffirmative/AffirmativeIdentityWithSregNoAssoc.aspx', 'ns.sreg': 'http://openid.net/extensions/sreg/1.1', 'response_nonce': '2008-05-22T17:27:22ZUoW5.\\NV', 'signed': 'return_to,identity,claimed_id,op_endpoint,response_nonce,ns.sreg,sreg.email,sreg.nickname,assoc_handle', 'sig': 'e3eGZ10+TNRZitgq5kQlk5KmTKzFaCRI8OrRoXyoFa4=', 'mode': 'check_authentication', 'op_endpoint': 'http://nerdbank.org/OPAffirmative/ProviderNoAssoc.aspx', 'sreg.nickname': 'Andy', 'return_to': 'http://localhost.localdomain:8001/process?janrain_nonce=2008-05-22T17%3A27%3A21ZnxHULd', 'invalidate_handle': '{{HMAC-SHA1}{1211477241.92242}{H0akXw==}', 'identity': 'http://nerdbank.org/OPAffirmative/AffirmativeIdentityWithSregNoAssoc.aspx', 'sreg.email': 'a@b.com' } m = message.Message.fromOpenIDArgs(openid_args) self.assertTrue( ('http://openid.net/extensions/sreg/1.1', 'sreg') in m.namespaces.items()) missing = [] if isinstance(openid_args['signed'], bytes): oid_args_signed = openid_args['signed'].decode("utf-8") else: oid_args_signed = openid_args['signed'] for k in oid_args_signed.split(','): if ("openid." + k) not in m.toPostArgs(): missing.append(k) self.assertEqual([], missing, missing) self.assertEqual(openid_args, m.toArgs()) self.assertTrue(m.isOpenID1()) def test_112B(self): args = { 'openid.assoc_handle': 'fa1f5ff0-cde4-11dc-a183-3714bfd55ca8', 'openid.claimed_id': 'http://binkley.lan/user/test01', 'openid.identity': 'http://test01.binkley.lan/', 'openid.mode': 'id_res', 'openid.ns': 'http://specs.openid.net/auth/2.0', 'openid.ns.pape': 'http://specs.openid.net/extensions/pape/1.0', 'openid.op_endpoint': 'http://binkley.lan/server', 'openid.pape.auth_policies': 'none', 'openid.pape.auth_time': '2008-01-28T20:42:36Z', 'openid.pape.nist_auth_level': '0', 'openid.response_nonce': '2008-01-28T21:07:04Z99Q=', 'openid.return_to': 'http://binkley.lan:8001/process?janrain_nonce=2008-01-28T21%3A07%3A02Z0tMIKx', 'openid.sig': 'YJlWH4U6SroB1HoPkmEKx9AyGGg=', 'openid.signed': 'assoc_handle,identity,response_nonce,return_to,claimed_id,op_endpoint,pape.auth_time,ns.pape,pape.nist_auth_level,pape.auth_policies' } m = message.Message.fromPostArgs(args) missing = [] if isinstance(args['openid.signed'], bytes): oid_args_signed = args['openid.signed'].decode("utf-8") else: oid_args_signed = args['openid.signed'] for k in oid_args_signed.split(','): if ("openid." + k) not in m.toPostArgs(): missing.append(k) self.assertEqual([], missing, missing) self.assertEqual(args, m.toPostArgs()) self.assertTrue(m.isOpenID2()) def test_implicit_sreg_ns(self): openid_args = { 'sreg.email': 'a@b.com' } m = message.Message.fromOpenIDArgs(openid_args) self.assertTrue((sreg.ns_uri, 'sreg') in list(m.namespaces.items())) self.assertEqual('a@b.com', m.getArg(sreg.ns_uri, 'email')) self.assertEqual(openid_args, m.toArgs()) self.assertTrue(m.isOpenID1()) def _test_delArgNS(self, ns): key = 'Camper van Beethoven' value = 'David Lowery' self.assertRaises(KeyError, self.msg.delArg, ns, key) self.msg.setArg(ns, key, value) self.assertEqual(self.msg.getArg(ns, key), value) self.msg.delArg(ns, key) self.assertEqual(self.msg.getArg(ns, key), None) def test_delArgOpenID(self): self._test_delArgNS(message.OPENID_NS) def test_delArgBARE(self): self._test_delArgNS(message.BARE_NS) def test_delArgNS1(self): self._test_delArgNS(message.OPENID1_NS) def test_delArgNS2(self): self._test_delArgNS(message.OPENID2_NS) def test_delArgNS3(self): self._test_delArgNS('urn:nothing-significant') def test_overwriteExtensionArg(self): ns = 'urn:unittest_extension' key = 'mykey' value_1 = 'value_1' value_2 = 'value_2' self.msg.setArg(ns, key, value_1) self.assertTrue(self.msg.getArg(ns, key) == value_1) self.msg.setArg(ns, key, value_2) self.assertTrue(self.msg.getArg(ns, key) == value_2) def test_argList(self): self.assertRaises(TypeError, self.msg.fromPostArgs, {'arg': [1, 2, 3]}) def test_isOpenID1(self): self.assertFalse(self.msg.isOpenID1()) def test_isOpenID2(self): self.assertTrue(self.msg.isOpenID2()) class MessageTest(unittest.TestCase): def setUp(self): self.postargs = { 'openid.ns': message.OPENID2_NS, 'openid.mode': 'checkid_setup', 'openid.identity': 'http://bogus.example.invalid:port/', 'openid.assoc_handle': 'FLUB', 'openid.return_to': 'Neverland', } self.action_url = 'scheme://host:port/path?query' self.form_tag_attrs = { 'company': 'janrain', 'class': 'fancyCSS', } self.submit_text = 'GO!' ### Expected data regardless of input self.required_form_attrs = { 'accept-charset': 'UTF-8', 'enctype': 'application/x-www-form-urlencoded', 'method': 'post', } def _checkForm(self, html, message_, action_url, form_tag_attrs, submit_text): E = oidutil.importElementTree() # Build element tree from HTML source input_tree = E.ElementTree(E.fromstring(html)) # Get root element form = input_tree.getroot() # Check required form attributes for k, v in self.required_form_attrs.items(): assert form.attrib[k] == v, \ "Expected '%s' for required form attribute '%s', got '%s'" % \ (v, k, form.attrib[k]) # Check extra form attributes for k, v in form_tag_attrs.items(): # Skip attributes that already passed the required # attribute check, since they should be ignored by the # form generation code. if k in self.required_form_attrs: continue assert form.attrib[k] == v, \ "Form attribute '%s' should be '%s', found '%s'" % \ (k, v, form.attrib[k]) # Check hidden fields against post args hiddens = [e for e in form \ if e.tag.upper() == 'INPUT' and \ e.attrib['type'].upper() == 'HIDDEN'] # For each post arg, make sure there is a hidden with that # value. Make sure there are no other hiddens. for name, value in message_.toPostArgs().items(): for e in hiddens: if e.attrib['name'] == name: # value will be bytes if isinstance(e.attrib['value'], bytes): value = value.encode("utf-8") assert e.attrib['value'] == value, \ "Expected value of hidden input '%s' to be '%s', got '%s'" % \ (e.attrib['name'], value, e.attrib['value']) break else: self.fail("Post arg '%s' not found in form" % (name,)) for e in hiddens: assert e.attrib['name'] in list(message_.toPostArgs().keys()), \ "Form element for '%s' not in " + \ "original message" % (e.attrib['name']) # Check action URL assert form.attrib['action'] == action_url, \ "Expected form 'action' to be '%s', got '%s'" % \ (action_url, form.attrib['action']) # Check submit text submits = [e for e in form \ if e.tag.upper() == 'INPUT' and \ e.attrib['type'].upper() == 'SUBMIT'] assert len(submits) == 1, \ "Expected only one 'input' with type = 'submit', got %d" % \ (len(submits),) assert submits[0].attrib['value'] == submit_text, \ "Expected submit value to be '%s', got '%s'" % \ (submit_text, submits[0].attrib['value']) def test_toFormMarkup(self): m = message.Message.fromPostArgs(self.postargs) html = m.toFormMarkup(self.action_url, self.form_tag_attrs, self.submit_text) self._checkForm(html, m, self.action_url, self.form_tag_attrs, self.submit_text) def test_toFormMarkup_bug_with_utf8_values(self): postargs = { 'openid.ns': message.OPENID2_NS, 'openid.mode': 'checkid_setup', 'openid.identity': 'http://bogus.example.invalid:port/', 'openid.assoc_handle': 'FLUB', 'openid.return_to': 'Neverland', 'ünicöde_key': 'ünicöde_välüe', } m = message.Message.fromPostArgs(postargs) # Calling m.toFormMarkup with lxml used for ElementTree will throw # a ValueError. html = m.toFormMarkup(self.action_url, self.form_tag_attrs, self.submit_text) # Using the (c)ElementTree from stdlib will result in the UTF-8 # encoded strings to be converted to XML character references, # "ünicöde_key" becomes "ünicöde_key" and # "ünicöde_välüe" becomes "ünicöde_välüe" self.assertFalse( 'ünicöde_key' in html, 'UTF-8 bytes should not convert to XML character references') self.assertFalse( 'ünicöde_välüe' in html, 'UTF-8 bytes should not convert to XML character references') def test_overrideMethod(self): """Be sure that caller cannot change form method to GET.""" m = message.Message.fromPostArgs(self.postargs) tag_attrs = dict(self.form_tag_attrs) tag_attrs['method'] = 'GET' html = m.toFormMarkup(self.action_url, self.form_tag_attrs, self.submit_text) self._checkForm(html, m, self.action_url, self.form_tag_attrs, self.submit_text) def test_overrideRequired(self): """Be sure that caller CANNOT change the form charset for encoding type.""" m = message.Message.fromPostArgs(self.postargs) tag_attrs = dict(self.form_tag_attrs) tag_attrs['accept-charset'] = 'UCS4' tag_attrs['enctype'] = 'invalid/x-broken' html = m.toFormMarkup(self.action_url, tag_attrs, self.submit_text) self._checkForm(html, m, self.action_url, tag_attrs, self.submit_text) def test_setOpenIDNamespace_invalid(self): m = message.Message() invalid_things = [ # Empty string is not okay here. '', # Good guess! But wrong. 'http://openid.net/signon/2.0', # What? 'http://specs%\\\r2Eopenid.net/auth/2.0', # Too much escapings! 'http%3A%2F%2Fspecs.openid.net%2Fauth%2F2.0', # This is a Type URI, not a openid.ns value. 'http://specs.openid.net/auth/2.0/signon', ] for x in invalid_things: self.assertRaises(message.InvalidOpenIDNamespace, m.setOpenIDNamespace, x, False) def test_isOpenID1(self): v1_namespaces = [ # Yes, there are two of them. 'http://openid.net/signon/1.1', 'http://openid.net/signon/1.0', ] for ns in v1_namespaces: m = message.Message(ns) self.assertTrue(m.isOpenID1(), "%r not recognized as OpenID 1" % (ns,)) self.assertEqual(ns, m.getOpenIDNamespace()) self.assertTrue( m.namespaces.isImplicit(ns), m.namespaces.getNamespaceURI(message.NULL_NAMESPACE)) def test_isOpenID2(self): ns = 'http://specs.openid.net/auth/2.0' m = message.Message(ns) self.assertTrue(m.isOpenID2()) self.assertFalse(m.namespaces.isImplicit(message.NULL_NAMESPACE)) self.assertEqual(ns, m.getOpenIDNamespace()) def test_setOpenIDNamespace_explicit(self): m = message.Message() m.setOpenIDNamespace(message.THE_OTHER_OPENID1_NS, False) self.assertFalse(m.namespaces.isImplicit(message.THE_OTHER_OPENID1_NS)) def test_setOpenIDNamespace_implicit(self): m = message.Message() m.setOpenIDNamespace(message.THE_OTHER_OPENID1_NS, True) self.assertTrue(m.namespaces.isImplicit(message.THE_OTHER_OPENID1_NS)) def test_explicitOpenID11NSSerialzation(self): m = message.Message() m.setOpenIDNamespace(message.THE_OTHER_OPENID1_NS, implicit=False) post_args = m.toPostArgs() self.assertEqual(post_args, { 'openid.ns': message.THE_OTHER_OPENID1_NS, }) def test_fromPostArgs_ns11(self): # An example of the stuff that some Drupal installations send us, # which includes openid.ns but is 1.1. query = { 'openid.assoc_handle': '', 'openid.claimed_id': 'http://foobar.invalid/', 'openid.identity': 'http://foobar.myopenid.com', 'openid.mode': 'checkid_setup', 'openid.ns': 'http://openid.net/signon/1.1', 'openid.ns.sreg': 'http://openid.net/extensions/sreg/1.1', 'openid.return_to': 'http://drupal.invalid/return_to', 'openid.sreg.required': 'nickname,email', 'openid.trust_root': 'http://drupal.invalid', } m = message.Message.fromPostArgs(query) self.assertTrue(m.isOpenID1()) class NamespaceMapTest(unittest.TestCase): def test_onealias(self): nsm = message.NamespaceMap() uri = 'http://example.com/foo' alias = "foo" nsm.addAlias(uri, alias) self.assertTrue(nsm.getNamespaceURI(alias) == uri) self.assertTrue(nsm.getAlias(uri) == alias) def test_iteration(self): nsm = message.NamespaceMap() uripat = 'http://example.com/foo%r' nsm.add(uripat % 0) for n in range(1, 23): self.assertTrue(uripat % (n - 1) in nsm) self.assertTrue(nsm.isDefined(uripat % (n - 1))) nsm.add(uripat % n) for (uri, alias) in nsm.items(): self.assertTrue(uri[22:] == alias[3:]) i = 0 it = nsm.iterAliases() try: while True: next(it) i += 1 except StopIteration: self.assertTrue(i == 23) i = 0 it = nsm.iterNamespaceURIs() try: while True: next(it) i += 1 except StopIteration: self.assertTrue(i == 23) if __name__ == '__main__': unittest.main()