libs/openid/test/test_etxrd.py

   1 import unittest
   2 from openid.yadis import services, etxrd, xri
   3 import os.path
   4
   5 def datapath(filename):
   6     module_directory = os.path.dirname(os.path.abspath(__file__))
   7     return os.path.join(module_directory, 'data', 'test_etxrd', filename)
   8
   9 XRD_FILE =  datapath('valid-populated-xrds.xml')
  10 NOXRDS_FILE = datapath('not-xrds.xml')
  11 NOXRD_FILE = datapath('no-xrd.xml')
  12
  13 # None of the namespaces or service URIs below are official (or even
  14 # sanctioned by the owners of that piece of URL-space)
  15
  16 LID_2_0 = "http://lid.netmesh.org/sso/2.0b5"
  17 TYPEKEY_1_0 = "http://typekey.com/services/1.0"
  18
  19 def simpleOpenIDTransformer(endpoint):
  20     """Function to extract information from an OpenID service element"""
  21     if 'http://openid.net/signon/1.0' not in endpoint.type_uris:
  22         return None
  23
  24     delegates = list(endpoint.service_element.findall(
  25         '{http://openid.net/xmlns/1.0}Delegate'))
  26     assert len(delegates) == 1
  27     delegate = delegates[0].text
  28     return (endpoint.uri, delegate)
  29
  30 class TestServiceParser(unittest.TestCase):
  31     def setUp(self):
  32         self.xmldoc = file(XRD_FILE).read()
  33         self.yadis_url = 'http://unittest.url/'
  34
  35     def _getServices(self, flt=None):
  36         return list(services.applyFilter(self.yadis_url, self.xmldoc, flt))
  37
  38     def testParse(self):
  39         """Make sure that parsing succeeds at all"""
  40         services = self._getServices()
  41
  42     def testParseOpenID(self):
  43         """Parse for OpenID services with a transformer function"""
  44         services = self._getServices(simpleOpenIDTransformer)
  45
  46         expectedServices = [
  47             ("http://www.myopenid.com/server", "http://josh.myopenid.com/"),
  48             ("http://www.schtuff.com/openid", "http://users.schtuff.com/josh"),
  49             ("http://www.livejournal.com/openid/server.bml",
  50              "http://www.livejournal.com/users/nedthealpaca/"),
  51             ]
  52
  53         it = iter(services)
  54         for (server_url, delegate) in expectedServices:
  55             for (actual_url, actual_delegate) in it:
  56                 self.failUnlessEqual(server_url, actual_url)
  57                 self.failUnlessEqual(delegate, actual_delegate)
  58                 break
  59             else:
  60                 self.fail('Not enough services found')
  61
  62     def _checkServices(self, expectedServices):
  63         """Check to make sure that the expected services are found in
  64         that order in the parsed document."""
  65         it = iter(self._getServices())
  66         for (type_uri, uri) in expectedServices:
  67             for service in it:
  68                 if type_uri in service.type_uris:
  69                     self.failUnlessEqual(service.uri, uri)
  70                     break
  71             else:
  72                 self.fail('Did not find %r service' % (type_uri,))
  73
  74     def testGetSeveral(self):
  75         """Get some services in order"""
  76         expectedServices = [
  77             # type, URL
  78             (TYPEKEY_1_0, None),
  79             (LID_2_0, "http://mylid.net/josh"),
  80             ]
  81
  82         self._checkServices(expectedServices)
  83
  84     def testGetSeveralForOne(self):
  85         """Getting services for one Service with several Type elements."""
  86         types = [ 'http://lid.netmesh.org/sso/2.0b5'
  87                 , 'http://lid.netmesh.org/2.0b5'
  88                 ]
  89
  90         uri = "http://mylid.net/josh"
  91
  92         for service in self._getServices():
  93             if service.uri == uri:
  94                 found_types = service.matchTypes(types)
  95                 if found_types == types:
  96                     break
  97         else:
  98             self.fail('Did not find service with expected types and uris')
  99
 100     def testNoXRDS(self):
 101         """Make sure that we get an exception when an XRDS element is
 102         not present"""
 103         self.xmldoc = file(NOXRDS_FILE).read()
 104         self.failUnlessRaises(
 105             etxrd.XRDSError,
 106             services.applyFilter, self.yadis_url, self.xmldoc, None)
 107
 108     def testEmpty(self):
 109         """Make sure that we get an exception when an XRDS element is
 110         not present"""
 111         self.xmldoc = ''
 112         self.failUnlessRaises(
 113             etxrd.XRDSError,
 114             services.applyFilter, self.yadis_url, self.xmldoc, None)
 115
 116     def testNoXRD(self):
 117         """Make sure that we get an exception when there is no XRD
 118         element present."""
 119         self.xmldoc = file(NOXRD_FILE).read()
 120         self.failUnlessRaises(
 121             etxrd.XRDSError,
 122             services.applyFilter, self.yadis_url, self.xmldoc, None)
 123
 124
 125 class TestCanonicalID(unittest.TestCase):
 126
 127     def mkTest(iname, filename, expectedID):
 128         """This function builds a method that runs the CanonicalID
 129         test for the given set of inputs"""
 130
 131         filename = datapath(filename)
 132         def test(self):
 133             xrds = etxrd.parseXRDS(file(filename).read())
 134             self._getCanonicalID(iname, xrds, expectedID)
 135         return test
 136
 137     test_delegated = mkTest(
 138         "@ootao*test1", "delegated-20060809.xrds",
 139         "@!5BAD.2AA.3C72.AF46!0000.0000.3B9A.CA01")
 140
 141     test_delegated_r1 = mkTest(
 142         "@ootao*test1", "delegated-20060809-r1.xrds",
 143         "@!5BAD.2AA.3C72.AF46!0000.0000.3B9A.CA01")
 144
 145     test_delegated_r2 = mkTest(
 146         "@ootao*test1", "delegated-20060809-r2.xrds",
 147         "@!5BAD.2AA.3C72.AF46!0000.0000.3B9A.CA01")
 148
 149     test_sometimesprefix = mkTest(
 150         "@ootao*test1", "sometimesprefix.xrds",
 151         "@!5BAD.2AA.3C72.AF46!0000.0000.3B9A.CA01")
 152
 153     test_prefixsometimes = mkTest(
 154         "@ootao*test1", "prefixsometimes.xrds",
 155         "@!5BAD.2AA.3C72.AF46!0000.0000.3B9A.CA01")
 156
 157     test_spoof1 = mkTest("=keturn*isDrummond", "spoof1.xrds", etxrd.XRDSFraud)
 158
 159     test_spoof2 = mkTest("=keturn*isDrummond", "spoof2.xrds", etxrd.XRDSFraud)
 160
 161     test_spoof3 = mkTest("@keturn*is*drummond", "spoof3.xrds", etxrd.XRDSFraud)
 162
 163     test_status222 = mkTest("=x", "status222.xrds", None)
 164
 165     test_multisegment_xri = mkTest('xri://=nishitani*masaki',
 166                                    'subsegments.xrds',
 167                                    '=!E117.EF2F.454B.C707!0000.0000.3B9A.CA01')
 168
 169     test_iri_auth_not_allowed = mkTest(
 170         "phreak.example.com", "delegated-20060809-r2.xrds", etxrd.XRDSFraud)
 171     test_iri_auth_not_allowed.__doc__ = \
 172         "Don't let IRI authorities be canonical for the GCS."
 173
 174     # TODO: Refs
 175     # test_ref = mkTest("@ootao*test.ref", "ref.xrds", "@!BAE.A650.823B.2475")
 176
 177     # TODO: Add a IRI authority with an IRI canonicalID.
 178     # TODO: Add test cases with real examples of multiple CanonicalIDs
 179     #   somewhere in the resolution chain.
 180
 181     def _getCanonicalID(self, iname, xrds, expectedID):
 182         if isinstance(expectedID, (str, unicode, type(None))):
 183             cid = etxrd.getCanonicalID(iname, xrds)
 184             self.failUnlessEqual(cid, expectedID and xri.XRI(expectedID))
 185         elif issubclass(expectedID, etxrd.XRDSError):
 186             self.failUnlessRaises(expectedID, etxrd.getCanonicalID,
 187                                   iname, xrds)
 188         else:
 189             self.fail("Don't know how to test for expected value %r"
 190                       % (expectedID,))
 191
 192
 193 if __name__ == '__main__':
 194     unittest.main()