diff --git a/src/olympia/blocklist/tests/test_cron.py b/src/olympia/blocklist/tests/test_cron.py index 8b751d0ce785..0466e0135ec2 100644 --- a/src/olympia/blocklist/tests/test_cron.py +++ b/src/olympia/blocklist/tests/test_cron.py @@ -542,44 +542,196 @@ def test_upload_stash_unless_enough_changes(self): assert len(data['softblocked']) == 1 @mock.patch('olympia.blocklist.mlbf.BASE_REPLACE_THRESHOLD', 1) - def test_upload_stash_even_if_filter_is_updated(self): - """ - If enough changes of one type are made, update the filter, but still upload - a stash if there are changes of other types. - """ - self._block_version(is_signed=True, block_type=BlockType.BLOCKED) - self._block_version(is_signed=True, block_type=BlockType.BLOCKED) - self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) - upload_mlbf_to_remote_settings() - assert self.mocks[ - 'olympia.blocklist.cron.upload_filter.delay' - ].call_args_list == [ - mock.call( - self.current_time, - filter_list=[BlockType.BLOCKED.name], - create_stash=False, - ) - ] + def _test_upload_stash_and_filter( + self, + enable_soft_blocking: bool, + expected_stash: dict | None, + expected_filters: List[BlockType], + ): + with override_switch('enable-soft-blocking', active=enable_soft_blocking): + upload_mlbf_to_remote_settings() + + # generation time is set to current time so we can load the mlbf mlbf = MLBF.load_from_storage(self.current_time) - assert mlbf.storage.exists(mlbf.filter_path(BlockType.BLOCKED)) - assert not mlbf.storage.exists(mlbf.stash_path) - with override_switch('enable-soft-blocking', active=True): - self._block_version(is_signed=True, block_type=BlockType.BLOCKED) - self._block_version(is_signed=True, block_type=BlockType.BLOCKED) - upload_mlbf_to_remote_settings() - self.mocks['olympia.blocklist.cron.upload_filter.delay'].assert_called_with( - self.current_time, - filter_list=[BlockType.BLOCKED.name], - create_stash=True, + if expected_stash is None: + assert not mlbf.storage.exists(mlbf.stash_path), ( + 'Expected no stash but one exists' + ) + else: + assert mlbf.storage.exists(mlbf.stash_path), ( + f'Expected stash {expected_stash} but none exists' ) - mlbf = MLBF.load_from_storage(self.current_time) - assert mlbf.storage.exists(mlbf.filter_path(BlockType.BLOCKED)) - assert mlbf.storage.exists(mlbf.stash_path) with mlbf.storage.open(mlbf.stash_path, 'r') as f: data = json.load(f) - assert len(data['blocked']) == 0 - assert len(data['softblocked']) == 1 + for key, expected_count in expected_stash.items(): + assert len(data[key]) == expected_count, ( + f'Expected {expected_count} {key} but got {len(data[key])}' + ) + + for expected_filter in expected_filters: + assert mlbf.storage.exists(mlbf.filter_path(expected_filter)), ( + f'Expected filter {expected_filter} but none exists' + ) + + def test_upload_blocked_stash_and_softblock_filter(self): + # Enough blocks for a stash + self._block_version(is_signed=True, block_type=BlockType.BLOCKED) + # Enough soft blocks for a filter + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + + # Expected stash does not change + expected_stash = { + 'blocked': 1, + # Expect no soft blocks becuase there are enough for a filter + 'softblocked': 0, + # There are no unblocked versions + 'unblocked': 0, + } + + self._test_upload_stash_and_filter( + # Even though there are enough soft blocks, soft blocking is disabled + enable_soft_blocking=False, + expected_stash=expected_stash, + # Expect no filter as soft blocking is disabled + expected_filters=[], + ) + + # Now try again with soft blocking enabled + self._test_upload_stash_and_filter( + # Soft blocking is enabled, so we expect the same stash and a new filter + enable_soft_blocking=True, + expected_stash=expected_stash, + # Expect a soft blocked filter + expected_filters=[BlockType.SOFT_BLOCKED], + ) + + def test_upload_blocked_stash_and_softblock_filter_with_switch_enabled(self): + # Enough blocks for a stash + self._block_version(is_signed=True, block_type=BlockType.BLOCKED) + # Enough soft blocks for a filter + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + + self._test_upload_stash_and_filter( + # Soft blocking is enabled, so we expect a stash and a filter + enable_soft_blocking=True, + expected_stash={ + 'blocked': 1, + # Expect no soft blocks becuase there are enough for a filter + 'softblocked': 0, + 'unblocked': 0, + }, + # Expect a soft blocked filter + expected_filters=[BlockType.SOFT_BLOCKED], + ) + + def test_upload_soft_blocked_stash_and_blocked_filter(self): + # Enough soft blocks for a stash + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + # Enough blocked versions for a filter + hard_to_soft = self._block_version(is_signed=True, block_type=BlockType.BLOCKED) + self._block_version(is_signed=True, block_type=BlockType.BLOCKED) + + self._test_upload_stash_and_filter( + enable_soft_blocking=False, + # Expect no stash because there are enough blocked versions for a filter + # and soft blocking is disabled + expected_stash=None, + # Expect a blocked filter + expected_filters=[BlockType.BLOCKED], + ) + + # Now try again with soft blocking enabled + self._test_upload_stash_and_filter( + enable_soft_blocking=True, + # Expect a stash and a blocked filter + expected_stash={ + # Expect no blocked versions because there are enough for a filter + 'blocked': 0, + # Expect a soft blocked version when there is one + # and soft blocking is enabled + 'softblocked': 1, + # There are no unblocked versions + 'unblocked': 0, + }, + # Expect a blocked filter + expected_filters=[BlockType.BLOCKED], + ) + + # Now try again converting a hard block to a soft block + hard_to_soft.update(block_type=BlockType.SOFT_BLOCKED) + + self._test_upload_stash_and_filter( + enable_soft_blocking=True, + expected_stash={ + # Expect a blocked version now, because we converted a hard block to a soft block + 'blocked': 1, + # Expect no soft blocks because there are enough for a filter + 'softblocked': 0, + # There are no unblocked versions + 'unblocked': 0, + }, + # Expect a soft blocked filter + expected_filters=[BlockType.SOFT_BLOCKED], + ) + + def test_upload_blocked_and_softblocked_filter(self): + # Enough blocked versions for a filter + self._block_version(is_signed=True, block_type=BlockType.BLOCKED) + self._block_version(is_signed=True, block_type=BlockType.BLOCKED) + # Enough soft blocks for a filter + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + + self._test_upload_stash_and_filter( + enable_soft_blocking=False, + expected_stash=None, + expected_filters=[BlockType.BLOCKED], + ) + + # Now try again with soft blocking enabled + self._test_upload_stash_and_filter( + enable_soft_blocking=True, + expected_stash=None, + expected_filters=[BlockType.BLOCKED, BlockType.SOFT_BLOCKED], + ) + + def test_upload_blocked_and_softblocked_stash(self): + # Enough blocked versions for a stash + self._block_version(is_signed=True, block_type=BlockType.BLOCKED) + # Enough soft blocks for a stash + self._block_version(is_signed=True, block_type=BlockType.SOFT_BLOCKED) + + self._test_upload_stash_and_filter( + enable_soft_blocking=False, + expected_stash={ + # Expect a blocked version + 'blocked': 1, + # Expect no soft blocks because soft blocking is disabled + 'softblocked': 0, + # There are no unblocked versions + 'unblocked': 0, + }, + expected_filters=[], + ) + + # Now try again with soft blocking enabled + self._test_upload_stash_and_filter( + enable_soft_blocking=True, + expected_stash={ + # We still have the blocked version + 'blocked': 1, + # Expect a soft blocked version because there is one + # and soft blocking is enabled + 'softblocked': 1, + # There are no unblocked versions + 'unblocked': 0, + }, + expected_filters=[], + ) + def test_remove_storage_if_no_update(self): """