1
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
28
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
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
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
82 self.oldfetcher = fetchers.getDefaultFetcher()
83 fetchers.setDefaultFetcher(self.MockFetcher())
84
86 fetchers.setDefaultFetcher(self.oldfetcher)
87
91
92
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
102
103
104
105
106 unittest.TestCase.__init__(self, methodName='runCustomTest')
107
118
120 fetchers.setDefaultFetcher(None)
121
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
151 try:
152 n = self.input_url
153 except AttributeError:
154
155 n = self.input_name
156 return "%s (%s)" % (
157 n,
158 self.__class__.__module__)
159
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
171
172 if __name__ == '__main__':
173 test()
174