Launch a server, connect a client to it and try various reads and writes.
(client_context, server_context, indata=b"FOO\n",
chatty=True, connectionchatty=False, sni_name=None,
session=None)
| 2683 | self.server.close() |
| 2684 | |
| 2685 | def server_params_test(client_context, server_context, indata=b"FOO\n", |
| 2686 | chatty=True, connectionchatty=False, sni_name=None, |
| 2687 | session=None): |
| 2688 | """ |
| 2689 | Launch a server, connect a client to it and try various reads |
| 2690 | and writes. |
| 2691 | """ |
| 2692 | stats = {} |
| 2693 | server = ThreadedEchoServer(context=server_context, |
| 2694 | chatty=chatty, |
| 2695 | connectionchatty=False) |
| 2696 | with server: |
| 2697 | with client_context.wrap_socket(socket.socket(), |
| 2698 | server_hostname=sni_name, session=session) as s: |
| 2699 | s.connect((HOST, server.port)) |
| 2700 | for arg in [indata, bytearray(indata), memoryview(indata)]: |
| 2701 | if connectionchatty: |
| 2702 | if support.verbose: |
| 2703 | sys.stdout.write( |
| 2704 | " client: sending %r...\n" % indata) |
| 2705 | s.write(arg) |
| 2706 | outdata = s.read() |
| 2707 | if connectionchatty: |
| 2708 | if support.verbose: |
| 2709 | sys.stdout.write(" client: read %r\n" % outdata) |
| 2710 | if outdata != indata.lower(): |
| 2711 | raise AssertionError( |
| 2712 | "bad data <<%r>> (%d) received; expected <<%r>> (%d)\n" |
| 2713 | % (outdata[:20], len(outdata), |
| 2714 | indata[:20].lower(), len(indata))) |
| 2715 | s.write(b"over\n") |
| 2716 | if connectionchatty: |
| 2717 | if support.verbose: |
| 2718 | sys.stdout.write(" client: closing connection.\n") |
| 2719 | stats.update({ |
| 2720 | 'compression': s.compression(), |
| 2721 | 'cipher': s.cipher(), |
| 2722 | 'peercert': s.getpeercert(), |
| 2723 | 'client_alpn_protocol': s.selected_alpn_protocol(), |
| 2724 | 'version': s.version(), |
| 2725 | 'session_reused': s.session_reused, |
| 2726 | 'session': s.session, |
| 2727 | }) |
| 2728 | s.close() |
| 2729 | stats['server_alpn_protocols'] = server.selected_alpn_protocols |
| 2730 | stats['server_shared_ciphers'] = server.shared_ciphers |
| 2731 | return stats |
| 2732 | |
| 2733 | def try_protocol_combo(server_protocol, client_protocol, expect_success, |
| 2734 | certsreqs=None, server_options=0, client_options=0): |
no test coverage detected