(self)
| 950 | ['::1', '::2', '::1']) |
| 951 | |
| 952 | def test_create_server(self): |
| 953 | proto = MyProto(self.loop) |
| 954 | f = self.loop.create_server(lambda: proto, '0.0.0.0', 0) |
| 955 | server = self.loop.run_until_complete(f) |
| 956 | self.assertEqual(len(server.sockets), 1) |
| 957 | sock = server.sockets[0] |
| 958 | host, port = sock.getsockname() |
| 959 | self.assertEqual(host, '0.0.0.0') |
| 960 | client = socket.socket() |
| 961 | client.connect(('127.0.0.1', port)) |
| 962 | client.sendall(b'xxx') |
| 963 | |
| 964 | self.loop.run_until_complete(proto.connected) |
| 965 | self.assertEqual('CONNECTED', proto.state) |
| 966 | |
| 967 | test_utils.run_until(self.loop, lambda: proto.nbytes > 0) |
| 968 | self.assertEqual(3, proto.nbytes) |
| 969 | |
| 970 | # extra info is available |
| 971 | self.assertIsNotNone(proto.transport.get_extra_info('sockname')) |
| 972 | self.assertEqual('127.0.0.1', |
| 973 | proto.transport.get_extra_info('peername')[0]) |
| 974 | |
| 975 | # close connection |
| 976 | proto.transport.close() |
| 977 | self.loop.run_until_complete(proto.done) |
| 978 | |
| 979 | self.assertEqual('CLOSED', proto.state) |
| 980 | |
| 981 | # the client socket must be closed after to avoid ECONNRESET upon |
| 982 | # recv()/send() on the serving socket |
| 983 | client.close() |
| 984 | |
| 985 | # close server |
| 986 | server.close() |
| 987 | |
| 988 | def test_create_server_trsock(self): |
| 989 | proto = MyProto(self.loop) |
nothing calls this directly
no test coverage detected