cqrenet/relay
4
0
Fork 0
mirror of https://github.com/chatmail/relay.git synced 2026-05-20 04:48:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

test and fix for edge case

Browse Source
This commit is contained in:
holger krekel
2024-03-28 10:46:21 +01:00
parent 16f237dc60
commit 6ab3e9657d
2 changed files with 15 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
chatmaild/src/chatmaild/metadata.py
View File

@@ -35,24 +35,22 @@ class Notifier:
self.to_notify_queue = Queue() self.to_notify_queue = Queue()
def get_metadata_dict(self, addr): def get_metadata_dict(self, addr):
addr_path = self.vmail_dir.joinpath(addr) return FileDict(self.vmail_dir / addr / "metadata.marshalled")
return FileDict(addr_path / "metadata.marshalled")
def add_token(self, addr, token): def add_token(self, addr, token):
with self.get_metadata_dict(addr).modify() as data: with self.get_metadata_dict(addr).modify() as data:
tokens = data.get(METADATA_TOKEN_KEY) tokens = data.get(METADATA_TOKEN_KEY)
if tokens is None: if tokens is None:
data[METADATA_TOKEN_KEY] = tokens = [] data[METADATA_TOKEN_KEY] = [token]
if token not in tokens: elif token not in tokens:
tokens.append(token) tokens.append(token)
def remove_token(self, addr, token): def remove_token(self, addr, token):
with self.get_metadata_dict(addr).modify() as data: with self.get_metadata_dict(addr).modify() as data:
tokens = data.get(METADATA_TOKEN_KEY) tokens = data.get(METADATA_TOKEN_KEY, [])
if tokens:
try: try:
tokens.remove(token) tokens.remove(token)
except KeyError: except ValueError:
pass pass
def get_tokens(self, addr): def get_tokens(self, addr):

7
chatmaild/src/chatmaild/tests/test_metadata.py
View File

@@ -40,6 +40,13 @@ def test_notifier_persistence(tmp_path, testaddr, testaddr2):
assert notifier1.get_tokens(testaddr2) == ["456"] assert notifier1.get_tokens(testaddr2) == ["456"]
def test_remove_nonexisting(tmp_path, testaddr):
notifier1 = Notifier(tmp_path)
notifier1.add_token(testaddr, "123")
notifier1.remove_token(testaddr, "1l23k1l2k3")
assert notifier1.get_tokens(testaddr) == ["123"]
def test_notifier_delete_without_set(notifier, testaddr): def test_notifier_delete_without_set(notifier, testaddr):
notifier.remove_token(testaddr, "123") notifier.remove_token(testaddr, "123")
assert not notifier.get_tokens(testaddr) assert not notifier.get_tokens(testaddr)