From a0795a6d17cb5192dd913aef0717edf4204e15d7 Mon Sep 17 00:00:00 2001 From: Balakrishnan Balasubramanian Date: Tue, 6 Jun 2023 23:20:27 -0400 Subject: [PATCH] Add pop3 test suite --- mail4one/pop3.py | 2 +- mail4one/pop_test.py | 76 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) create mode 100644 mail4one/pop_test.py diff --git a/mail4one/pop3.py b/mail4one/pop3.py index d391a29..543f52f 100644 --- a/mail4one/pop3.py +++ b/mail4one/pop3.py @@ -246,7 +246,7 @@ async def start_session(): def parse_users(users: list[User]): def inner(): - for user in Users: + for user in users: user = User(user) pwinfo = parse_hash(user.password_hash) yield user.username, (pwinfo, user.mbox) diff --git a/mail4one/pop_test.py b/mail4one/pop_test.py new file mode 100644 index 0000000..661f794 --- /dev/null +++ b/mail4one/pop_test.py @@ -0,0 +1,76 @@ +import unittest +import asyncio +import logging +from .pop3 import create_pop_server + + +class TestPop3(unittest.IsolatedAsyncioTestCase): + + def setUp(self): + logging.basicConfig(level=logging.CRITICAL) + + async def asyncSetUp(self): + pop_server = await create_pop_server(host='127.0.0.1', + port=7995, + mails_path='w.tmp', + users=[]) + self.task = asyncio.create_task(pop_server.serve_forever()) + + async def test_QUIT(self): + reader, writer = await asyncio.open_connection('127.0.0.1', 7995) + dialog = """ + S: +OK Server Ready + C: QUIT + S: +OK Bye + """ + await self.dialog_checker(reader, writer, dialog) + + async def test_BAD(self): + reader, writer = await asyncio.open_connection('127.0.0.1', 7995) + dialog = """ + S: +OK Server Ready + C: HELO + S: -ERR Bad command + C: HEYA + S: -ERR Bad command + C: LIST + S: -ERR Something went wrong + C: HELO + """ + await self.dialog_checker(reader, writer, dialog) + # TODO fix + # self.assertTrue(reader.at_eof(), "server should close the connection") + + async def test_CAPA(self): + reader, writer = await asyncio.open_connection('127.0.0.1', 7995) + dialog = """ + S: +OK Server Ready + C: CAPA + S: +OK Following are supported + S: USER + S: . + C: QUIT + S: +OK Bye + """ + await self.dialog_checker(reader, writer, dialog) + + async def asyncTearDown(self): + self.task.cancel("test done") + + async def dialog_checker(self, reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, dialog: str): + for line in dialog.splitlines(): + line = line.strip() + if not line: + continue + side, data_str = line.split(maxsplit=1) + data = f"{data_str}\r\n".encode() + if side == "C:": + writer.write(data) + else: + resp = await reader.readline() + self.assertEqual(data, resp) + + +if __name__ == '__main__': + unittest.main()