more tests

pioreactor/background_jobs/monitor.py

@@ -46,7 +46,7 @@
     from pioreactor.utils.mock import MockHandle
 
 
-class TimedSet:
+class ExpiringMembershipSet:
     """
     A quick lookup to see if a object was recently added or not (min_time duration).
 
@@ -69,6 +69,9 @@ def add(self, item):
         now = time()
         self.data[item] = now
 
+    def clear(self):
+        self.data = dict()
+
 
 class Monitor(LongRunningBackgroundJob):
     """
@@ -130,7 +133,7 @@ def off(): # functions don't take any arguments, nothing is passed in
     led_in_use: bool = False
     _pre_button: list[Callable] = []
     _post_button: list[Callable] = []
-    _processed_topics_and_payloads = TimedSet(min_time=0.5)
+    _processed_topics_and_payloads = ExpiringMembershipSet(min_time=0.5)
 
     def __init__(self, unit: str, experiment: str) -> None:
         super().__init__(unit=unit, experiment=experiment)
@@ -159,9 +162,9 @@ def pretty_version(info: tuple) -> str:
         )
 
         self.button_down = False
-        # set up GPIO for accessing the button and changing the LED
 
         try:
+            # set up GPIO for accessing the button and changing the LED
             # if these fail, don't kill the entire job - sucks for onboarding.
             self._setup_GPIO()
         except Exception as e:
@@ -281,7 +284,7 @@ def self_checks(self) -> None:
             self.check_for_mqtt_connection_to_leader()
 
         # clean up our queue so it doesn't grow without bound.
-        self._processed_topics_and_payloads.data = dict()
+        self._processed_topics_and_payloads.clear()
 
     def check_for_correct_permissions(self) -> None:
         if whoami.is_testing_env():

pioreactor/tests/test_monitor.py

@@ -106,6 +106,55 @@ def test_run_job_with_monitor_deduplicates() -> None:
         assert len([msg for msg in bucket if "pio run example_plugin" in msg["message"]]) == 1
 
 
+def test_run_job_with_monitor_deduplicates_but_not_if_args_are_different() -> None:
+    unit = get_unit_name()
+    exp = UNIVERSAL_EXPERIMENT
+    cl = create_client()
+
+    with collect_all_logs_of_level("DEBUG", unit, exp) as bucket:
+        with Monitor(unit=unit, experiment=exp):
+            pause()
+
+            cl.publish(
+                f"pioreactor/{unit}/{get_assigned_experiment_name(unit)}/run/example_plugin",
+                r'{"test": "t"}',
+            )
+            cl.publish(
+                f"pioreactor/{unit}/{get_assigned_experiment_name(unit)}/run/example_plugin",
+                r"",
+            )
+            pause()
+            pause()
+
+        assert len([msg for msg in bucket if "pio run example_plugin" in msg["message"]]) == 2
+
+
+def test_run_job_with_monitor_deduplicates_passes_if_pause() -> None:
+    unit = get_unit_name()
+    exp = UNIVERSAL_EXPERIMENT
+    cl = create_client()
+
+    with collect_all_logs_of_level("DEBUG", unit, exp) as bucket:
+        with Monitor(unit=unit, experiment=exp):
+            pause()
+
+            cl.publish(
+                f"pioreactor/{unit}/{get_assigned_experiment_name(unit)}/run/example_plugin",
+                b"",
+            )
+            pause()
+            pause()
+
+            cl.publish(
+                f"pioreactor/{unit}/{get_assigned_experiment_name(unit)}/run/example_plugin",
+                b"",
+            )
+            pause()
+            pause()
+
+        assert len([msg for msg in bucket if "pio run example_plugin" in msg["message"]]) == 2
+
+
 def test_job_options_and_args_to_shell_command() -> None:
     m = Monitor
     assert (