test_endtoend.py (27161B)
1 #!/usr/bin/env python 2 # 3 # Copyright 2012, Google Inc. 4 # All rights reserved. 5 # 6 # Redistribution and use in source and binary forms, with or without 7 # modification, are permitted provided that the following conditions are 8 # met: 9 # 10 # * Redistributions of source code must retain the above copyright 11 # notice, this list of conditions and the following disclaimer. 12 # * Redistributions in binary form must reproduce the above 13 # copyright notice, this list of conditions and the following disclaimer 14 # in the documentation and/or other materials provided with the 15 # distribution. 16 # * Neither the name of Google Inc. nor the names of its 17 # contributors may be used to endorse or promote products derived from 18 # this software without specific prior written permission. 19 # 20 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 21 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 22 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 23 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 24 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 25 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 26 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 27 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 28 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 29 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 30 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 31 """End-to-end tests for pywebsocket3. Tests standalone.py. 32 """ 33 34 from __future__ import absolute_import 35 36 import locale 37 import logging 38 import os 39 import socket 40 import subprocess 41 import sys 42 import time 43 import unittest 44 45 from six.moves import urllib 46 47 import set_sys_path # Update sys.path to locate pywebsocket3 module. 48 from test import client_for_testing 49 50 # Special message that tells the echo server to start closing handshake 51 _GOODBYE_MESSAGE = 'Goodbye' 52 53 _SERVER_WARMUP_IN_SEC = 0.2 54 55 56 # Test body functions 57 def _echo_check_procedure(client): 58 client.connect() 59 60 client.send_message('test') 61 client.assert_receive('test') 62 client.send_message('helloworld') 63 client.assert_receive('helloworld') 64 65 client.send_close() 66 client.assert_receive_close() 67 68 client.assert_connection_closed() 69 70 71 def _echo_check_procedure_with_binary(client): 72 client.connect() 73 74 client.send_message(b'binary', binary=True) 75 client.assert_receive(b'binary', binary=True) 76 client.send_message(b'\x00\x80\xfe\xff\x00\x80', binary=True) 77 client.assert_receive(b'\x00\x80\xfe\xff\x00\x80', binary=True) 78 79 client.send_close() 80 client.assert_receive_close() 81 82 client.assert_connection_closed() 83 84 85 def _echo_check_procedure_with_goodbye(client): 86 client.connect() 87 88 client.send_message('test') 89 client.assert_receive('test') 90 91 client.send_message(_GOODBYE_MESSAGE) 92 client.assert_receive(_GOODBYE_MESSAGE) 93 94 client.assert_receive_close() 95 client.send_close() 96 97 client.assert_connection_closed() 98 99 100 def _echo_check_procedure_with_code_and_reason(client, code, reason): 101 client.connect() 102 103 client.send_close(code, reason) 104 client.assert_receive_close(code, reason) 105 106 client.assert_connection_closed() 107 108 109 def _unmasked_frame_check_procedure(client): 110 client.connect() 111 112 client.send_message('test', mask=False) 113 client.assert_receive_close(client_for_testing.STATUS_PROTOCOL_ERROR, '') 114 115 client.assert_connection_closed() 116 117 118 def _check_handshake_with_basic_auth(client): 119 client.connect() 120 121 client.send_message(_GOODBYE_MESSAGE) 122 client.assert_receive(_GOODBYE_MESSAGE) 123 124 client.assert_receive_close() 125 client.send_close() 126 127 client.assert_connection_closed() 128 129 130 class EndToEndTestBase(unittest.TestCase): 131 """Base class for end-to-end tests that launch pywebsocket standalone 132 server as a separate process, connect to it using the client_for_testing 133 module, and check if the server behaves correctly by exchanging opening 134 handshake and frames over a TCP connection. 135 """ 136 def setUp(self): 137 self.server_stderr = None 138 self.top_dir = os.path.join(os.path.dirname(__file__), '..') 139 os.putenv('PYTHONPATH', os.path.pathsep.join(sys.path)) 140 self.standalone_command = os.path.join(self.top_dir, 'pywebsocket3', 141 'standalone.py') 142 self.document_root = os.path.join(self.top_dir, 'example') 143 s = socket.socket() 144 s.bind(('localhost', 0)) 145 (_, self.test_port) = s.getsockname() 146 s.close() 147 148 self._options = client_for_testing.ClientOptions() 149 self._options.server_host = 'localhost' 150 self._options.origin = 'http://localhost' 151 self._options.resource = '/echo' 152 153 self._options.server_port = self.test_port 154 155 # TODO(tyoshino): Use tearDown to kill the server. 156 157 def _run_python_command(self, commandline, stdout=None, stderr=None): 158 close_fds = True if sys.platform != 'win32' else None 159 return subprocess.Popen([sys.executable] + commandline, 160 close_fds=close_fds, 161 stdout=stdout, 162 stderr=stderr) 163 164 def _run_server(self, extra_args=[]): 165 args = [ 166 self.standalone_command, '-H', 'localhost', '-V', 'localhost', 167 '-p', 168 str(self.test_port), '-P', 169 str(self.test_port), '-d', self.document_root 170 ] 171 172 # Inherit the level set to the root logger by test runner. 173 root_logger = logging.getLogger() 174 log_level = root_logger.getEffectiveLevel() 175 if log_level != logging.NOTSET: 176 args.append('--log-level') 177 args.append(logging.getLevelName(log_level).lower()) 178 179 args += extra_args 180 181 return self._run_python_command(args, stderr=self.server_stderr) 182 183 def _close_server(self, server): 184 """ 185 186 This method mimics Popen.__exit__ to gracefully kill the server process. 187 Its main purpose is to maintain comptaibility between python 2 and 3, 188 since Popen in python 2 does not have __exit__ attribute. 189 190 """ 191 server.kill() 192 193 if server.stdout: 194 server.stdout.close() 195 if server.stderr: 196 server.stderr.close() 197 if server.stdin: 198 server.stdin.close() 199 200 server.wait() 201 202 203 class EndToEndHyBiTest(EndToEndTestBase): 204 def setUp(self): 205 EndToEndTestBase.setUp(self) 206 207 def _run_test_with_options(self, 208 test_function, 209 options, 210 server_options=[]): 211 server = self._run_server(server_options) 212 try: 213 # TODO(tyoshino): add some logic to poll the server until it 214 # becomes ready 215 time.sleep(_SERVER_WARMUP_IN_SEC) 216 217 client = client_for_testing.create_client(options) 218 try: 219 test_function(client) 220 finally: 221 client.close_socket() 222 finally: 223 self._close_server(server) 224 225 def _run_test(self, test_function): 226 self._run_test_with_options(test_function, self._options) 227 228 def _run_permessage_deflate_test(self, offer, response_checker, 229 test_function): 230 server = self._run_server() 231 try: 232 time.sleep(_SERVER_WARMUP_IN_SEC) 233 234 self._options.extensions += offer 235 self._options.check_permessage_deflate = response_checker 236 client = client_for_testing.create_client(self._options) 237 238 try: 239 client.connect() 240 241 if test_function is not None: 242 test_function(client) 243 244 client.assert_connection_closed() 245 finally: 246 client.close_socket() 247 finally: 248 self._close_server(server) 249 250 def _run_close_with_code_and_reason_test(self, 251 test_function, 252 code, 253 reason, 254 server_options=[]): 255 server = self._run_server() 256 try: 257 time.sleep(_SERVER_WARMUP_IN_SEC) 258 259 client = client_for_testing.create_client(self._options) 260 try: 261 test_function(client, code, reason) 262 finally: 263 client.close_socket() 264 finally: 265 self._close_server(server) 266 267 def _run_http_fallback_test(self, options, status): 268 server = self._run_server() 269 try: 270 time.sleep(_SERVER_WARMUP_IN_SEC) 271 272 client = client_for_testing.create_client(options) 273 try: 274 client.connect() 275 self.fail('Could not catch HttpStatusException') 276 except client_for_testing.HttpStatusException as e: 277 self.assertEqual(status, e.status) 278 except Exception as e: 279 self.fail('Catch unexpected exception') 280 finally: 281 client.close_socket() 282 finally: 283 self._close_server(server) 284 285 def test_echo(self): 286 self._run_test(_echo_check_procedure) 287 288 def test_echo_binary(self): 289 self._run_test(_echo_check_procedure_with_binary) 290 291 def test_echo_server_close(self): 292 self._run_test(_echo_check_procedure_with_goodbye) 293 294 def test_unmasked_frame(self): 295 self._run_test(_unmasked_frame_check_procedure) 296 297 def test_echo_permessage_deflate(self): 298 def test_function(client): 299 # From the examples in the spec. 300 compressed_hello = b'\xf2\x48\xcd\xc9\xc9\x07\x00' 301 client._stream.send_data(compressed_hello, 302 client_for_testing.OPCODE_TEXT, 303 rsv1=1) 304 client._stream.assert_receive_binary( 305 compressed_hello, 306 opcode=client_for_testing.OPCODE_TEXT, 307 rsv1=1) 308 309 client.send_close() 310 client.assert_receive_close() 311 312 def response_checker(parameter): 313 self.assertEqual('permessage-deflate', parameter.name()) 314 self.assertEqual([], parameter.get_parameters()) 315 316 self._run_permessage_deflate_test(['permessage-deflate'], 317 response_checker, test_function) 318 319 def test_echo_permessage_deflate_two_frames(self): 320 def test_function(client): 321 # From the examples in the spec. 322 client._stream.send_data(b'\xf2\x48\xcd', 323 client_for_testing.OPCODE_TEXT, 324 end=False, 325 rsv1=1) 326 client._stream.send_data(b'\xc9\xc9\x07\x00', 327 client_for_testing.OPCODE_TEXT) 328 client._stream.assert_receive_binary( 329 b'\xf2\x48\xcd\xc9\xc9\x07\x00', 330 opcode=client_for_testing.OPCODE_TEXT, 331 rsv1=1) 332 333 client.send_close() 334 client.assert_receive_close() 335 336 def response_checker(parameter): 337 self.assertEqual('permessage-deflate', parameter.name()) 338 self.assertEqual([], parameter.get_parameters()) 339 340 self._run_permessage_deflate_test(['permessage-deflate'], 341 response_checker, test_function) 342 343 def test_echo_permessage_deflate_two_messages(self): 344 def test_function(client): 345 # From the examples in the spec. 346 client._stream.send_data(b'\xf2\x48\xcd\xc9\xc9\x07\x00', 347 client_for_testing.OPCODE_TEXT, 348 rsv1=1) 349 client._stream.send_data(b'\xf2\x00\x11\x00\x00', 350 client_for_testing.OPCODE_TEXT, 351 rsv1=1) 352 client._stream.assert_receive_binary( 353 b'\xf2\x48\xcd\xc9\xc9\x07\x00', 354 opcode=client_for_testing.OPCODE_TEXT, 355 rsv1=1) 356 client._stream.assert_receive_binary( 357 b'\xf2\x00\x11\x00\x00', 358 opcode=client_for_testing.OPCODE_TEXT, 359 rsv1=1) 360 361 client.send_close() 362 client.assert_receive_close() 363 364 def response_checker(parameter): 365 self.assertEqual('permessage-deflate', parameter.name()) 366 self.assertEqual([], parameter.get_parameters()) 367 368 self._run_permessage_deflate_test(['permessage-deflate'], 369 response_checker, test_function) 370 371 def test_echo_permessage_deflate_two_msgs_server_no_context_takeover(self): 372 def test_function(client): 373 # From the examples in the spec. 374 client._stream.send_data(b'\xf2\x48\xcd\xc9\xc9\x07\x00', 375 client_for_testing.OPCODE_TEXT, 376 rsv1=1) 377 client._stream.send_data(b'\xf2\x00\x11\x00\x00', 378 client_for_testing.OPCODE_TEXT, 379 rsv1=1) 380 client._stream.assert_receive_binary( 381 b'\xf2\x48\xcd\xc9\xc9\x07\x00', 382 opcode=client_for_testing.OPCODE_TEXT, 383 rsv1=1) 384 client._stream.assert_receive_binary( 385 b'\xf2\x48\xcd\xc9\xc9\x07\x00', 386 opcode=client_for_testing.OPCODE_TEXT, 387 rsv1=1) 388 389 client.send_close() 390 client.assert_receive_close() 391 392 def response_checker(parameter): 393 self.assertEqual('permessage-deflate', parameter.name()) 394 self.assertEqual([('server_no_context_takeover', None)], 395 parameter.get_parameters()) 396 397 self._run_permessage_deflate_test( 398 ['permessage-deflate; server_no_context_takeover'], 399 response_checker, test_function) 400 401 def test_echo_permessage_deflate_preference(self): 402 def test_function(client): 403 # From the examples in the spec. 404 compressed_hello = b'\xf2\x48\xcd\xc9\xc9\x07\x00' 405 client._stream.send_data(compressed_hello, 406 client_for_testing.OPCODE_TEXT, 407 rsv1=1) 408 client._stream.assert_receive_binary( 409 compressed_hello, 410 opcode=client_for_testing.OPCODE_TEXT, 411 rsv1=1) 412 413 client.send_close() 414 client.assert_receive_close() 415 416 def response_checker(parameter): 417 self.assertEqual('permessage-deflate', parameter.name()) 418 self.assertEqual([], parameter.get_parameters()) 419 420 self._run_permessage_deflate_test( 421 ['permessage-deflate', 'deflate-frame'], response_checker, 422 test_function) 423 424 def test_echo_permessage_deflate_with_parameters(self): 425 def test_function(client): 426 # From the examples in the spec. 427 compressed_hello = b'\xf2\x48\xcd\xc9\xc9\x07\x00' 428 client._stream.send_data(compressed_hello, 429 client_for_testing.OPCODE_TEXT, 430 rsv1=1) 431 client._stream.assert_receive_binary( 432 compressed_hello, 433 opcode=client_for_testing.OPCODE_TEXT, 434 rsv1=1) 435 436 client.send_close() 437 client.assert_receive_close() 438 439 def response_checker(parameter): 440 self.assertEqual('permessage-deflate', parameter.name()) 441 self.assertEqual([('server_max_window_bits', '10'), 442 ('server_no_context_takeover', None)], 443 parameter.get_parameters()) 444 445 self._run_permessage_deflate_test([ 446 'permessage-deflate; server_max_window_bits=10; ' 447 'server_no_context_takeover' 448 ], response_checker, test_function) 449 450 def test_echo_permessage_deflate_with_bad_server_max_window_bits(self): 451 def test_function(client): 452 client.send_close() 453 client.assert_receive_close() 454 455 def response_checker(parameter): 456 raise Exception('Unexpected acceptance of permessage-deflate') 457 458 self._run_permessage_deflate_test( 459 ['permessage-deflate; server_max_window_bits=3000000'], 460 response_checker, test_function) 461 462 def test_echo_permessage_deflate_with_bad_server_max_window_bits(self): 463 def test_function(client): 464 client.send_close() 465 client.assert_receive_close() 466 467 def response_checker(parameter): 468 raise Exception('Unexpected acceptance of permessage-deflate') 469 470 self._run_permessage_deflate_test( 471 ['permessage-deflate; server_max_window_bits=3000000'], 472 response_checker, test_function) 473 474 def test_echo_permessage_deflate_with_undefined_parameter(self): 475 def test_function(client): 476 client.send_close() 477 client.assert_receive_close() 478 479 def response_checker(parameter): 480 raise Exception('Unexpected acceptance of permessage-deflate') 481 482 self._run_permessage_deflate_test(['permessage-deflate; foo=bar'], 483 response_checker, test_function) 484 485 def test_echo_close_with_code_and_reason(self): 486 self._options.resource = '/close' 487 self._run_close_with_code_and_reason_test( 488 _echo_check_procedure_with_code_and_reason, 3333, 'sunsunsunsun') 489 490 def test_echo_close_with_empty_body(self): 491 self._options.resource = '/close' 492 self._run_close_with_code_and_reason_test( 493 _echo_check_procedure_with_code_and_reason, None, '') 494 495 def test_close_on_protocol_error(self): 496 """Tests that the server sends a close frame with protocol error status 497 code when the client sends data with some protocol error. 498 """ 499 def test_function(client): 500 client.connect() 501 502 # Intermediate frame without any preceding start of fragmentation 503 # frame. 504 client.send_frame_of_arbitrary_bytes(b'\x80\x80', '') 505 client.assert_receive_close( 506 client_for_testing.STATUS_PROTOCOL_ERROR) 507 508 self._run_test(test_function) 509 510 def test_close_on_unsupported_frame(self): 511 """Tests that the server sends a close frame with unsupported operation 512 status code when the client sends data asking some operation that is 513 not supported by the server. 514 """ 515 def test_function(client): 516 client.connect() 517 518 # Text frame with RSV3 bit raised. 519 client.send_frame_of_arbitrary_bytes(b'\x91\x80', '') 520 client.assert_receive_close( 521 client_for_testing.STATUS_UNSUPPORTED_DATA) 522 523 self._run_test(test_function) 524 525 def test_close_on_invalid_frame(self): 526 """Tests that the server sends a close frame with invalid frame payload 527 data status code when the client sends an invalid frame like containing 528 invalid UTF-8 character. 529 """ 530 def test_function(client): 531 client.connect() 532 533 # Text frame with invalid UTF-8 string. 534 client.send_message(b'\x80', raw=True) 535 client.assert_receive_close( 536 client_for_testing.STATUS_INVALID_FRAME_PAYLOAD_DATA) 537 538 self._run_test(test_function) 539 540 def test_close_on_internal_endpoint_error(self): 541 """Tests that the server sends a close frame with internal endpoint 542 error status code when the handler does bad operation. 543 """ 544 545 self._options.resource = '/internal_error' 546 547 def test_function(client): 548 client.connect() 549 client.assert_receive_close( 550 client_for_testing.STATUS_INTERNAL_ENDPOINT_ERROR) 551 552 self._run_test(test_function) 553 554 def test_absolute_uri(self): 555 """Tests absolute uri request.""" 556 557 options = self._options 558 options.resource = 'ws://localhost:%d/echo' % options.server_port 559 self._run_test_with_options(_echo_check_procedure, options) 560 561 def test_invalid_absolute_uri(self): 562 """Tests invalid absolute uri request.""" 563 564 options = self._options 565 options.resource = 'ws://invalidlocalhost:%d/echo' % options.server_port 566 options.server_stderr = subprocess.PIPE 567 568 self._run_http_fallback_test(options, 404) 569 570 def test_origin_check(self): 571 """Tests http fallback on origin check fail.""" 572 573 options = self._options 574 options.resource = '/origin_check' 575 # Server shows warning message for http 403 fallback. This warning 576 # message is confusing. Following pipe disposes warning messages. 577 self.server_stderr = subprocess.PIPE 578 self._run_http_fallback_test(options, 403) 579 580 def test_invalid_resource(self): 581 """Tests invalid resource path.""" 582 583 options = self._options 584 options.resource = '/no_resource' 585 586 self.server_stderr = subprocess.PIPE 587 self._run_http_fallback_test(options, 404) 588 589 def test_fragmentized_resource(self): 590 """Tests resource name with fragment""" 591 592 options = self._options 593 options.resource = '/echo#fragment' 594 595 self.server_stderr = subprocess.PIPE 596 self._run_http_fallback_test(options, 400) 597 598 def test_version_check(self): 599 """Tests http fallback on version check fail.""" 600 601 options = self._options 602 options.version = 99 603 self._run_http_fallback_test(options, 400) 604 605 def test_basic_auth_connection(self): 606 """Test successful basic auth""" 607 608 options = self._options 609 options.use_basic_auth = True 610 611 self.server_stderr = subprocess.PIPE 612 self._run_test_with_options(_check_handshake_with_basic_auth, 613 options, 614 server_options=['--basic-auth']) 615 616 def test_invalid_basic_auth_connection(self): 617 """Tests basic auth with invalid credentials""" 618 619 options = self._options 620 options.use_basic_auth = True 621 options.basic_auth_credential = 'invalid:test' 622 623 self.server_stderr = subprocess.PIPE 624 625 with self.assertRaises(client_for_testing.HttpStatusException) as e: 626 self._run_test_with_options(_check_handshake_with_basic_auth, 627 options, 628 server_options=['--basic-auth']) 629 self.assertEqual(101, e.exception.status) 630 631 632 class EndToEndTestWithEchoClient(EndToEndTestBase): 633 def setUp(self): 634 EndToEndTestBase.setUp(self) 635 636 def _check_example_echo_client_result(self, expected, stdoutdata, 637 stderrdata): 638 actual = stdoutdata.decode(locale.getpreferredencoding()) 639 640 # In Python 3 on Windows we get "\r\n" terminators back from 641 # the subprocess and we need to replace them with "\n" to get 642 # a match. This is a bit of a hack, but avoids platform- and 643 # version- specific code. 644 actual = actual.replace('\r\n', '\n') 645 646 if actual != expected: 647 raise Exception('Unexpected result on example echo client: ' 648 '%r (expected) vs %r (actual)' % 649 (expected, actual)) 650 if stderrdata is not None: 651 raise Exception('Unexpected error message on example echo ' 652 'client: %r' % stderrdata) 653 654 def test_example_echo_client(self): 655 """Tests that the echo_client.py example can talk with the server.""" 656 657 server = self._run_server() 658 try: 659 time.sleep(_SERVER_WARMUP_IN_SEC) 660 661 client_command = os.path.join(self.top_dir, 'example', 662 'echo_client.py') 663 664 # Expected output for the default messages. 665 default_expectation = (u'Send: Hello\n' 666 u'Recv: Hello\n' 667 u'Send: <>\n' 668 u'Recv: <>\n' 669 u'Send close\n' 670 u'Recv ack\n') 671 672 args = [client_command, '-p', str(self._options.server_port)] 673 client = self._run_python_command(args, stdout=subprocess.PIPE) 674 stdoutdata, stderrdata = client.communicate() 675 self._check_example_echo_client_result(default_expectation, 676 stdoutdata, stderrdata) 677 678 # Process a big message for which extended payload length is used. 679 # To handle extended payload length, ws_version attribute will be 680 # accessed. This test checks that ws_version is correctly set. 681 big_message = 'a' * 1024 682 args = [ 683 client_command, '-p', 684 str(self._options.server_port), '-m', big_message 685 ] 686 client = self._run_python_command(args, stdout=subprocess.PIPE) 687 stdoutdata, stderrdata = client.communicate() 688 expected = ('Send: %s\nRecv: %s\nSend close\nRecv ack\n' % 689 (big_message, big_message)) 690 self._check_example_echo_client_result(expected, stdoutdata, 691 stderrdata) 692 693 # Test the permessage-deflate extension. 694 args = [ 695 client_command, '-p', 696 str(self._options.server_port), '--use_permessage_deflate' 697 ] 698 client = self._run_python_command(args, stdout=subprocess.PIPE) 699 stdoutdata, stderrdata = client.communicate() 700 self._check_example_echo_client_result(default_expectation, 701 stdoutdata, stderrdata) 702 finally: 703 self._close_server(server) 704 705 706 class EndToEndTestWithCgi(EndToEndTestBase): 707 def setUp(self): 708 EndToEndTestBase.setUp(self) 709 710 def test_cgi(self): 711 """Verifies that CGI scripts work.""" 712 713 server = self._run_server(extra_args=['--cgi-paths', '/cgi-bin']) 714 time.sleep(_SERVER_WARMUP_IN_SEC) 715 716 url = 'http://localhost:%d/cgi-bin/hi.py' % self._options.server_port 717 718 # urlopen() in Python 2.7 doesn't support "with". 719 try: 720 f = urllib.request.urlopen(url) 721 except: 722 self._close_server(server) 723 raise 724 725 try: 726 self.assertEqual(f.getcode(), 200) 727 self.assertEqual(f.info().get('Content-Type'), 'text/plain') 728 body = f.read() 729 self.assertEqual(body.rstrip(b'\r\n'), b'Hi from hi.py') 730 finally: 731 f.close() 732 self._close_server(server) 733 734 735 if __name__ == '__main__': 736 unittest.main() 737 738 # vi:sts=4 sw=4 et