cqrenet/relay
4
0
Fork 0
mirror of https://github.com/chatmail/relay.git synced 2026-05-20 12:58:04 +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 917da899c6
commit d74b25adea
2 changed files with 15 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,25 +35,23 @@ 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 ValueError:
except KeyError: pass
pass
def get_tokens(self, addr): def get_tokens(self, addr):
return self.get_metadata_dict(addr).read().get(METADATA_TOKEN_KEY, []) return self.get_metadata_dict(addr).read().get(METADATA_TOKEN_KEY, [])

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)