libs/openid/test/test_yadis_discover.py

   1 #!/usr/bin/env python
   2
   3 """Tests for yadis.discover.
   4
   5 @todo: Now that yadis.discover uses urljr.fetchers, we should be able to do
   6    tests with a mock fetcher instead of spawning threads with BaseHTTPServer.
   7 """
   8
   9 import unittest
  10 import urlparse
  11 import re
  12 import types
  13
  14 from openid.yadis.discover import discover, DiscoveryFailure
  15
  16 from openid import fetchers
  17
  18 import discoverdata
  19
  20 status_header_re = re.compile(r'Status: (\d+) .*?$', re.MULTILINE)
  21
  22 four04_pat = """\
  23 Content-Type: text/plain
  24
  25 No such file %s
  26 """
  27
  28 class QuitServer(Exception): pass
  29
  30 def mkResponse(data):
  31     status_mo = status_header_re.match(data)
  32     headers_str, body = data.split('\n\n', 1)
  33     headers = {}
  34     for line in headers_str.split('\n'):
  35         k, v = line.split(':', 1)
  36         k = k.strip().lower()
  37         v = v.strip()
  38         headers[k] = v
  39     status = int(status_mo.group(1))
  40     return fetchers.HTTPResponse(status=status,
  41                                  headers=headers,
  42                                  body=body)
  43
  44 class TestFetcher(object):
  45     def __init__(self, base_url):
  46         self.base_url = base_url
  47
  48     def fetch(self, url, headers, body):
  49         current_url = url
  50         while True:
  51             parsed = urlparse.urlparse(current_url)
  52             path = parsed[2][1:]
  53             try:
  54                 data = discoverdata.generateSample(path, self.base_url)
  55             except KeyError:
  56                 return fetchers.HTTPResponse(status=404,
  57                                              final_url=current_url,
  58                                              headers={},
  59                                              body='')
  60
  61             response = mkResponse(data)
  62             if response.status in [301, 302, 303, 307]:
  63                 current_url = response.headers['location']
  64             else:
  65                 response.final_url = current_url
  66                 return response
  67
  68 class TestSecondGet(unittest.TestCase):
  69     class MockFetcher(object):
  70         def __init__(self):
  71             self.count = 0
  72         def fetch(self, uri, headers=None, body=None):
  73             self.count += 1
  74             if self.count == 1:
  75                 headers = {
  76                     'X-XRDS-Location'.lower(): 'http://unittest/404',
  77                     }
  78                 return fetchers.HTTPResponse(uri, 200, headers, '')
  79             else:
  80                 return fetchers.HTTPResponse(uri, 404)
  81
  82     def setUp(self):
  83         self.oldfetcher = fetchers.getDefaultFetcher()
  84         fetchers.setDefaultFetcher(self.MockFetcher())
  85
  86     def tearDown(self):
  87         fetchers.setDefaultFetcher(self.oldfetcher)
  88
  89     def test_404(self):
  90         uri = "http://something.unittest/"
  91         self.failUnlessRaises(DiscoveryFailure, discover, uri)
  92
  93
  94 class _TestCase(unittest.TestCase):
  95     base_url = 'http://invalid.unittest/'
  96
  97     def __init__(self, input_name, id_name, result_name, success):
  98         self.input_name = input_name
  99         self.id_name = id_name
 100         self.result_name = result_name
 101         self.success = success
 102         # Still not quite sure how to best construct these custom tests.
 103         # Between python2.3 and python2.4, a patch attached to pyunit.sf.net
 104         # bug #469444 got applied which breaks loadTestsFromModule on this
 105         # class if it has test_ or runTest methods.  So, kludge to change
 106         # the method name.
 107         unittest.TestCase.__init__(self, methodName='runCustomTest')
 108
 109     def setUp(self):
 110         fetchers.setDefaultFetcher(TestFetcher(self.base_url),
 111                                    wrap_exceptions=False)
 112
 113         self.input_url, self.expected = discoverdata.generateResult(
 114             self.base_url,
 115             self.input_name,
 116             self.id_name,
 117             self.result_name,
 118             self.success)
 119
 120     def tearDown(self):
 121         fetchers.setDefaultFetcher(None)
 122
 123     def runCustomTest(self):
 124         if self.expected is DiscoveryFailure:
 125             self.failUnlessRaises(DiscoveryFailure,
 126                                   discover, self.input_url)
 127         else:
 128             result = discover(self.input_url)
 129             self.failUnlessEqual(self.input_url, result.request_uri)
 130
 131             msg = 'Identity URL mismatch: actual = %r, expected = %r' % (
 132                 result.normalized_uri, self.expected.normalized_uri)
 133             self.failUnlessEqual(
 134                 self.expected.normalized_uri, result.normalized_uri, msg)
 135
 136             msg = 'Content mismatch: actual = %r, expected = %r' % (
 137                 result.response_text, self.expected.response_text)
 138             self.failUnlessEqual(
 139                 self.expected.response_text, result.response_text, msg)
 140
 141             expected_keys = dir(self.expected)
 142             expected_keys.sort()
 143             actual_keys = dir(result)
 144             actual_keys.sort()
 145             self.failUnlessEqual(actual_keys, expected_keys)
 146
 147             for k in dir(self.expected):
 148                 if k.startswith('__') and k.endswith('__'):
 149                     continue
 150                 exp_v = getattr(self.expected, k)
 151                 if isinstance(exp_v, types.MethodType):
 152                     continue
 153                 act_v = getattr(result, k)
 154                 assert act_v == exp_v, (k, exp_v, act_v)
 155
 156     def shortDescription(self):
 157         try:
 158             n = self.input_url
 159         except AttributeError:
 160             # run before setUp, or if setUp did not complete successfully.
 161             n = self.input_name
 162         return "%s (%s)" % (
 163             n,
 164             self.__class__.__module__)
 165
 166 def pyUnitTests():
 167     s = unittest.TestSuite()
 168     for success, input_name, id_name, result_name in discoverdata.testlist:
 169         test = _TestCase(input_name, id_name, result_name, success)
 170         s.addTest(test)
 171
 172     return s
 173
 174 def test():
 175     runner = unittest.TextTestRunner()
 176     return runner.run(loadTests())
 177
 178 if __name__ == '__main__':
 179     test()