s3lph/multischleuder
1
0
Fork 0
Code Issues Pull requests Projects Releases 8 Packages 2 Wiki Activity Actions

Fix test code so that it emulates open(a+)

Browse source
This commit is contained in:
s3lph 2022-04-18 17:14:12 +02:00
parent ade119ca69
commit 6dba5cc37f
2 changed files with 18 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
multischleuder/conflict.py
View file

@ -86,12 +86,11 @@ class KeyConflictResolution:
def _should_send(self, digest: str) -> bool:
now = int(datetime.utcnow().timestamp())
exists = os.path.exists(self._state_file)
with open(self._state_file, 'a+') as f:
f.seek(0)
if not exists:
state: Dict[str, int] = {}
else:
state: Dict[str, int] = {}
if f.tell() > 0:
# Only load the state if the file is not empty
f.seek(0)
try:
state = json.load(f)
except BaseException:

22
multischleuder/test/test_conflict.py
View file

@ -93,6 +93,7 @@ class TestKeyConflictResolution(unittest.TestCase):
def test_empty(self):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_NONE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_NONE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve('', '', [])
@ -124,6 +125,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_NONE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_NONE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(
@ -150,6 +152,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_NONE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_NONE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(
@ -186,6 +189,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_NONE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_NONE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(
@ -205,6 +209,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_NONE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_NONE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(
@ -220,31 +225,28 @@ class TestKeyConflictResolution(unittest.TestCase):
date1 = datetime(2022, 4, 15, 5, 23, 42, 0, tzinfo=tzutc())
date2 = datetime(2022, 4, 13, 5, 23, 42, 0, tzinfo=tzutc())
sub1 = SchleuderSubscriber(3, 'foo@example.org', key1, sch1.id, date1)
sub2 = SchleuderSubscriber(4, 'bar@example.org', key1, sch1.id, date2)
# This subscription is older, so its key will be preferred
sch2 = SchleuderList(23, 'test-south@schleuder.example.org', 'AF586C0625CF77BBB659747515D41C5D84BF99D3')
key2 = SchleuderKey(_PRIVKEY_2.fingerprint.replace(' ', ''), 'foo@example.org', str(_PRIVKEY_2.pubkey), sch2.id)
sub3 = SchleuderSubscriber(7, 'foo@example.org', key2, sch2.id, date2)
sub4 = SchleuderSubscriber(8, 'bar@example.org', key2, sch2.id, date1)
sub2 = SchleuderSubscriber(7, 'foo@example.org', key2, sch2.id, date2)
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_NONE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_NONE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, msgs = kcr.resolve(
target='test@schleuder.example.org',
mail_from='test-owner@schleuder.example.org',
subscriptions=[sub1, sub2, sub3, sub4])
self.assertEqual(2, len(msgs))
subscriptions=[sub1, sub2])
self.assertEqual(1, len(msgs))
now = datetime.utcnow().timestamp()
mock_statefile.assert_called_with('/tmp/state.json', 'a+')
contents.seek(0)
state = json.loads(contents.read())
self.assertEqual(2, len(state))
self.assertEqual(1, len(state))
self.assertIn(msgs[0].mime['X-MultiSchleuder-Digest'], state)
self.assertIn(msgs[1].mime['X-MultiSchleuder-Digest'], state)
self.assertLess(now - state[msgs[0].mime['X-MultiSchleuder-Digest']], 60)
self.assertLess(now - state[msgs[0].mime['X-MultiSchleuder-Digest']], 60)
# Todo: move this over to test_multilist
@ -266,6 +268,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_STALE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_STALE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(
@ -296,6 +299,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(86400, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_RECENT)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_RECENT)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(
@ -328,6 +332,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
kcr.dry_run()
contents = io.StringIO(_CONFLICT_STATE_STALE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_STALE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(
@ -359,6 +364,7 @@ class TestKeyConflictResolution(unittest.TestCase):
kcr = KeyConflictResolution(3600, '/tmp/state.json', _TEMPLATE)
contents = io.StringIO(_CONFLICT_STATE_NONE)
contents.seek(io.SEEK_END) # Opened with 'a+'
with patch('builtins.open', mock_open(read_data=_CONFLICT_STATE_NONE)) as mock_statefile:
mock_statefile().__enter__.return_value = contents
resolved, messages = kcr.resolve(