libs/openid/test/test_message.py

   1 from openid import message
   2 from openid import oidutil
   3 from openid.extensions import sreg
   4
   5 import urllib
   6 import cgi
   7 import unittest
   8
   9 def mkGetArgTest(ns, key, expected=None):
  10     def test(self):
  11         a_default = object()
  12         self.failUnlessEqual(self.msg.getArg(ns, key), expected)
  13         if expected is None:
  14             self.failUnlessEqual(
  15                 self.msg.getArg(ns, key, a_default), a_default)
  16             self.failUnlessRaises(
  17                 KeyError, self.msg.getArg, ns, key, message.no_default)
  18         else:
  19             self.failUnlessEqual(
  20                 self.msg.getArg(ns, key, a_default), expected)
  21             self.failUnlessEqual(
  22                 self.msg.getArg(ns, key, message.no_default), expected)
  23
  24     return test
  25
  26 class EmptyMessageTest(unittest.TestCase):
  27     def setUp(self):
  28         self.msg = message.Message()
  29
  30     def test_toPostArgs(self):
  31         self.failUnlessEqual(self.msg.toPostArgs(), {})
  32
  33     def test_toArgs(self):
  34         self.failUnlessEqual(self.msg.toArgs(), {})
  35
  36     def test_toKVForm(self):
  37         self.failUnlessEqual(self.msg.toKVForm(), '')
  38
  39     def test_toURLEncoded(self):
  40         self.failUnlessEqual(self.msg.toURLEncoded(), '')
  41
  42     def test_toURL(self):
  43         base_url = 'http://base.url/'
  44         self.failUnlessEqual(self.msg.toURL(base_url), base_url)
  45
  46     def test_getOpenID(self):
  47         self.failUnlessEqual(self.msg.getOpenIDNamespace(), None)
  48
  49     def test_getKeyOpenID(self):
  50         # Could reasonably return None instead of raising an
  51         # exception. I'm not sure which one is more right, since this
  52         # case should only happen when you're building a message from
  53         # scratch and so have no default namespace.
  54         self.failUnlessRaises(message.UndefinedOpenIDNamespace,
  55                               self.msg.getKey, message.OPENID_NS, 'foo')
  56
  57     def test_getKeyBARE(self):
  58         self.failUnlessEqual(self.msg.getKey(message.BARE_NS, 'foo'), 'foo')
  59
  60     def test_getKeyNS1(self):
  61         self.failUnlessEqual(self.msg.getKey(message.OPENID1_NS, 'foo'), None)
  62
  63     def test_getKeyNS2(self):
  64         self.failUnlessEqual(self.msg.getKey(message.OPENID2_NS, 'foo'), None)
  65
  66     def test_getKeyNS3(self):
  67         self.failUnlessEqual(self.msg.getKey('urn:nothing-significant', 'foo'),
  68                              None)
  69
  70     def test_hasKey(self):
  71         # Could reasonably return False instead of raising an
  72         # exception. I'm not sure which one is more right, since this
  73         # case should only happen when you're building a message from
  74         # scratch and so have no default namespace.
  75         self.failUnlessRaises(message.UndefinedOpenIDNamespace,
  76                               self.msg.hasKey, message.OPENID_NS, 'foo')
  77
  78     def test_hasKeyBARE(self):
  79         self.failUnlessEqual(self.msg.hasKey(message.BARE_NS, 'foo'), False)
  80
  81     def test_hasKeyNS1(self):
  82         self.failUnlessEqual(self.msg.hasKey(message.OPENID1_NS, 'foo'), False)
  83
  84     def test_hasKeyNS2(self):
  85         self.failUnlessEqual(self.msg.hasKey(message.OPENID2_NS, 'foo'), False)
  86
  87     def test_hasKeyNS3(self):
  88         self.failUnlessEqual(self.msg.hasKey('urn:nothing-significant', 'foo'),
  89                              False)
  90
  91     def test_getAliasedArgSuccess(self):
  92         msg = message.Message.fromPostArgs({'openid.ns.test': 'urn://foo',
  93                                             'openid.test.flub': 'bogus'})
  94         actual_uri = msg.getAliasedArg('ns.test', message.no_default)
  95         self.assertEquals("urn://foo", actual_uri)
  96
  97     def test_getAliasedArgFailure(self):
  98         msg = message.Message.fromPostArgs({'openid.test.flub': 'bogus'})
  99         self.assertRaises(KeyError,
 100                           msg.getAliasedArg, 'ns.test', message.no_default)
 101
 102     def test_getArg(self):
 103         # Could reasonably return None instead of raising an
 104         # exception. I'm not sure which one is more right, since this
 105         # case should only happen when you're building a message from
 106         # scratch and so have no default namespace.
 107         self.failUnlessRaises(message.UndefinedOpenIDNamespace,
 108                               self.msg.getArg, message.OPENID_NS, 'foo')
 109
 110     test_getArgBARE = mkGetArgTest(message.BARE_NS, 'foo')
 111     test_getArgNS1 = mkGetArgTest(message.OPENID1_NS, 'foo')
 112     test_getArgNS2 = mkGetArgTest(message.OPENID2_NS, 'foo')
 113     test_getArgNS3 = mkGetArgTest('urn:nothing-significant', 'foo')
 114
 115     def test_getArgs(self):
 116         # Could reasonably return {} instead of raising an
 117         # exception. I'm not sure which one is more right, since this
 118         # case should only happen when you're building a message from
 119         # scratch and so have no default namespace.
 120         self.failUnlessRaises(message.UndefinedOpenIDNamespace,
 121                               self.msg.getArgs, message.OPENID_NS)
 122
 123     def test_getArgsBARE(self):
 124         self.failUnlessEqual(self.msg.getArgs(message.BARE_NS), {})
 125
 126     def test_getArgsNS1(self):
 127         self.failUnlessEqual(self.msg.getArgs(message.OPENID1_NS), {})
 128
 129     def test_getArgsNS2(self):
 130         self.failUnlessEqual(self.msg.getArgs(message.OPENID2_NS), {})
 131
 132     def test_getArgsNS3(self):
 133         self.failUnlessEqual(self.msg.getArgs('urn:nothing-significant'), {})
 134
 135     def test_updateArgs(self):
 136         self.failUnlessRaises(message.UndefinedOpenIDNamespace,
 137                               self.msg.updateArgs, message.OPENID_NS,
 138                               {'does not':'matter'})
 139
 140     def _test_updateArgsNS(self, ns):
 141         update_args = {
 142             'Camper van Beethoven':'David Lowery',
 143             'Magnolia Electric Co.':'Jason Molina',
 144             }
 145
 146         self.failUnlessEqual(self.msg.getArgs(ns), {})
 147         self.msg.updateArgs(ns, update_args)
 148         self.failUnlessEqual(self.msg.getArgs(ns), update_args)
 149
 150     def test_updateArgsBARE(self):
 151         self._test_updateArgsNS(message.BARE_NS)
 152
 153     def test_updateArgsNS1(self):
 154         self._test_updateArgsNS(message.OPENID1_NS)
 155
 156     def test_updateArgsNS2(self):
 157         self._test_updateArgsNS(message.OPENID2_NS)
 158
 159     def test_updateArgsNS3(self):
 160         self._test_updateArgsNS('urn:nothing-significant')
 161
 162     def test_setArg(self):
 163         self.failUnlessRaises(message.UndefinedOpenIDNamespace,
 164                               self.msg.setArg, message.OPENID_NS,
 165                               'does not', 'matter')
 166
 167     def _test_setArgNS(self, ns):
 168         key = 'Camper van Beethoven'
 169         value = 'David Lowery'
 170         self.failUnlessEqual(self.msg.getArg(ns, key), None)
 171         self.msg.setArg(ns, key, value)
 172         self.failUnlessEqual(self.msg.getArg(ns, key), value)
 173
 174     def test_setArgBARE(self):
 175         self._test_setArgNS(message.BARE_NS)
 176
 177     def test_setArgNS1(self):
 178         self._test_setArgNS(message.OPENID1_NS)
 179
 180     def test_setArgNS2(self):
 181         self._test_setArgNS(message.OPENID2_NS)
 182
 183     def test_setArgNS3(self):
 184         self._test_setArgNS('urn:nothing-significant')
 185
 186     def test_setArgToNone(self):
 187         self.failUnlessRaises(AssertionError, self.msg.setArg,
 188                               message.OPENID1_NS, 'op_endpoint', None)
 189
 190     def test_delArg(self):
 191         # Could reasonably raise KeyError instead of raising
 192         # UndefinedOpenIDNamespace. I'm not sure which one is more
 193         # right, since this case should only happen when you're
 194         # building a message from scratch and so have no default
 195         # namespace.
 196         self.failUnlessRaises(message.UndefinedOpenIDNamespace,
 197                               self.msg.delArg, message.OPENID_NS, 'key')
 198
 199     def _test_delArgNS(self, ns):
 200         key = 'Camper van Beethoven'
 201         self.failUnlessRaises(KeyError, self.msg.delArg, ns, key)
 202
 203     def test_delArgBARE(self):
 204         self._test_delArgNS(message.BARE_NS)
 205
 206     def test_delArgNS1(self):
 207         self._test_delArgNS(message.OPENID1_NS)
 208
 209     def test_delArgNS2(self):
 210         self._test_delArgNS(message.OPENID2_NS)
 211
 212     def test_delArgNS3(self):
 213         self._test_delArgNS('urn:nothing-significant')
 214
 215     def test_isOpenID1(self):
 216         self.failIf(self.msg.isOpenID1())
 217
 218     def test_isOpenID2(self):
 219         self.failIf(self.msg.isOpenID2())
 220
 221 class OpenID1MessageTest(unittest.TestCase):
 222     def setUp(self):
 223         self.msg = message.Message.fromPostArgs({'openid.mode':'error',
 224                                                  'openid.error':'unit test'})
 225
 226     def test_toPostArgs(self):
 227         self.failUnlessEqual(self.msg.toPostArgs(),
 228                              {'openid.mode':'error',
 229                               'openid.error':'unit test'})
 230
 231     def test_toArgs(self):
 232         self.failUnlessEqual(self.msg.toArgs(), {'mode':'error',
 233                                                  'error':'unit test'})
 234
 235     def test_toKVForm(self):
 236         self.failUnlessEqual(self.msg.toKVForm(),
 237                              'error:unit test\nmode:error\n')
 238
 239     def test_toURLEncoded(self):
 240         self.failUnlessEqual(self.msg.toURLEncoded(),
 241                              'openid.error=unit+test&openid.mode=error')
 242
 243     def test_toURL(self):
 244         base_url = 'http://base.url/'
 245         actual = self.msg.toURL(base_url)
 246         actual_base = actual[:len(base_url)]
 247         self.failUnlessEqual(actual_base, base_url)
 248         self.failUnlessEqual(actual[len(base_url)], '?')
 249         query = actual[len(base_url) + 1:]
 250         parsed = cgi.parse_qs(query)
 251         self.failUnlessEqual(parsed, {'openid.mode':['error'],
 252                                       'openid.error':['unit test']})
 253
 254     def test_getOpenID(self):
 255         self.failUnlessEqual(self.msg.getOpenIDNamespace(), message.OPENID1_NS)
 256
 257     def test_getKeyOpenID(self):
 258         self.failUnlessEqual(self.msg.getKey(message.OPENID_NS, 'mode'),
 259                              'openid.mode')
 260
 261     def test_getKeyBARE(self):
 262         self.failUnlessEqual(self.msg.getKey(message.BARE_NS, 'mode'), 'mode')
 263
 264     def test_getKeyNS1(self):
 265         self.failUnlessEqual(
 266             self.msg.getKey(message.OPENID1_NS, 'mode'), 'openid.mode')
 267
 268     def test_getKeyNS2(self):
 269         self.failUnlessEqual(self.msg.getKey(message.OPENID2_NS, 'mode'), None)
 270
 271     def test_getKeyNS3(self):
 272         self.failUnlessEqual(
 273             self.msg.getKey('urn:nothing-significant', 'mode'), None)
 274
 275     def test_hasKey(self):
 276         self.failUnlessEqual(self.msg.hasKey(message.OPENID_NS, 'mode'), True)
 277
 278     def test_hasKeyBARE(self):
 279         self.failUnlessEqual(self.msg.hasKey(message.BARE_NS, 'mode'), False)
 280
 281     def test_hasKeyNS1(self):
 282         self.failUnlessEqual(self.msg.hasKey(message.OPENID1_NS, 'mode'), True)
 283
 284     def test_hasKeyNS2(self):
 285         self.failUnlessEqual(
 286             self.msg.hasKey(message.OPENID2_NS, 'mode'), False)
 287
 288     def test_hasKeyNS3(self):
 289         self.failUnlessEqual(
 290             self.msg.hasKey('urn:nothing-significant', 'mode'), False)
 291
 292     test_getArgBARE = mkGetArgTest(message.BARE_NS, 'mode')
 293     test_getArgNS = mkGetArgTest(message.OPENID_NS, 'mode', 'error')
 294     test_getArgNS1 = mkGetArgTest(message.OPENID1_NS, 'mode', 'error')
 295     test_getArgNS2 = mkGetArgTest(message.OPENID2_NS, 'mode')
 296     test_getArgNS3 = mkGetArgTest('urn:nothing-significant', 'mode')
 297
 298     def test_getArgs(self):
 299         self.failUnlessEqual(self.msg.getArgs(message.OPENID_NS),
 300                              {'mode':'error',
 301                               'error':'unit test',
 302                               })
 303
 304     def test_getArgsBARE(self):
 305         self.failUnlessEqual(self.msg.getArgs(message.BARE_NS), {})
 306
 307     def test_getArgsNS1(self):
 308         self.failUnlessEqual(self.msg.getArgs(message.OPENID1_NS),
 309                              {'mode':'error',
 310                               'error':'unit test',
 311                               })
 312
 313     def test_getArgsNS2(self):
 314         self.failUnlessEqual(self.msg.getArgs(message.OPENID2_NS), {})
 315
 316     def test_getArgsNS3(self):
 317         self.failUnlessEqual(self.msg.getArgs('urn:nothing-significant'), {})
 318
 319     def _test_updateArgsNS(self, ns, before=None):
 320         if before is None:
 321             before = {}
 322         update_args = {
 323             'Camper van Beethoven':'David Lowery',
 324             'Magnolia Electric Co.':'Jason Molina',
 325             }
 326
 327         self.failUnlessEqual(self.msg.getArgs(ns), before)
 328         self.msg.updateArgs(ns, update_args)
 329         after = dict(before)
 330         after.update(update_args)
 331         self.failUnlessEqual(self.msg.getArgs(ns), after)
 332
 333     def test_updateArgs(self):
 334         self._test_updateArgsNS(message.OPENID_NS,
 335                                 before={'mode':'error', 'error':'unit test'})
 336
 337     def test_updateArgsBARE(self):
 338         self._test_updateArgsNS(message.BARE_NS)
 339
 340     def test_updateArgsNS1(self):
 341         self._test_updateArgsNS(message.OPENID1_NS,
 342                                 before={'mode':'error', 'error':'unit test'})
 343
 344     def test_updateArgsNS2(self):
 345         self._test_updateArgsNS(message.OPENID2_NS)
 346
 347     def test_updateArgsNS3(self):
 348         self._test_updateArgsNS('urn:nothing-significant')
 349
 350     def _test_setArgNS(self, ns):
 351         key = 'Camper van Beethoven'
 352         value = 'David Lowery'
 353         self.failUnlessEqual(self.msg.getArg(ns, key), None)
 354         self.msg.setArg(ns, key, value)
 355         self.failUnlessEqual(self.msg.getArg(ns, key), value)
 356
 357     def test_setArg(self):
 358         self._test_setArgNS(message.OPENID_NS)
 359
 360     def test_setArgBARE(self):
 361         self._test_setArgNS(message.BARE_NS)
 362
 363     def test_setArgNS1(self):
 364         self._test_setArgNS(message.OPENID1_NS)
 365
 366     def test_setArgNS2(self):
 367         self._test_setArgNS(message.OPENID2_NS)
 368
 369     def test_setArgNS3(self):
 370         self._test_setArgNS('urn:nothing-significant')
 371
 372     def _test_delArgNS(self, ns):
 373         key = 'Camper van Beethoven'
 374         value = 'David Lowery'
 375
 376         self.failUnlessRaises(KeyError, self.msg.delArg, ns, key)
 377         self.msg.setArg(ns, key, value)
 378         self.failUnlessEqual(self.msg.getArg(ns, key), value)
 379         self.msg.delArg(ns, key)
 380         self.failUnlessEqual(self.msg.getArg(ns, key), None)
 381
 382     def test_delArg(self):
 383         self._test_delArgNS(message.OPENID_NS)
 384
 385     def test_delArgBARE(self):
 386         self._test_delArgNS(message.BARE_NS)
 387
 388     def test_delArgNS1(self):
 389         self._test_delArgNS(message.OPENID1_NS)
 390
 391     def test_delArgNS2(self):
 392         self._test_delArgNS(message.OPENID2_NS)
 393
 394     def test_delArgNS3(self):
 395         self._test_delArgNS('urn:nothing-significant')
 396
 397
 398     def test_isOpenID1(self):
 399         self.failUnless(self.msg.isOpenID1())
 400
 401     def test_isOpenID2(self):
 402         self.failIf(self.msg.isOpenID2())
 403
 404 class OpenID1ExplicitMessageTest(unittest.TestCase):
 405     def setUp(self):
 406         self.msg = message.Message.fromPostArgs({'openid.mode':'error',
 407                                                  'openid.error':'unit test',
 408                                                  'openid.ns':message.OPENID1_NS
 409                                                  })
 410
 411     def test_toPostArgs(self):
 412         self.failUnlessEqual(self.msg.toPostArgs(),
 413                              {'openid.mode':'error',
 414                               'openid.error':'unit test',
 415                               'openid.ns':message.OPENID1_NS
 416                               })
 417
 418     def test_toArgs(self):
 419         self.failUnlessEqual(self.msg.toArgs(), {'mode':'error',
 420                                                  'error':'unit test',
 421                                                  'ns':message.OPENID1_NS})
 422
 423     def test_toKVForm(self):
 424         self.failUnlessEqual(self.msg.toKVForm(),
 425                              'error:unit test\nmode:error\nns:%s\n'
 426                               %message.OPENID1_NS)
 427
 428     def test_toURLEncoded(self):
 429         self.failUnlessEqual(self.msg.toURLEncoded(),
 430                              'openid.error=unit+test&openid.mode=error&openid.ns=http%3A%2F%2Fopenid.net%2Fsignon%2F1.0')
 431
 432     def test_toURL(self):
 433         base_url = 'http://base.url/'
 434         actual = self.msg.toURL(base_url)
 435         actual_base = actual[:len(base_url)]
 436         self.failUnlessEqual(actual_base, base_url)
 437         self.failUnlessEqual(actual[len(base_url)], '?')
 438         query = actual[len(base_url) + 1:]
 439         parsed = cgi.parse_qs(query)
 440         self.failUnlessEqual(parsed, {'openid.mode':['error'],
 441                                       'openid.error':['unit test'],
 442                                       'openid.ns':[message.OPENID1_NS]
 443                                       })
 444
 445     def test_isOpenID1(self):
 446         self.failUnless(self.msg.isOpenID1())
 447
 448
 449 class OpenID2MessageTest(unittest.TestCase):
 450     def setUp(self):
 451         self.msg = message.Message.fromPostArgs({'openid.mode':'error',
 452                                                  'openid.error':'unit test',
 453                                                  'openid.ns':message.OPENID2_NS
 454                                                  })
 455         self.msg.setArg(message.BARE_NS, "xey", "value")
 456
 457     def test_toPostArgs(self):
 458         self.failUnlessEqual(self.msg.toPostArgs(),
 459                              {'openid.mode':'error',
 460                               'openid.error':'unit test',
 461                               'openid.ns':message.OPENID2_NS,
 462                               'xey': 'value',
 463                               })
 464
 465     def test_toArgs(self):
 466         # This method can't tolerate BARE_NS.
 467         self.msg.delArg(message.BARE_NS, "xey")
 468         self.failUnlessEqual(self.msg.toArgs(), {'mode':'error',
 469                                                  'error':'unit test',
 470                                                  'ns':message.OPENID2_NS,
 471                                                  })
 472
 473     def test_toKVForm(self):
 474         # Can't tolerate BARE_NS in kvform
 475         self.msg.delArg(message.BARE_NS, "xey")
 476         self.failUnlessEqual(self.msg.toKVForm(),
 477                              'error:unit test\nmode:error\nns:%s\n' %
 478                              (message.OPENID2_NS,))
 479
 480     def _test_urlencoded(self, s):
 481         expected = ('openid.error=unit+test&openid.mode=error&'
 482                     'openid.ns=%s&xey=value' % (
 483             urllib.quote(message.OPENID2_NS, ''),))
 484         self.failUnlessEqual(s, expected)
 485
 486
 487     def test_toURLEncoded(self):
 488         self._test_urlencoded(self.msg.toURLEncoded())
 489
 490     def test_toURL(self):
 491         base_url = 'http://base.url/'
 492         actual = self.msg.toURL(base_url)
 493         actual_base = actual[:len(base_url)]
 494         self.failUnlessEqual(actual_base, base_url)
 495         self.failUnlessEqual(actual[len(base_url)], '?')
 496         query = actual[len(base_url) + 1:]
 497         self._test_urlencoded(query)
 498
 499     def test_getOpenID(self):
 500         self.failUnlessEqual(self.msg.getOpenIDNamespace(), message.OPENID2_NS)
 501
 502     def test_getKeyOpenID(self):
 503         self.failUnlessEqual(self.msg.getKey(message.OPENID_NS, 'mode'),
 504                              'openid.mode')
 505
 506     def test_getKeyBARE(self):
 507         self.failUnlessEqual(self.msg.getKey(message.BARE_NS, 'mode'), 'mode')
 508
 509     def test_getKeyNS1(self):
 510         self.failUnlessEqual(
 511             self.msg.getKey(message.OPENID1_NS, 'mode'), None)
 512
 513     def test_getKeyNS2(self):
 514         self.failUnlessEqual(
 515             self.msg.getKey(message.OPENID2_NS, 'mode'), 'openid.mode')
 516
 517     def test_getKeyNS3(self):
 518         self.failUnlessEqual(
 519             self.msg.getKey('urn:nothing-significant', 'mode'), None)
 520
 521     def test_hasKeyOpenID(self):
 522         self.failUnlessEqual(self.msg.hasKey(message.OPENID_NS, 'mode'), True)
 523
 524     def test_hasKeyBARE(self):
 525         self.failUnlessEqual(self.msg.hasKey(message.BARE_NS, 'mode'), False)
 526
 527     def test_hasKeyNS1(self):
 528         self.failUnlessEqual(
 529             self.msg.hasKey(message.OPENID1_NS, 'mode'), False)
 530
 531     def test_hasKeyNS2(self):
 532         self.failUnlessEqual(
 533             self.msg.hasKey(message.OPENID2_NS, 'mode'), True)
 534
 535     def test_hasKeyNS3(self):
 536         self.failUnlessEqual(
 537             self.msg.hasKey('urn:nothing-significant', 'mode'), False)
 538
 539     test_getArgBARE = mkGetArgTest(message.BARE_NS, 'mode')
 540     test_getArgNS = mkGetArgTest(message.OPENID_NS, 'mode', 'error')
 541     test_getArgNS1 = mkGetArgTest(message.OPENID1_NS, 'mode')
 542     test_getArgNS2 = mkGetArgTest(message.OPENID2_NS, 'mode', 'error')
 543     test_getArgNS3 = mkGetArgTest('urn:nothing-significant', 'mode')
 544
 545     def test_getArgsOpenID(self):
 546         self.failUnlessEqual(self.msg.getArgs(message.OPENID_NS),
 547                              {'mode':'error',
 548                               'error':'unit test',
 549                               })
 550
 551     def test_getArgsBARE(self):
 552         self.failUnlessEqual(self.msg.getArgs(message.BARE_NS),
 553                              {'xey': 'value'})
 554
 555     def test_getArgsNS1(self):
 556         self.failUnlessEqual(self.msg.getArgs(message.OPENID1_NS), {})
 557
 558     def test_getArgsNS2(self):
 559         self.failUnlessEqual(self.msg.getArgs(message.OPENID2_NS),
 560                              {'mode':'error',
 561                               'error':'unit test',
 562                               })
 563
 564     def test_getArgsNS3(self):
 565         self.failUnlessEqual(self.msg.getArgs('urn:nothing-significant'), {})
 566
 567     def _test_updateArgsNS(self, ns, before=None):
 568         if before is None:
 569             before = {}
 570         update_args = {
 571             'Camper van Beethoven':'David Lowery',
 572             'Magnolia Electric Co.':'Jason Molina',
 573             }
 574
 575         self.failUnlessEqual(self.msg.getArgs(ns), before)
 576         self.msg.updateArgs(ns, update_args)
 577         after = dict(before)
 578         after.update(update_args)
 579         self.failUnlessEqual(self.msg.getArgs(ns), after)
 580
 581     def test_updateArgsOpenID(self):
 582         self._test_updateArgsNS(message.OPENID_NS,
 583                                 before={'mode':'error', 'error':'unit test'})
 584
 585     def test_updateArgsBARE(self):
 586         self._test_updateArgsNS(message.BARE_NS,
 587                                 before={'xey':'value'})
 588
 589     def test_updateArgsNS1(self):
 590         self._test_updateArgsNS(message.OPENID1_NS)
 591
 592     def test_updateArgsNS2(self):
 593         self._test_updateArgsNS(message.OPENID2_NS,
 594                                 before={'mode':'error', 'error':'unit test'})
 595
 596     def test_updateArgsNS3(self):
 597         self._test_updateArgsNS('urn:nothing-significant')
 598
 599     def _test_setArgNS(self, ns):
 600         key = 'Camper van Beethoven'
 601         value = 'David Lowery'
 602         self.failUnlessEqual(self.msg.getArg(ns, key), None)
 603         self.msg.setArg(ns, key, value)
 604         self.failUnlessEqual(self.msg.getArg(ns, key), value)
 605
 606     def test_setArgOpenID(self):
 607         self._test_setArgNS(message.OPENID_NS)
 608
 609     def test_setArgBARE(self):
 610         self._test_setArgNS(message.BARE_NS)
 611
 612     def test_setArgNS1(self):
 613         self._test_setArgNS(message.OPENID1_NS)
 614
 615     def test_setArgNS2(self):
 616         self._test_setArgNS(message.OPENID2_NS)
 617
 618     def test_setArgNS3(self):
 619         self._test_setArgNS('urn:nothing-significant')
 620
 621     def test_badAlias(self):
 622         """Make sure dotted aliases and OpenID protocol fields are not
 623         allowed as namespace aliases."""
 624
 625         for f in message.OPENID_PROTOCOL_FIELDS + ['dotted.alias']:
 626             args = {'openid.ns.%s' % f: 'blah',
 627                     'openid.%s.foo' % f: 'test'}
 628
 629             # .fromPostArgs covers .fromPostArgs, .fromOpenIDArgs,
 630             # ._fromOpenIDArgs, and .fromOpenIDArgs (since it calls
 631             # .fromPostArgs).
 632             self.failUnlessRaises(AssertionError, self.msg.fromPostArgs,
 633                                   args)
 634
 635     def test_mysterious_missing_namespace_bug(self):
 636         """A failing test for bug #112"""
 637         openid_args = {
 638           'assoc_handle': '{{HMAC-SHA256}{1211477242.29743}{v5cadg==}',
 639           'claimed_id': 'http://nerdbank.org/OPAffirmative/AffirmativeIdentityWithSregNoAssoc.aspx',
 640           'ns.sreg': 'http://openid.net/extensions/sreg/1.1',
 641           'response_nonce': '2008-05-22T17:27:22ZUoW5.\\NV',
 642           'signed': 'return_to,identity,claimed_id,op_endpoint,response_nonce,ns.sreg,sreg.email,sreg.nickname,assoc_handle',
 643           'sig': 'e3eGZ10+TNRZitgq5kQlk5KmTKzFaCRI8OrRoXyoFa4=',
 644           'mode': 'check_authentication',
 645           'op_endpoint': 'http://nerdbank.org/OPAffirmative/ProviderNoAssoc.aspx',
 646           'sreg.nickname': 'Andy',
 647           'return_to': 'http://localhost.localdomain:8001/process?janrain_nonce=2008-05-22T17%3A27%3A21ZnxHULd',
 648           'invalidate_handle': '{{HMAC-SHA1}{1211477241.92242}{H0akXw==}',
 649           'identity': 'http://nerdbank.org/OPAffirmative/AffirmativeIdentityWithSregNoAssoc.aspx',
 650           'sreg.email': 'a@b.com'
 651           }
 652         m = message.Message.fromOpenIDArgs(openid_args)
 653
 654         self.failUnless(('http://openid.net/extensions/sreg/1.1', 'sreg') in
 655                         list(m.namespaces.iteritems()))
 656         missing = []
 657         for k in openid_args['signed'].split(','):
 658             if not ("openid."+k) in m.toPostArgs().keys():
 659                 missing.append(k)
 660         self.assertEqual([], missing, missing)
 661         self.assertEqual(openid_args, m.toArgs())
 662         self.failUnless(m.isOpenID1())
 663
 664     def test_112B(self):
 665         args = {'openid.assoc_handle': 'fa1f5ff0-cde4-11dc-a183-3714bfd55ca8',
 666                 'openid.claimed_id': 'http://binkley.lan/user/test01',
 667                 'openid.identity': 'http://test01.binkley.lan/',
 668                 'openid.mode': 'id_res',
 669                 'openid.ns': 'http://specs.openid.net/auth/2.0',
 670                 'openid.ns.pape': 'http://specs.openid.net/extensions/pape/1.0',
 671                 'openid.op_endpoint': 'http://binkley.lan/server',
 672                 'openid.pape.auth_policies': 'none',
 673                 'openid.pape.auth_time': '2008-01-28T20:42:36Z',
 674                 'openid.pape.nist_auth_level': '0',
 675                 'openid.response_nonce': '2008-01-28T21:07:04Z99Q=',
 676                 'openid.return_to': 'http://binkley.lan:8001/process?janrain_nonce=2008-01-28T21%3A07%3A02Z0tMIKx',
 677                 'openid.sig': 'YJlWH4U6SroB1HoPkmEKx9AyGGg=',
 678                 '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'
 679                 }
 680         m = message.Message.fromPostArgs(args)
 681         missing = []
 682         for k in args['openid.signed'].split(','):
 683             if not ("openid."+k) in m.toPostArgs().keys():
 684                 missing.append(k)
 685         self.assertEqual([], missing, missing)
 686         self.assertEqual(args, m.toPostArgs())
 687         self.failUnless(m.isOpenID2())
 688
 689     def test_implicit_sreg_ns(self):
 690         openid_args = {
 691           'sreg.email': 'a@b.com'
 692           }
 693         m = message.Message.fromOpenIDArgs(openid_args)
 694         self.failUnless((sreg.ns_uri, 'sreg') in
 695                         list(m.namespaces.iteritems()))
 696         self.assertEqual('a@b.com', m.getArg(sreg.ns_uri, 'email'))
 697         self.assertEqual(openid_args, m.toArgs())
 698         self.failUnless(m.isOpenID1())
 699
 700     def _test_delArgNS(self, ns):
 701         key = 'Camper van Beethoven'
 702         value = 'David Lowery'
 703
 704         self.failUnlessRaises(KeyError, self.msg.delArg, ns, key)
 705         self.msg.setArg(ns, key, value)
 706         self.failUnlessEqual(self.msg.getArg(ns, key), value)
 707         self.msg.delArg(ns, key)
 708         self.failUnlessEqual(self.msg.getArg(ns, key), None)
 709
 710     def test_delArgOpenID(self):
 711         self._test_delArgNS(message.OPENID_NS)
 712
 713     def test_delArgBARE(self):
 714         self._test_delArgNS(message.BARE_NS)
 715
 716     def test_delArgNS1(self):
 717         self._test_delArgNS(message.OPENID1_NS)
 718
 719     def test_delArgNS2(self):
 720         self._test_delArgNS(message.OPENID2_NS)
 721
 722     def test_delArgNS3(self):
 723         self._test_delArgNS('urn:nothing-significant')
 724
 725     def test_overwriteExtensionArg(self):
 726         ns = 'urn:unittest_extension'
 727         key = 'mykey'
 728         value_1 = 'value_1'
 729         value_2 = 'value_2'
 730
 731         self.msg.setArg(ns, key, value_1)
 732         self.failUnless(self.msg.getArg(ns, key) == value_1)
 733         self.msg.setArg(ns, key, value_2)
 734         self.failUnless(self.msg.getArg(ns, key) == value_2)
 735
 736     def test_argList(self):
 737         self.failUnlessRaises(TypeError, self.msg.fromPostArgs,
 738                               {'arg': [1, 2, 3]})
 739
 740     def test_isOpenID1(self):
 741         self.failIf(self.msg.isOpenID1())
 742
 743     def test_isOpenID2(self):
 744         self.failUnless(self.msg.isOpenID2())
 745
 746 class MessageTest(unittest.TestCase):
 747     def setUp(self):
 748         self.postargs = {
 749             'openid.ns': message.OPENID2_NS,
 750             'openid.mode': 'checkid_setup',
 751             'openid.identity': 'http://bogus.example.invalid:port/',
 752             'openid.assoc_handle': 'FLUB',
 753             'openid.return_to': 'Neverland',
 754             }
 755
 756         self.action_url = 'scheme://host:port/path?query'
 757
 758         self.form_tag_attrs = {
 759             'company': 'janrain',
 760             'class': 'fancyCSS',
 761             }
 762
 763         self.submit_text = 'GO!'
 764
 765         ### Expected data regardless of input
 766
 767         self.required_form_attrs = {
 768             'accept-charset':'UTF-8',
 769             'enctype':'application/x-www-form-urlencoded',
 770             'method': 'post',
 771             }
 772
 773     def _checkForm(self, html, message_, action_url,
 774                    form_tag_attrs, submit_text):
 775         E = oidutil.importElementTree()
 776
 777         # Build element tree from HTML source
 778         input_tree = E.ElementTree(E.fromstring(html))
 779
 780         # Get root element
 781         form = input_tree.getroot()
 782
 783         # Check required form attributes
 784         for k, v in self.required_form_attrs.iteritems():
 785             assert form.attrib[k] == v, \
 786                    "Expected '%s' for required form attribute '%s', got '%s'" % \
 787                    (v, k, form.attrib[k])
 788
 789         # Check extra form attributes
 790         for k, v in form_tag_attrs.iteritems():
 791
 792             # Skip attributes that already passed the required
 793             # attribute check, since they should be ignored by the
 794             # form generation code.
 795             if k in self.required_form_attrs:
 796                 continue
 797
 798             assert form.attrib[k] == v, \
 799                    "Form attribute '%s' should be '%s', found '%s'" % \
 800                    (k, v, form.attrib[k])
 801
 802         # Check hidden fields against post args
 803         hiddens = [e for e in form \
 804                    if e.tag.upper() == 'INPUT' and \
 805                    e.attrib['type'].upper() == 'HIDDEN']
 806
 807         # For each post arg, make sure there is a hidden with that
 808         # value.  Make sure there are no other hiddens.
 809         for name, value in message_.toPostArgs().iteritems():
 810             for e in hiddens:
 811                 if e.attrib['name'] == name:
 812                     assert e.attrib['value'] == value, \
 813                            "Expected value of hidden input '%s' to be '%s', got '%s'" % \
 814                            (e.attrib['name'], value, e.attrib['value'])
 815                     break
 816             else:
 817                 self.fail("Post arg '%s' not found in form" % (name,))
 818
 819         for e in hiddens:
 820             assert e.attrib['name'] in message_.toPostArgs().keys(), \
 821                    "Form element for '%s' not in " + \
 822                    "original message" % (e.attrib['name'])
 823
 824         # Check action URL
 825         assert form.attrib['action'] == action_url, \
 826                "Expected form 'action' to be '%s', got '%s'" % \
 827                (action_url, form.attrib['action'])
 828
 829         # Check submit text
 830         submits = [e for e in form \
 831                    if e.tag.upper() == 'INPUT' and \
 832                    e.attrib['type'].upper() == 'SUBMIT']
 833
 834         assert len(submits) == 1, \
 835                "Expected only one 'input' with type = 'submit', got %d" % \
 836                (len(submits),)
 837
 838         assert submits[0].attrib['value'] == submit_text, \
 839                "Expected submit value to be '%s', got '%s'" % \
 840                (submit_text, submits[0].attrib['value'])
 841
 842     def test_toFormMarkup(self):
 843         m = message.Message.fromPostArgs(self.postargs)
 844         html = m.toFormMarkup(self.action_url, self.form_tag_attrs,
 845                               self.submit_text)
 846         self._checkForm(html, m, self.action_url,
 847                         self.form_tag_attrs, self.submit_text)
 848
 849     def test_overrideMethod(self):
 850         """Be sure that caller cannot change form method to GET."""
 851         m = message.Message.fromPostArgs(self.postargs)
 852
 853         tag_attrs = dict(self.form_tag_attrs)
 854         tag_attrs['method'] = 'GET'
 855
 856         html = m.toFormMarkup(self.action_url, self.form_tag_attrs,
 857                               self.submit_text)
 858         self._checkForm(html, m, self.action_url,
 859                         self.form_tag_attrs, self.submit_text)
 860
 861     def test_overrideRequired(self):
 862         """Be sure that caller CANNOT change the form charset for
 863         encoding type."""
 864         m = message.Message.fromPostArgs(self.postargs)
 865
 866         tag_attrs = dict(self.form_tag_attrs)
 867         tag_attrs['accept-charset'] = 'UCS4'
 868         tag_attrs['enctype'] = 'invalid/x-broken'
 869
 870         html = m.toFormMarkup(self.action_url, tag_attrs,
 871                               self.submit_text)
 872         self._checkForm(html, m, self.action_url,
 873                         tag_attrs, self.submit_text)
 874
 875
 876     def test_setOpenIDNamespace_invalid(self):
 877         m = message.Message()
 878         invalid_things = [
 879             # Empty string is not okay here.
 880             '',
 881             # Good guess!  But wrong.
 882             'http://openid.net/signon/2.0',
 883             # What?
 884             u'http://specs%\\\r2Eopenid.net/auth/2.0',
 885             # Too much escapings!
 886             'http%3A%2F%2Fspecs.openid.net%2Fauth%2F2.0',
 887             # This is a Type URI, not a openid.ns value.
 888             'http://specs.openid.net/auth/2.0/signon',
 889             ]
 890
 891         for x in invalid_things:
 892             self.failUnlessRaises(message.InvalidOpenIDNamespace,
 893                                   m.setOpenIDNamespace, x, False)
 894
 895
 896     def test_isOpenID1(self):
 897         v1_namespaces = [
 898             # Yes, there are two of them.
 899             'http://openid.net/signon/1.1',
 900             'http://openid.net/signon/1.0',
 901             ]
 902
 903         for ns in v1_namespaces:
 904             m = message.Message(ns)
 905             self.failUnless(m.isOpenID1(), "%r not recognized as OpenID 1" %
 906                             (ns,))
 907             self.failUnlessEqual(ns, m.getOpenIDNamespace())
 908             self.failUnless(m.namespaces.isImplicit(ns),
 909                             m.namespaces.getNamespaceURI(message.NULL_NAMESPACE))
 910
 911     def test_isOpenID2(self):
 912         ns = 'http://specs.openid.net/auth/2.0'
 913         m = message.Message(ns)
 914         self.failUnless(m.isOpenID2())
 915         self.failIf(m.namespaces.isImplicit(message.NULL_NAMESPACE))
 916         self.failUnlessEqual(ns, m.getOpenIDNamespace())
 917
 918     def test_setOpenIDNamespace_explicit(self):
 919         m = message.Message()
 920         m.setOpenIDNamespace(message.THE_OTHER_OPENID1_NS, False)
 921         self.failIf(m.namespaces.isImplicit(message.THE_OTHER_OPENID1_NS))
 922
 923     def test_setOpenIDNamespace_implicit(self):
 924         m = message.Message()
 925         m.setOpenIDNamespace(message.THE_OTHER_OPENID1_NS, True)
 926         self.failUnless(m.namespaces.isImplicit(message.THE_OTHER_OPENID1_NS))
 927
 928
 929     def test_explicitOpenID11NSSerialzation(self):
 930         m = message.Message()
 931         m.setOpenIDNamespace(message.THE_OTHER_OPENID1_NS, implicit=False)
 932
 933         post_args = m.toPostArgs()
 934         self.failUnlessEqual(post_args,
 935                              {'openid.ns':message.THE_OTHER_OPENID1_NS})
 936
 937     def test_fromPostArgs_ns11(self):
 938         # An example of the stuff that some Drupal installations send us,
 939         # which includes openid.ns but is 1.1.
 940         query = {
 941             u'openid.assoc_handle': u'',
 942             u'openid.claimed_id': u'http://foobar.invalid/',
 943             u'openid.identity': u'http://foobar.myopenid.com',
 944             u'openid.mode': u'checkid_setup',
 945             u'openid.ns': u'http://openid.net/signon/1.1',
 946             u'openid.ns.sreg': u'http://openid.net/extensions/sreg/1.1',
 947             u'openid.return_to': u'http://drupal.invalid/return_to',
 948             u'openid.sreg.required': u'nickname,email',
 949             u'openid.trust_root': u'http://drupal.invalid',
 950             }
 951         m = message.Message.fromPostArgs(query)
 952         self.failUnless(m.isOpenID1())
 953
 954
 955
 956 class NamespaceMapTest(unittest.TestCase):
 957     def test_onealias(self):
 958         nsm = message.NamespaceMap()
 959         uri = 'http://example.com/foo'
 960         alias = "foo"
 961         nsm.addAlias(uri, alias)
 962         self.failUnless(nsm.getNamespaceURI(alias) == uri)
 963         self.failUnless(nsm.getAlias(uri) == alias)
 964
 965     def test_iteration(self):
 966         nsm = message.NamespaceMap()
 967         uripat = 'http://example.com/foo%r'
 968
 969         nsm.add(uripat%0)
 970         for n in range(1,23):
 971             self.failUnless(uripat%(n-1) in nsm)
 972             self.failUnless(nsm.isDefined(uripat%(n-1)))
 973             nsm.add(uripat%n)
 974
 975         for (uri, alias) in nsm.iteritems():
 976             self.failUnless(uri[22:]==alias[3:])
 977
 978         i=0
 979         it = nsm.iterAliases()
 980         try:
 981             while True:
 982                 it.next()
 983                 i += 1
 984         except StopIteration:
 985             self.failUnless(i == 23)
 986
 987         i=0
 988         it = nsm.iterNamespaceURIs()
 989         try:
 990             while True:
 991                 it.next()
 992                 i += 1
 993         except StopIteration:
 994             self.failUnless(i == 23)
 995
 996
 997 if __name__ == '__main__':
 998     unittest.main()