pylint (#8)
Current score: Your code has been rated at 8.08/10 (previous run: 7.98/10, +0.09) Reviewed-on: #8
This commit is contained in:
parent
64dcc74b8d
commit
57d0eeaf0f
38
DEVNOTES.md
Normal file
38
DEVNOTES.md
Normal file
@ -0,0 +1,38 @@
|
||||
Notes for developers
|
||||
|
||||
## Running just one test
|
||||
|
||||
```
|
||||
python -m unittest tests.test_pop.TestPop3.test_CAPA
|
||||
```
|
||||
|
||||
## Patch for enable logging in test
|
||||
|
||||
Patch generated using below
|
||||
|
||||
```
|
||||
git diff --patch -U1 tests >> ./DEVNOTES.md
|
||||
```
|
||||
|
||||
Apply with below
|
||||
|
||||
```bash
|
||||
git apply - <<PATCH
|
||||
diff --git a/tests/test_pop.py b/tests/test_pop.py
|
||||
index 55c1a91..a825665 100644
|
||||
--- a/tests/test_pop.py
|
||||
+++ b/tests/test_pop.py
|
||||
@@ -55,3 +55,3 @@ def setUpModule() -> None:
|
||||
global MAILS_PATH
|
||||
- logging.basicConfig(level=logging.CRITICAL)
|
||||
+ logging.basicConfig(level=logging.DEBUG)
|
||||
td = tempfile.TemporaryDirectory(prefix="m41.pop.")
|
||||
PATCH
|
||||
```
|
||||
|
||||
## pylint
|
||||
|
||||
```
|
||||
pylint mail4one/*py > /tmp/errs
|
||||
vim +"cfile /tmp/errs"
|
||||
```
|
@ -96,11 +96,10 @@ def parse_checkers(cfg: Config) -> list[Checker]:
|
||||
raise Exception("Both addrs and addr_rexs is set")
|
||||
if m.addrs:
|
||||
return lambda malias: malias in m.addrs
|
||||
elif m.addr_rexs:
|
||||
if m.addr_rexs:
|
||||
compiled_res = [re.compile(reg) for reg in m.addr_rexs]
|
||||
return lambda malias: any(reg.match(malias) for reg in compiled_res)
|
||||
else:
|
||||
raise Exception("Neither addrs nor addr_rexs is set")
|
||||
raise Exception("Neither addrs nor addr_rexs is set")
|
||||
|
||||
matches = {m.name: make_match_fn(Match(m)) for m in cfg.matches or []}
|
||||
matches[DEFAULT_MATCH_ALL] = lambda _: True
|
||||
|
@ -72,14 +72,14 @@ class PopLogger(logging.LoggerAdapter):
|
||||
def __init__(self):
|
||||
super().__init__(logging.getLogger("pop3"), None)
|
||||
|
||||
def process(self, msg, kwargs):
|
||||
state: State = c_state.get(None)
|
||||
if not state:
|
||||
return super().process(msg, kwargs)
|
||||
def process(self, log_msg, kwargs):
|
||||
st: State = c_state.get(None)
|
||||
if not st:
|
||||
return super().process(log_msg, kwargs)
|
||||
user = "NA"
|
||||
if state.username:
|
||||
user = state.username
|
||||
return super().process(f"{state.ip} {state.req_id} {user} {msg}", kwargs)
|
||||
if st.username:
|
||||
user = st.username
|
||||
return super().process(f"{st.ip} {st.req_id} {user} {log_msg}", kwargs)
|
||||
|
||||
|
||||
logger = PopLogger()
|
||||
@ -101,8 +101,7 @@ async def next_req() -> Request:
|
||||
if request.cmd == Command.QUIT:
|
||||
raise ClientQuit
|
||||
return request
|
||||
else:
|
||||
raise ClientError(f"Bad command {InvalidCommand.RETRIES} times")
|
||||
raise ClientError(f"Bad command {InvalidCommand.RETRIES} times")
|
||||
|
||||
|
||||
async def expect_cmd(*commands: Command) -> Request:
|
||||
@ -150,25 +149,23 @@ async def auth_stage() -> None:
|
||||
write(ok("Following are supported"))
|
||||
write(msg("USER"))
|
||||
write(end())
|
||||
else:
|
||||
await handle_user_pass_auth(req)
|
||||
if state().username in scfg().loggedin_users:
|
||||
logger.warning(
|
||||
f"User: {state().username} already has an active session"
|
||||
)
|
||||
raise AuthError("Already logged in")
|
||||
else:
|
||||
scfg().loggedin_users.add(state().username)
|
||||
write(ok("Login successful"))
|
||||
return
|
||||
continue
|
||||
await handle_user_pass_auth(req)
|
||||
if state().username in scfg().loggedin_users:
|
||||
logger.warning(
|
||||
f"User: {state().username} already has an active session"
|
||||
)
|
||||
raise AuthError("Already logged in")
|
||||
scfg().loggedin_users.add(state().username)
|
||||
write(ok("Login successful"))
|
||||
return
|
||||
except AuthError as ae:
|
||||
write(err(f"Auth Failed: {ae}"))
|
||||
except ClientQuit as c:
|
||||
except ClientQuit:
|
||||
write(ok("Bye"))
|
||||
logger.warning("Client has QUIT before auth succeeded")
|
||||
raise
|
||||
else:
|
||||
raise ClientError("Failed to authenticate")
|
||||
raise ClientError("Failed to authenticate")
|
||||
|
||||
|
||||
def trans_command_capa(_, __) -> None:
|
||||
@ -269,9 +266,8 @@ async def process_transactions(mails_list: list[MailEntry]) -> set[str]:
|
||||
except KeyError:
|
||||
write(err("Not implemented"))
|
||||
raise ClientError("We shouldn't reach here")
|
||||
else:
|
||||
func(mails, req)
|
||||
await state().writer.drain()
|
||||
func(mails, req)
|
||||
await state().writer.drain()
|
||||
|
||||
|
||||
def get_deleted_items(deleted_items_path: Path) -> set[str]:
|
||||
@ -302,7 +298,7 @@ async def transaction_stage() -> None:
|
||||
deleted_items_path, existing_deleted_items.union(new_deleted_items)
|
||||
)
|
||||
|
||||
logger.info(f"Saved deleted items")
|
||||
logger.info("Saved deleted items")
|
||||
|
||||
|
||||
async def start_session() -> None:
|
||||
@ -339,13 +335,13 @@ def parse_users(users: list[User]) -> dict[str, tuple[PWInfo, str]]:
|
||||
|
||||
|
||||
def make_pop_server_callback(mails_path: Path, users: list[User], timeout_seconds: int):
|
||||
scfg = SharedState(mails_path=mails_path, users=parse_users(users))
|
||||
s_state = SharedState(mails_path=mails_path, users=parse_users(users))
|
||||
|
||||
async def session_cb(reader: StreamReader, writer: StreamWriter):
|
||||
c_shared_state.set(scfg)
|
||||
c_shared_state.set(s_state)
|
||||
ip, _ = writer.get_extra_info("peername")
|
||||
c_state.set(State(reader=reader, writer=writer, ip=ip, req_id=scfg.next_id()))
|
||||
logger.info(f"Got pop server callback")
|
||||
c_state.set(State(reader=reader, writer=writer, ip=ip, req_id=s_state.next_id()))
|
||||
logger.info("Got pop server callback")
|
||||
try:
|
||||
try:
|
||||
return await asyncio.wait_for(start_session(), timeout_seconds)
|
||||
@ -367,7 +363,7 @@ async def create_pop_server(
|
||||
timeout_seconds: int = 60,
|
||||
) -> asyncio.Server:
|
||||
logging.info(
|
||||
f"Starting POP3 server {host=}, {port=}, {mails_path=!s}, {len(users)=}, {ssl_context != None=}, {timeout_seconds=}"
|
||||
f"Starting POP3 server {host=}, {port=}, {mails_path=!s}, {len(users)=}, {bool(ssl_context)=}, {timeout_seconds=}"
|
||||
)
|
||||
return await asyncio.start_server(
|
||||
make_pop_server_callback(mails_path, users, timeout_seconds),
|
||||
|
@ -20,12 +20,10 @@ class ClientDisconnected(ClientError):
|
||||
class InvalidCommand(ClientError):
|
||||
RETRIES = 3
|
||||
"""WIll allow NUM_BAD_COMMANDS times"""
|
||||
pass
|
||||
|
||||
|
||||
class AuthError(ClientError):
|
||||
RETRIES = 3
|
||||
pass
|
||||
|
||||
|
||||
class Command(Enum):
|
||||
|
@ -42,17 +42,15 @@ async def a_main(cfg: config.Config) -> None:
|
||||
def get_tls_context(tls: Union[config.TLSCfg, str]):
|
||||
if tls == "default":
|
||||
return default_tls_context
|
||||
elif tls == "disable":
|
||||
if tls == "disable":
|
||||
return None
|
||||
else:
|
||||
tls_cfg = config.TLSCfg(tls)
|
||||
return create_tls_context(tls_cfg.certfile, tls_cfg.keyfile)
|
||||
tls_cfg = config.TLSCfg(tls)
|
||||
return create_tls_context(tls_cfg.certfile, tls_cfg.keyfile)
|
||||
|
||||
def get_host(host):
|
||||
if host == "default":
|
||||
return cfg.default_host
|
||||
else:
|
||||
return host
|
||||
return host
|
||||
|
||||
mbox_finder = config.gen_addr_to_mboxes(cfg)
|
||||
servers: list[asyncio.Server] = []
|
||||
|
@ -24,6 +24,8 @@ class MyHandler(AsyncMessage):
|
||||
super().__init__()
|
||||
self.mails_path = mails_path
|
||||
self.mbox_finder = mbox_finder
|
||||
self.rcpt_tos = []
|
||||
self.peer = None
|
||||
|
||||
async def handle_DATA(
|
||||
self, server: SMTP, session: SMTPSession, envelope: SMTPEnvelope
|
||||
@ -32,7 +34,7 @@ class MyHandler(AsyncMessage):
|
||||
self.peer = session.peer
|
||||
return await super().handle_DATA(server, session, envelope)
|
||||
|
||||
async def handle_message(self, m: Message): # type: ignore[override]
|
||||
async def handle_message(self, message: Message): # type: ignore[override]
|
||||
all_mboxes: set[str] = set()
|
||||
for addr in self.rcpt_tos:
|
||||
for mbox in self.mbox_finder(addr.lower()):
|
||||
@ -49,7 +51,7 @@ class MyHandler(AsyncMessage):
|
||||
temp_email_path = Path(tmpdir) / filename
|
||||
with open(temp_email_path, "wb") as fp:
|
||||
gen = BytesGenerator(fp, policy=email.policy.SMTP)
|
||||
gen.flatten(m)
|
||||
gen.flatten(message)
|
||||
for mbox in all_mboxes:
|
||||
shutil.copy(temp_email_path, self.mails_path / mbox / "new")
|
||||
logger.info(
|
||||
@ -102,7 +104,7 @@ async def create_smtp_server_starttls(
|
||||
smtputf8: bool,
|
||||
) -> asyncio.Server:
|
||||
logging.info(
|
||||
f"Starting SMTP STARTTLS server {host=}, {port=}, {mails_path=!s}, {ssl_context != None=}"
|
||||
f"Starting SMTP STARTTLS server {host=}, {port=}, {mails_path=!s}, {bool(ssl_context)=}"
|
||||
)
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.create_server(
|
||||
@ -129,7 +131,7 @@ async def create_smtp_server(
|
||||
smtputf8: bool,
|
||||
) -> asyncio.Server:
|
||||
logging.info(
|
||||
f"Starting SMTP server {host=}, {port=}, {mails_path=!s}, {ssl_context != None=}"
|
||||
f"Starting SMTP server {host=}, {port=}, {mails_path=!s}, {bool(ssl_context)=}"
|
||||
)
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.create_server(
|
||||
|
Loading…
Reference in New Issue
Block a user