/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Matt Nordhoff
  • Date: 2009-04-04 02:50:01 UTC
  • mfrom: (4253 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4256.
  • Revision ID: mnordhoff@mattnordhoff.com-20090404025001-z1403k0tatmc8l91
Merge bzr.dev, fixing conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Tests for HTTP implementations.
18
18
 
71
71
def load_tests(standard_tests, module, loader):
72
72
    """Multiply tests for http clients and protocol versions."""
73
73
    result = loader.suiteClass()
74
 
    adapter = tests.TestScenarioApplier()
75
 
    remaining_tests = standard_tests
76
74
 
77
 
    # one for each transport
 
75
    # one for each transport implementation
78
76
    t_tests, remaining_tests = tests.split_suite_by_condition(
79
 
        remaining_tests, tests.condition_isinstance((
 
77
        standard_tests, tests.condition_isinstance((
80
78
                TestHttpTransportRegistration,
81
79
                TestHttpTransportUrls,
82
80
                Test_redirected_to,
91
89
            ('pycurl', dict(_transport=PyCurlTransport,
92
90
                            _server=http_server.HttpServer_PyCurl,
93
91
                            _qualified_prefix='http+pycurl',)))
94
 
    adapter.scenarios = transport_scenarios
95
 
    tests.adapt_tests(t_tests, adapter, result)
 
92
    tests.multiply_tests(t_tests, transport_scenarios, result)
96
93
 
97
 
    # multiplied by one for each protocol version
 
94
    # each implementation tested with each HTTP version
98
95
    tp_tests, remaining_tests = tests.split_suite_by_condition(
99
96
        remaining_tests, tests.condition_isinstance((
100
97
                SmartHTTPTunnellingTest,
112
109
            ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
113
110
            ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
114
111
            ]
115
 
    tp_scenarios = tests.multiply_scenarios(adapter.scenarios,
 
112
    tp_scenarios = tests.multiply_scenarios(transport_scenarios,
116
113
                                            protocol_scenarios)
117
 
    adapter.scenarios = tp_scenarios
118
 
    tests.adapt_tests(tp_tests, adapter, result)
 
114
    tests.multiply_tests(tp_tests, tp_scenarios, result)
119
115
 
120
 
    # multiplied by one for each authentication scheme
 
116
    # auth: each auth scheme on all http versions on all implementations.
121
117
    tpa_tests, remaining_tests = tests.split_suite_by_condition(
122
118
        remaining_tests, tests.condition_isinstance((
123
119
                TestAuth,
126
122
        ('basic', dict(_auth_scheme='basic')),
127
123
        ('digest', dict(_auth_scheme='digest')),
128
124
        ]
129
 
    adapter.scenarios = tests.multiply_scenarios(adapter.scenarios,
130
 
                                                 auth_scheme_scenarios)
131
 
    tests.adapt_tests(tpa_tests, adapter, result)
 
125
    tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
 
126
        auth_scheme_scenarios)
 
127
    tests.multiply_tests(tpa_tests, tpa_scenarios, result)
132
128
 
 
129
    # activity: activity on all http versions on all implementations
133
130
    tpact_tests, remaining_tests = tests.split_suite_by_condition(
134
131
        remaining_tests, tests.condition_isinstance((
135
132
                TestActivity,
139
136
        ]
140
137
    if tests.HTTPSServerFeature.available():
141
138
        activity_scenarios.append(
142
 
            ('https', dict(_activity_server=ActivityHTTPSServer,)))
143
 
    adapter.scenarios = tests.multiply_scenarios(tp_scenarios,
144
 
                                                 activity_scenarios)
145
 
    tests.adapt_tests(tpact_tests, adapter, result)
 
139
            ('https', dict(_activity_server=ActivityHTTPSServer)))
 
140
    tpact_scenarios = tests.multiply_scenarios(tp_scenarios,
 
141
        activity_scenarios)
 
142
    tests.multiply_tests(tpact_tests, tpact_scenarios, result)
146
143
 
147
144
    # No parametrization for the remaining tests
148
145
    result.addTests(remaining_tests)
161
158
 
162
159
class RecordingServer(object):
163
160
    """A fake HTTP server.
164
 
    
 
161
 
165
162
    It records the bytes sent to it, and replies with a 200.
166
163
    """
167
164
 
216
213
        self.port = None
217
214
 
218
215
 
 
216
class TestAuthHeader(tests.TestCase):
 
217
 
 
218
    def parse_header(self, header):
 
219
        ah =  _urllib2_wrappers.AbstractAuthHandler()
 
220
        return ah._parse_auth_header(header)
 
221
 
 
222
    def test_empty_header(self):
 
223
        scheme, remainder = self.parse_header('')
 
224
        self.assertEquals('', scheme)
 
225
        self.assertIs(None, remainder)
 
226
 
 
227
    def test_negotiate_header(self):
 
228
        scheme, remainder = self.parse_header('Negotiate')
 
229
        self.assertEquals('negotiate', scheme)
 
230
        self.assertIs(None, remainder)
 
231
 
 
232
    def test_basic_header(self):
 
233
        scheme, remainder = self.parse_header(
 
234
            'Basic realm="Thou should not pass"')
 
235
        self.assertEquals('basic', scheme)
 
236
        self.assertEquals('realm="Thou should not pass"', remainder)
 
237
 
 
238
    def test_digest_header(self):
 
239
        scheme, remainder = self.parse_header(
 
240
            'Digest realm="Thou should not pass"')
 
241
        self.assertEquals('digest', scheme)
 
242
        self.assertEquals('realm="Thou should not pass"', remainder)
 
243
 
 
244
 
219
245
class TestHTTPServer(tests.TestCase):
220
246
    """Test the HTTP servers implementations."""
221
247
 
800
826
        # bytes on the socket
801
827
        ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
802
828
        self.assertEqual((0, '0'), ireadv.next())
803
 
        # The server should have issued one request so far 
 
829
        # The server should have issued one request so far
804
830
        self.assertEqual(1, server.GET_request_nb)
805
831
        self.assertEqual('0123456789', t.get_bytes('a'))
806
832
        # get_bytes issued an additional request, the readv pending ones are
1281
1307
        # Since the tests using this class will replace
1282
1308
        # _urllib2_wrappers.Request, we can't just call the base class __init__
1283
1309
        # or we'll loop.
1284
 
        RedirectedRequest.init_orig(self, method, url, args, kwargs)
 
1310
        RedirectedRequest.init_orig(self, method, url, *args, **kwargs)
1285
1311
        self.follow_redirections = True
1286
1312
 
1287
1313