Add tests for downsample

tests/test_model/test_handling_data.py

@@ -12,7 +12,7 @@ from copy import copy as shallow_copy
 import tracemalloc
 
 from unittest import TestCase, expectedFailure
-from unittest.mock import patch
+from unittest.mock import patch, DEFAULT
 
 from obspy.core import Trace, read as read_ms
 import numpy as np
@@ -31,7 +31,9 @@ from sohstationviewer.model.handling_data import (
     checkWFChan,
     checkSOHChan,
     sortData,
-    squash_gaps
+    squash_gaps,
+    downsample,
+    chunk_minmax
 )
 from sohstationviewer.model.reftek.from_rt2ms.core import Reftek130
 
@@ -775,3 +777,53 @@ class TestSquashGaps(TestCase):
     def test_no_gap_in_data(self):
         gaps = []
         self.assertListEqual(squash_gaps(gaps), [])
+
+
+class TestDownsample(TestCase):
+    def setUp(self) -> None:
+        patcher = patch('sohstationviewer.model.handling_data.chunk_minmax')
+        self.addCleanup(patcher.stop)
+        self.mock_chunk_minmax = patcher.start()
+        self.times = np.arange(1000)
+        self.data = np.arange(1000)
+
+    def test_first_downsample_step_remove_enough_points(self):
+        req_points = 999
+        downsample(self.times, self.data, req_points)
+        self.assertFalse(self.mock_chunk_minmax.called)
+
+    def test_second_downsample_step_required(self):
+        req_points = 1
+        downsample(self.times, self.data, req_points)
+        self.assertTrue(self.mock_chunk_minmax.called)
+        times, data, _ = self.mock_chunk_minmax.call_args[0]
+        self.assertIsNot(times, self.times)
+        self.assertIsNot(data, self.data)
+
+    def test_requested_points_greater_than_data_size(self):
+        req_points = 10000
+        times, data, = downsample(self.times, self.data, req_points)
+        self.assertFalse(self.mock_chunk_minmax.called)
+        # Check that we did not do any processing on the times and data arrays.
+        # This ensures that we don't do two unneeded copy operations.
+        self.assertIs(times, self.times)
+        self.assertIs(data, self.data)
+
+    def test_requested_points_is_zero(self):
+        req_points = 0
+        downsample(self.times, self.data, req_points)
+        self.assertTrue(self.mock_chunk_minmax.called)
+        times, data, _ = self.mock_chunk_minmax.call_args[0]
+        self.assertIsNot(times, self.times)
+        self.assertIsNot(data, self.data)
+
+    def test_empty_times_and_data(self):
+        req_points = 1000
+        self.times = np.empty((0, 0))
+        self.data = np.empty((0, 0))
+        times, data = downsample(self.times, self.data, req_points)
+        self.assertFalse(self.mock_chunk_minmax.called)
+        # Check that we did not do any processing on the times and data arrays.
+        # This ensures that we don't do two unneeded copy operations.
+        self.assertIs(times, self.times)
+        self.assertIs(data, self.data)