Package yadis :: Package test :: Module test_discover
[hide private]
[frames] | no frames]

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