cqrenet/relay
3
0
Fork 0
mirror of https://github.com/chatmail/relay.git synced 2026-05-10 16:04:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: use absolute path instead of relative path, and streamline some code parts according to comments at https://github.com/chatmail/relay/pull/785

Browse Source
This commit is contained in:
holger krekel
2025-12-18 22:51:38 +01:00
parent 24e3f33acd
commit f04a624e19
2 changed files with 22 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
chatmaild/src/chatmaild/expire.py
View File

@@ -14,7 +14,7 @@ from stat import S_ISREG
from chatmaild.config import read_config from chatmaild.config import read_config
FileEntry = namedtuple("FileEntry", ("relpath", "mtime", "size")) FileEntry = namedtuple("FileEntry", ("path", "mtime", "size"))
def iter_mailboxes(basedir, maxnum): def iter_mailboxes(basedir, maxnum):
@@ -51,35 +51,22 @@ class MailboxStat:
def __init__(self, basedir): def __init__(self, basedir):
self.basedir = str(basedir) self.basedir = str(basedir)
# all detected messages in cur/new/tmp folders
self.messages = [] self.messages = []
# all detected files in mailbox top dir
self.extrafiles = [] self.extrafiles = []
self.scandir(self.basedir)
# scan all relevant files (without recursion) def scandir(self, folderdir):
old_cwd = os.getcwd() for name in os_listdir_if_exists(folderdir):
try: path = f"{folderdir}/{name}"
os.chdir(self.basedir)
except FileNotFoundError:
return
try:
self.scandir(".")
finally:
os.chdir(old_cwd)
def scandir(self, dirname):
for name in os_listdir_if_exists(dirname):
relpath = f"{dirname}/{name}"
if name in ("cur", "new", "tmp"): if name in ("cur", "new", "tmp"):
for msg_name in os_listdir_if_exists(relpath): for msg_name in os_listdir_if_exists(path):
entry = get_file_entry(f"{relpath}/{msg_name}") entry = get_file_entry(f"{path}/{msg_name}")
if entry is not None: if entry is not None:
self.messages.append(entry) self.messages.append(entry)
elif relpath == "./.DeltaChat": elif os.path.isdir(path):
self.scandir(name) self.scandir(path)
else: else:
entry = get_file_entry(relpath) entry = get_file_entry(path)
if entry is not None: if entry is not None:
self.extrafiles.append(entry) self.extrafiles.append(entry)
if name == "password": if name == "password":
@@ -137,13 +124,6 @@ class Expiry:
self.remove_mailbox(mbox.basedir) self.remove_mailbox(mbox.basedir)
return return
# all to-be-removed files are relative to the mailbox basedir
try:
os.chdir(mbox.basedir)
except FileNotFoundError:
print_info(f"mailbox not found/vanished {mbox.basedir}")
return
mboxname = os.path.basename(mbox.basedir) mboxname = os.path.basename(mbox.basedir)
if self.verbose: if self.verbose:
date = datetime.fromtimestamp(mbox.last_login) if mbox.last_login else None date = datetime.fromtimestamp(mbox.last_login) if mbox.last_login else None
@@ -154,11 +134,12 @@ class Expiry:
self.all_files += len(mbox.messages) self.all_files += len(mbox.messages)
for message in mbox.messages: for message in mbox.messages:
if message.mtime < cutoff_mails: if message.mtime < cutoff_mails:
self.remove_file(message.relpath, mtime=message.mtime) self.remove_file(message.path, mtime=message.mtime)
elif message.size > 200000 and message.mtime < cutoff_large_mails: elif message.size > 200000 and message.mtime < cutoff_large_mails:
# we only remove noticed large files (not unnoticed ones in new/) # we only remove noticed large files (not unnoticed ones in new/)
if "cur" in message.relpath.split("/"): parts = message.path.split("/")
self.remove_file(message.relpath, mtime=message.mtime) if len(parts) >= 2 and parts[-2] == "cur":
self.remove_file(message.path, mtime=message.mtime)
else: else:
continue continue
changed = True changed = True

15
chatmaild/src/chatmaild/tests/test_expire.py
View File

@@ -17,16 +17,17 @@ from chatmaild.expire import main as expiry_main
from chatmaild.fsreport import main as report_main from chatmaild.fsreport import main as report_main
def fill_mbox(basedir1): def fill_mbox(folderdir):
password = basedir1.joinpath("password") password = folderdir.joinpath("password")
password.write_text("xxx") password.write_text("xxx")
basedir1.joinpath("maildirsize").write_text("xxx") folderdir.joinpath("maildirsize").write_text("xxx")
garbagedir = basedir1.joinpath("garbagedir") garbagedir = folderdir.joinpath("garbagedir")
garbagedir.mkdir() garbagedir.mkdir()
garbagedir.joinpath("bimbum").write_text("hello")
create_new_messages(basedir1, ["cur/msg1"], size=500) create_new_messages(folderdir, ["cur/msg1"], size=500)
create_new_messages(basedir1, ["new/msg2"], size=600) create_new_messages(folderdir, ["new/msg2"], size=600)
def create_new_messages(basedir, relpaths, size=1000, days=0): def create_new_messages(basedir, relpaths, size=1000, days=0):
@@ -86,7 +87,7 @@ def test_stats_mailbox(mbox1):
create_new_messages(mbox1.basedir, ["large-extra"], size=1000) create_new_messages(mbox1.basedir, ["large-extra"], size=1000)
create_new_messages(mbox1.basedir, ["index-something"], size=3) create_new_messages(mbox1.basedir, ["index-something"], size=3)
mbox2 = MailboxStat(mbox1.basedir) mbox2 = MailboxStat(mbox1.basedir)
assert len(mbox2.extrafiles) == 4 assert len(mbox2.extrafiles) == 5
assert mbox2.extrafiles[0].size == 1000 assert mbox2.extrafiles[0].size == 1000
# cope well with mailbox dirs that have no password (for whatever reason) # cope well with mailbox dirs that have no password (for whatever reason)