Fix module-level side effects and global random seeding in univariate ML anomaly tests

* Remove top-level dataset generation, stdout printing, and global seeding from perf_test.py by moving setup logic into PerfTest.setUpClass.

* Replace calls to global random.seed() in perf_test.py, mean_test.py, quantile_test.py, and stdev_test.py with isolated random.Random instances.
  This prevents state leakage and unneeded computation when modules are imported during test collection.

Author: Abacn

sdks/python/apache_beam/ml/anomaly/univariate/mean_test.py

@@ -70,13 +70,13 @@ def test_with_float64_max(self):
 
   def test_accuracy_fuzz(self):
     seed = int(time.time())
-    random.seed(seed)
+    rng = random.Random(seed)
     print("Random seed: %d" % seed)
 
     for _ in range(10):
       numbers = []
       for _ in range(5000):
-        numbers.append(random.randint(0, 1000))
+        numbers.append(rng.randint(0, 1000))
 
       with warnings.catch_warnings(record=False):
         warnings.simplefilter("ignore")
@@ -140,13 +140,13 @@ def test_with_float64_max(self, tracker):
 
   def test_accuracy_fuzz(self):
     seed = int(time.time())
-    random.seed(seed)
+    rng = random.Random(seed)
     print("Random seed: %d" % seed)
 
     for _ in range(10):
       numbers = []
       for _ in range(5000):
-        numbers.append(random.randint(0, 1000))
+        numbers.append(rng.randint(0, 1000))
 
       t1 = IncSlidingMeanTracker(100)
       t2 = SimpleSlidingMeanTracker(100)

sdks/python/apache_beam/ml/anomaly/univariate/perf_test.py

@@ -27,54 +27,57 @@
 from apache_beam.ml.anomaly.univariate.quantile import *
 from apache_beam.ml.anomaly.univariate.stdev import *
 
-seed_value_time = int(time.time())
-random.seed(seed_value_time)
-print(f"{'Seed value':32s}{seed_value_time}")
-
-numbers = []
-for _ in range(50000):
-  numbers.append(random.randint(0, 1000))
-
 
 def run_tracker(tracker, numbers):
   for i in range(len(numbers)):
     tracker.push(numbers[i])
     _ = tracker.get()
 
 
-def print_result(tracker, number=10, repeat=5):
-  runtimes = timeit.repeat(
-      lambda: run_tracker(tracker, numbers), number=number, repeat=repeat)
-  mean = statistics.mean(runtimes)
-  sd = statistics.stdev(runtimes)
-  print(f"{tracker.__class__.__name__:32s}{mean:.6f} ± {sd:.6f}")
+class PerfTest(unittest.TestCase):
+  @classmethod
+  def setUpClass(cls):
+    seed_value_time = int(time.time())
+    rng = random.Random(seed_value_time)
+    print(f"{'Seed value':32s}{seed_value_time}")
 
+    cls.numbers = []
+    for _ in range(50000):
+      cls.numbers.append(rng.randint(0, 1000))
+
+  def print_result(self, tracker, number=10, repeat=5):
+    runtimes = timeit.repeat(
+        lambda: run_tracker(tracker, self.numbers),
+        number=number,
+        repeat=repeat)
+    mean = statistics.mean(runtimes)
+    sd = statistics.stdev(runtimes)
+    print(f"{tracker.__class__.__name__:32s}{mean:.6f} ± {sd:.6f}")
 
-class PerfTest(unittest.TestCase):
   def test_mean_perf(self):
     print()
-    print_result(IncLandmarkMeanTracker())
-    print_result(IncSlidingMeanTracker(100))
+    self.print_result(IncLandmarkMeanTracker())
+    self.print_result(IncSlidingMeanTracker(100))
     # SimpleSlidingMeanTracker (numpy-based batch approach) is an order of
     # magnitude slower than other methods. To prevent excessively long test
     # runs, we reduce the number of repetitions.
-    print_result(SimpleSlidingMeanTracker(100), number=1)
+    self.print_result(SimpleSlidingMeanTracker(100), number=1)
 
   def test_stdev_perf(self):
     print()
-    print_result(IncLandmarkStdevTracker())
-    print_result(IncSlidingStdevTracker(100))
+    self.print_result(IncLandmarkStdevTracker())
+    self.print_result(IncSlidingStdevTracker(100))
     # Same as test_mean_perf, we reduce the number of repetitions here.
-    print_result(SimpleSlidingStdevTracker(100), number=1)
+    self.print_result(SimpleSlidingStdevTracker(100), number=1)
 
   def test_quantile_perf(self):
     print()
     with warnings.catch_warnings(record=False):
       warnings.simplefilter("ignore")
-      print_result(BufferedLandmarkQuantileTracker(0.5))
-    print_result(BufferedSlidingQuantileTracker(100, 0.5))
+      self.print_result(BufferedLandmarkQuantileTracker(0.5))
+    self.print_result(BufferedSlidingQuantileTracker(100, 0.5))
     # Same as test_mean_perf, we reduce the number of repetitions here.
-    print_result(SimpleSlidingQuantileTracker(100, 0.5), number=1)
+    self.print_result(SimpleSlidingQuantileTracker(100, 0.5), number=1)
 
 
 if __name__ == '__main__':

sdks/python/apache_beam/ml/anomaly/univariate/quantile_test.py

@@ -68,13 +68,13 @@ def test_with_nan(self):
 
   def test_accuracy_fuzz(self):
     seed = int(time.time())
-    random.seed(seed)
+    rng = random.Random(seed)
     print("Random seed: %d" % seed)
 
     def _accuracy_helper():
       numbers = []
       for _ in range(5000):
-        numbers.append(random.randint(0, 1000))
+        numbers.append(rng.randint(0, 1000))
 
       with warnings.catch_warnings(record=False):
         warnings.simplefilter("ignore")
@@ -138,13 +138,13 @@ def test_with_nan(self, tracker):
 
   def test_accuracy_fuzz(self):
     seed = int(time.time())
-    random.seed(seed)
+    rng = random.Random(seed)
     print("Random seed: %d" % seed)
 
     def _accuracy_helper():
       numbers = []
       for _ in range(5000):
-        numbers.append(random.randint(0, 1000))
+        numbers.append(rng.randint(0, 1000))
 
       t1 = BufferedSlidingQuantileTracker(100, 0.1)
       t2 = SimpleSlidingQuantileTracker(100, 0.1)

sdks/python/apache_beam/ml/anomaly/univariate/stdev_test.py

@@ -66,13 +66,13 @@ def test_with_nan(self):
 
   def test_accuracy_fuzz(self):
     seed = int(time.time())
-    random.seed(seed)
+    rng = random.Random(seed)
     print("Random seed: %d" % seed)
 
     for _ in range(10):
       numbers = []
       for _ in range(5000):
-        numbers.append(random.randint(0, 1000))
+        numbers.append(rng.randint(0, 1000))
 
       t1 = IncLandmarkStdevTracker()
       t2 = SimpleSlidingStdevTracker(len(numbers))
@@ -135,13 +135,13 @@ def test_stdev_with_nan(self, tracker):
 
   def test_accuracy_fuzz(self):
     seed = int(time.time())
-    random.seed(seed)
+    rng = random.Random(seed)
     print("Random seed: %d" % seed)
 
     for _ in range(10):
       numbers = []
       for _ in range(5000):
-        numbers.append(random.randint(0, 1000))
+        numbers.append(rng.randint(0, 1000))
 
       t1 = IncSlidingStdevTracker(100)
       t2 = SimpleSlidingStdevTracker(100)