Source Code for Module openid.test.test_yadis_discover

  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() 
180