test: spam protection multiple bursts

src/libs/common.py

@@ -38,3 +38,11 @@ def to_app_id(n: int) -> list:
     if n < 0:
         raise ValueError("Input must be an unsigned integer (non-negative)")
     return list(n.to_bytes(32, byteorder="big"))
+
+
+def random_divide_k(n, k):
+    if n < k:
+        raise ValueError("n must be at least k to split into k parts.")
+    cuts = sorted(random.sample(range(1, n), k - 1))
+    parts = [cuts[0]] + [cuts[i] - cuts[i - 1] for i in range(1, len(cuts))] + [n - cuts[-1]]
+    return parts

tests/dos_robustness/dos_robustness.py

@@ -1,6 +1,6 @@
 import pytest
 
-from src.libs.common import delay, to_app_id, to_index
+from src.libs.common import delay, to_app_id, to_index, random_divide_k
 from src.libs.custom_logger import get_custom_logger
 from src.steps.da import StepsDataAvailability
 from src.test_data import DATA_TO_DISPERSE
@@ -29,8 +29,10 @@ class TestDosRobustness(StepsDataAvailability):
 
     @pytest.mark.usefixtures("setup_2_node_cluster")
     def test_spam_protection_single_burst(self):
+        rate_limit = 1000
+        spam_per_burst = rate_limit + 10
         successful_dispersals = 0
-        for i in range(1000):
+        for i in range(spam_per_burst):
             try:
                 self.disperse_data(DATA_TO_DISPERSE[0], to_app_id(1), to_index(0), timeout_duration=0)
                 successful_dispersals = i
@@ -38,4 +40,29 @@ class TestDosRobustness(StepsDataAvailability):
                 logger.debug(f"Dispersal #{i+1} was not successful with error {ex}")
                 break
 
-        assert successful_dispersals < 1000, "All 1000 consecutive dispersals were successful without any constraint"
+        assert successful_dispersals <= rate_limit, "All consecutive dispersals were successful without any constraint"
+
+    @pytest.mark.usefixtures("setup_2_node_cluster")
+    def test_spam_protection_multiple_bursts(self):
+        time_interval = 60
+        rate_limit = 1000
+        bursts = 3
+        spam_per_burst = int((rate_limit + 10) / bursts)
+
+        waiting_intervals = random_divide_k(time_interval, bursts)
+
+        logger.debug(f"Waiting intervals: {waiting_intervals}")
+
+        successful_dispersals = 0
+        for b in range(bursts):
+            for i in range(spam_per_burst):
+                try:
+                    self.disperse_data(DATA_TO_DISPERSE[0], to_app_id(1), to_index(0), timeout_duration=0)
+                    successful_dispersals = i
+                except Exception as ex:
+                    logger.debug(f"Dispersal #{i+1} was not successful with error {ex}")
+                    break
+
+            delay(waiting_intervals[b])
+
+        assert successful_dispersals <= 1000, "All dispersals were successful without any constraint"