~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

• browse files at revision 4842
• compare with another revision
• download diff
• download tarball
• view history from revision 4842

• reverse the comparison (4842 to 4443.2.5)
• stop comparing with revision 4443.2.5

Show diffs side-by-side

added

removed

bzrlib/tests/test_http.py

204

204

    It records the bytes sent to it, and replies with a 200.

205

205

    """

206

206

 

207

 

    def __init__(self, expect_body_tail=None):

 

207

    def __init__(self, expect_body_tail=None, scheme=''):

208

208

        """Constructor.

209

209

 

210

210

        :type expect_body_tail: str

215

215

        self.host = None

216

216

        self.port = None

217

217

        self.received_bytes = ''

 

218

        self.scheme = scheme

 

219

 

 

220

    def get_url(self):

 

221

        return '%s://%s:%s/' % (self.scheme, self.host, self.port)

218

222

 

219

223

    def setUp(self):

220

224

        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

304

308

 

305

309

        server = http_server.HttpServer(BogusRequestHandler)

306

310

        try:

307

 

            self.assertRaises(httplib.UnknownProtocol,server.setUp)

 

311

            self.assertRaises(httplib.UnknownProtocol, server.setUp)

308

312

        except:

309

313

            server.tearDown()

310

314

            self.fail('HTTP Server creation did not raise UnknownProtocol')

312

316

    def test_force_invalid_protocol(self):

313

317

        server = http_server.HttpServer(protocol_version='HTTP/0.1')

314

318

        try:

315

 

            self.assertRaises(httplib.UnknownProtocol,server.setUp)

 

319

            self.assertRaises(httplib.UnknownProtocol, server.setUp)

316

320

        except:

317

321

            server.tearDown()

318

322

            self.fail('HTTP Server creation did not raise UnknownProtocol')

320

324

    def test_server_start_and_stop(self):

321

325

        server = http_server.HttpServer()

322

326

        server.setUp()

323

 

        self.assertTrue(server._http_running)

324

 

        server.tearDown()

 

327

        try:

 

328

            self.assertTrue(server._http_running)

 

329

        finally:

 

330

            server.tearDown()

325

331

        self.assertFalse(server._http_running)

326

332

 

327

333

    def test_create_http_server_one_zero(self):

330

336

            protocol_version = 'HTTP/1.0'

331

337

 

332

338

        server = http_server.HttpServer(RequestHandlerOneZero)

333

 

        server.setUp()

334

 

        self.addCleanup(server.tearDown)

 

339

        self.start_server(server)

335

340

        self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)

336

341

 

337

342

    def test_create_http_server_one_one(self):

340

345

            protocol_version = 'HTTP/1.1'

341

346

 

342

347

        server = http_server.HttpServer(RequestHandlerOneOne)

343

 

        server.setUp()

344

 

        self.addCleanup(server.tearDown)

 

348

        self.start_server(server)

345

349

        self.assertIsInstance(server._httpd,

346

350

                              http_server.TestingThreadingHTTPServer)

347

351

 

352

356

 

353

357

        server = http_server.HttpServer(RequestHandlerOneZero,

354

358

                                        protocol_version='HTTP/1.1')

355

 

        server.setUp()

356

 

        self.addCleanup(server.tearDown)

 

359

        self.start_server(server)

357

360

        self.assertIsInstance(server._httpd,

358

361

                              http_server.TestingThreadingHTTPServer)

359

362

 

364

367

 

365

368

        server = http_server.HttpServer(RequestHandlerOneOne,

366

369

                                        protocol_version='HTTP/1.0')

367

 

        server.setUp()

368

 

        self.addCleanup(server.tearDown)

 

370

        self.start_server(server)

369

371

        self.assertIsInstance(server._httpd,

370

372

                              http_server.TestingHTTPServer)

371

373

 

431

433

    def test_http_impl_urls(self):

432

434

        """There are servers which ask for particular clients to connect"""

433

435

        server = self._server()

 

436

        server.setUp()

434

437

        try:

435

 

            server.setUp()

436

438

            url = server.get_url()

437

439

            self.assertTrue(url.startswith('%s://' % self._qualified_prefix))

438

440

        finally:

543

545

class TestPost(tests.TestCase):

544

546

 

545

547

    def test_post_body_is_received(self):

546

 

        server = RecordingServer(expect_body_tail='end-of-body')

547

 

        server.setUp()

548

 

        self.addCleanup(server.tearDown)

549

 

        scheme = self._qualified_prefix

550

 

        url = '%s://%s:%s/' % (scheme, server.host, server.port)

 

548

        server = RecordingServer(expect_body_tail='end-of-body',

 

549

            scheme=self._qualified_prefix)

 

550

        self.start_server(server)

 

551

        url = server.get_url()

551

552

        http_transport = self._transport(url)

552

553

        code, response = http_transport._post('abc def end-of-body')

553

554

        self.assertTrue(

779

780

        self.assertEqual(None, server.port)

780

781

 

781

782

    def test_send_receive_bytes(self):

782

 

        server = RecordingServer(expect_body_tail='c')

783

 

        server.setUp()

784

 

        self.addCleanup(server.tearDown)

 

783

        server = RecordingServer(expect_body_tail='c', scheme='http')

 

784

        self.start_server(server)

785

785

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

786

786

        sock.connect((server.host, server.port))

787

787

        sock.sendall('abc')

1178

1178

        if self._testing_pycurl():

1179

1179

            # Oh my ! pycurl does not check for the port as part of

1180

1180

            # no_proxy :-( So we just test the host part

1181

 

            self.no_proxy_host = 'localhost'

 

1181

            self.no_proxy_host = self.server.host

1182

1182

        else:

1183

1183

            self.no_proxy_host = self.proxy_address

1184

1184

        # The secondary server is the proxy

1956

1956

        pass

1957

1957

 

1958

1958

 

1959

 

class TestActivity(tests.TestCase):

 

1959

class TestActivityMixin(object):

1960

1960

    """Test socket activity reporting.

1961

1961

 

1962

1962

    We use a special purpose server to control the bytes sent and received and

2100

2100

        code, f = t._post('abc def end-of-body\n')

2101

2101

        self.assertEqual('lalala whatever as long as itsssss\n', f.read())

2102

2102

        self.assertActivitiesMatch()

 

2103

 

 

2104

 

 

2105

class TestActivity(tests.TestCase, TestActivityMixin):

 

2106

 

 

2107

    def setUp(self):

 

2108

        tests.TestCase.setUp(self)

 

2109

        self.server = self._activity_server(self._protocol_version)

 

2110

        self.server.setUp()

 

2111

        self.activities = {}

 

2112

        def report_activity(t, bytes, direction):

 

2113

            count = self.activities.get(direction, 0)

 

2114

            count += bytes

 

2115

            self.activities[direction] = count

 

2116

 

 

2117

        # We override at class level because constructors may propagate the

 

2118

        # bound method and render instance overriding ineffective (an

 

2119

        # alternative would be to define a specific ui factory instead...)

 

2120

        self.orig_report_activity = self._transport._report_activity

 

2121

        self._transport._report_activity = report_activity

 

2122

 

 

2123

    def tearDown(self):

 

2124

        self._transport._report_activity = self.orig_report_activity

 

2125

        self.server.tearDown()

 

2126

        tests.TestCase.tearDown(self)

 

2127

 

 

2128

 

 

2129

class TestNoReportActivity(tests.TestCase, TestActivityMixin):

 

2130

 

 

2131

    def setUp(self):

 

2132

        tests.TestCase.setUp(self)

 

2133

        # Unlike TestActivity, we are really testing ReportingFileSocket and

 

2134

        # ReportingSocket, so we don't need all the parametrization. Since

 

2135

        # ReportingFileSocket and ReportingSocket are wrappers, it's easier to

 

2136

        # test them through their use by the transport than directly (that's a

 

2137

        # bit less clean but far more simpler and effective).

 

2138

        self.server = ActivityHTTPServer('HTTP/1.1')

 

2139

        self._transport=_urllib.HttpTransport_urllib

 

2140

 

 

2141

        self.server.setUp()

 

2142

 

 

2143

        # We override at class level because constructors may propagate the

 

2144

        # bound method and render instance overriding ineffective (an

 

2145

        # alternative would be to define a specific ui factory instead...)

 

2146

        self.orig_report_activity = self._transport._report_activity

 

2147

        self._transport._report_activity = None

 

2148

 

 

2149

    def tearDown(self):

 

2150

        self._transport._report_activity = self.orig_report_activity

 

2151

        self.server.tearDown()

 

2152

        tests.TestCase.tearDown(self)

 

2153

 

 

2154

    def assertActivitiesMatch(self):

 

2155

        # Nothing to check here

 

2156

        pass

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1