Сценарии тестирования нового метода

[arhadthedev]

Сценарии:

• клиент включает, клиент выключает
• сервер включает, сервер выключает
• клиент включает, сервер выключает
• сервер включает, клиент выключает

Тест - это все комбинации по два с повторами

проверка сброса кэша при переключении:
клиент отправляет сырой, ждёт сырой
сервер принимает сырой, отправляет сырой
подъём защиты
проверка сброса кэша при переключении:
включившая сторона отправляет
другая сторона отправляет
отключение защиты
проверка сброса кэша при переключении:
включившая сторона отправляет
другая сторона отправляет

    @unittest.skipIf(ssl is None, 'No ssl module')
    def test_tls(self):

    @unittest.skipIf(ssl is None, 'No ssl module')
    def test_start_tls(self):

        class MyServer:

            def __init__(self, loop):
                self.server = None
                self.loop = loop

            async def handle_client(self, client_reader, client_writer):
                data1 = await client_reader.readline()
                client_writer.write(data1)
                await client_writer.drain()
                assert client_writer.get_extra_info('sslcontext') is None
                await client_writer.start_tls(
                    test_utils.simple_server_sslcontext())
                assert client_writer.get_extra_info('sslcontext') is not None
                data2 = await client_reader.readline()
                client_writer.write(data2)
                await client_writer.drain()
                client_writer.close()
                await client_writer.wait_closed()

            def start(self):
                sock = socket.create_server(('127.0.0.1', 0))
                self.server = self.loop.run_until_complete(
                    asyncio.start_server(self.handle_client,
                                         sock=sock))
                return sock.getsockname()

            def stop(self):
                if self.server is not None:
                    self.server.close()
                    self.loop.run_until_complete(self.server.wait_closed())
                    self.server = None

        async def client(addr):
            reader, writer = await asyncio.open_connection(*addr)
            writer.write(b"hello world 1!\n")
            await writer.drain()
            msgback1 = await reader.readline()
            assert writer.get_extra_info('sslcontext') is None
            await writer.start_tls(test_utils.simple_client_sslcontext())
            assert writer.get_extra_info('sslcontext') is not None
            writer.write(b"hello world 2!\n")
            await writer.drain()
            msgback2 = await reader.readline()
            writer.close()
            await writer.wait_closed()
            return msgback1, msgback2

        messages = []
        self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx))

        server = MyServer(self.loop)
        addr = server.start()
        msg1, msg2 = self.loop.run_until_complete(client(addr))
        server.stop()

        self.assertEqual(messages, [])
        self.assertEqual(msg1, b"hello world 1!\n")
        self.assertEqual(msg2, b"hello world 2!\n")