balki/mail4one
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases 5 Wiki Activity

add test for AUTH stage

Browse Source
This commit is contained in:
Balakrishnan Balasubramanian 2023-06-07 22:47:56 -04:00
parent fc7ff2a2b6
commit 53bb412721
1 changed files with 54 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
mail4one/pop_test.py
View File

@ -2,6 +2,7 @@ import unittest
import asyncio import asyncio
import logging import logging
from .pop3 import create_pop_server from .pop3 import create_pop_server
from .config import User
class TestPop3(unittest.IsolatedAsyncioTestCase): class TestPop3(unittest.IsolatedAsyncioTestCase):
@ -10,23 +11,30 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
logging.basicConfig(level=logging.CRITICAL) logging.basicConfig(level=logging.CRITICAL)
async def asyncSetUp(self): async def asyncSetUp(self):
test_hash = "".join((l.strip() for l in """
AFTY5EVN7AX47ZL7UMH3BETYWFBTAV3XHR73CEFAJBPN2NIHPWD
ZHV2UQSMSPHSQQ2A2BFQBNC77VL7F2UKATQNJZGYLCSU6C43UQD
AQXWXSWNGAEPGIMG2F3QDKBXL3MRHY6K2BPID64ZR6LABLPVSF
""".splitlines()))
users = [
User(username="foobar", password_hash=test_hash, mbox="mails")
]
pop_server = await create_pop_server(host='127.0.0.1', pop_server = await create_pop_server(host='127.0.0.1',
port=7995, port=7995,
mails_path='w.tmp', mails_path='w.tmp',
users=[]) users=users)
self.task = asyncio.create_task(pop_server.serve_forever()) self.task = asyncio.create_task(pop_server.serve_forever())
self.reader, self.writer = await asyncio.open_connection('127.0.0.1', 7995)
async def test_QUIT(self): async def test_QUIT(self):
reader, writer = await asyncio.open_connection('127.0.0.1', 7995)
dialog = """ dialog = """
S: +OK Server Ready S: +OK Server Ready
C: QUIT C: QUIT
S: +OK Bye S: +OK Bye
""" """
await self.dialog_checker(reader, writer, dialog) await self.dialog_checker(dialog)
async def test_BAD(self): async def test_BAD(self):
reader, writer = await asyncio.open_connection('127.0.0.1', 7995)
dialog = """ dialog = """
S: +OK Server Ready S: +OK Server Ready
C: HELO C: HELO
@ -37,12 +45,45 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
S: -ERR Something went wrong S: -ERR Something went wrong
C: HELO C: HELO
""" """
await self.dialog_checker(reader, writer, dialog) await self.dialog_checker(dialog)
# TODO fix # TODO fix
# self.assertTrue(reader.at_eof(), "server should close the connection") # self.assertTrue(reader.at_eof(), "server should close the connection")
async def test_AUTH(self):
dialog = """
S: +OK Server Ready
C: USER foobar
S: +OK Welcome
C: PASS helloworld
S: +OK Login successful
C: QUIT
S: +OK Bye
"""
await self.dialog_checker(dialog)
async def test_dupe_AUTH(self):
r1, w1 = await asyncio.open_connection('127.0.0.1', 7995)
r2, w2 = await asyncio.open_connection('127.0.0.1', 7995)
dialog = """
S: +OK Server Ready
C: USER foobar
S: +OK Welcome
C: PASS helloworld
"""
await self.dialog_checker_impl(r1, w1, dialog)
await self.dialog_checker_impl(r2, w2, dialog)
d1 = """S: +OK Login successful"""
d2 = """S: -ERR Auth Failed: Already logged in"""
await self.dialog_checker_impl(r1, w1, d1)
await self.dialog_checker_impl(r2, w2, d2)
end_dialog = """
C: QUIT
S: +OK Bye
"""
await self.dialog_checker_impl(r1, w1, end_dialog)
await self.dialog_checker_impl(r2, w2, end_dialog)
async def test_CAPA(self): async def test_CAPA(self):
reader, writer = await asyncio.open_connection('127.0.0.1', 7995)
dialog = """ dialog = """
S: +OK Server Ready S: +OK Server Ready
C: CAPA C: CAPA
@ -52,12 +93,17 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
C: QUIT C: QUIT
S: +OK Bye S: +OK Bye
""" """
await self.dialog_checker(reader, writer, dialog) await self.dialog_checker(dialog)
async def asyncTearDown(self): async def asyncTearDown(self):
self.writer.close()
await self.writer.wait_closed()
self.task.cancel("test done") self.task.cancel("test done")
async def dialog_checker(self, reader: asyncio.StreamReader, async def dialog_checker(self, dialog: str):
await self.dialog_checker_impl(self.reader, self.writer, dialog)
async def dialog_checker_impl(self, reader: asyncio.StreamReader,
writer: asyncio.StreamWriter, dialog: str): writer: asyncio.StreamWriter, dialog: str):
for line in dialog.splitlines(): for line in dialog.splitlines():
line = line.strip() line = line.strip()