Add tests for shape of stash

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):
         """