fix time_period latent crashes: treat missing time information as None

time_period is list[datetime] | None at runtime (None from old-format
HDF5 files, [] from the constructor), but several sites subscripted
it unguarded and crashed with TypeError/IndexError.

- annotate time_period as list[datetime.datetime] | None
- copy paths (get_accu_by_index, __iadd__ self-empty) use the guarded
  pattern already present in _copy_provenance
- merges in __add__/__iadd__ only combine min/max when both sides carry
  time information; otherwise the result is None (provenance unknown)
- write_to_tecmag raises ValueError when no time information is
  available instead of crashing while writing the header

Clears all remaining time_period pyrefly errors (72 -> 63). Unit tests
reproduce the crashes and pass with the fix.

Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
markusro committed 2026-10-18 15:12:57 +02:00
1 parent 27d05496ab
commit 804996cbd4
2 files changed
+120 -13

No files matched your search

+93
View File
@@ -430,5 +430,98 @@ class TestAccumulation(unittest.TestCase):
self.assertIsInstance(accu.y_square, list)
np.testing.assert_array_equal(accu.get_ysquare(0), y_square_2d[0])
def test_add_accumulations_without_time_period(self):
"""accu + accu without time information must not crash."""
accu1 = Accumulation()
accu1 += self.create_adc_result()
accu2 = Accumulation()
accu2 += self.create_adc_result()
accu1.time_period = None
accu2.time_period = []
r = accu1 + accu2
self.assertIsNone(r.time_period)
def test_iadd_accumulation_without_time_period(self):
"""accu += accu without time information must not crash (was IndexError)."""
accu1 = Accumulation()
accu1 += self.create_adc_result()
accu2 = Accumulation()
accu2 += self.create_adc_result()
accu1.time_period = []
accu2.time_period = []
accu1 += accu2
self.assertIsNone(accu1.time_period)
def test_add_adc_without_time_period(self):
"""accu + adc without time information must not crash."""
accu = Accumulation()
accu += self.create_adc_result()
accu.time_period = []
r = accu + self.create_adc_result()
self.assertIsNone(r.time_period)
def test_time_period_merged_on_add(self):
"""accu + accu with both time periods set merges to min/max."""
accu1 = Accumulation()
accu1 += self.create_adc_result()
accu2 = Accumulation()
accu2 += self.create_adc_result()
accu1.time_period = [datetime(2023, 1, 3), datetime(2023, 1, 5)]
accu2.time_period = [datetime(2023, 1, 2), datetime(2023, 1, 7)]
r = accu1 + accu2
self.assertEqual(r.time_period, [datetime(2023, 1, 2), datetime(2023, 1, 7)])
def test_time_period_merged_with_adc_job_date(self):
"""accu + adc merges self.time_period with other.job_date."""
accu = Accumulation()
accu += self.create_adc_result()
accu.time_period = [datetime(2023, 1, 3), datetime(2023, 1, 5)]
adc = self.create_adc_result()
adc.job_date = datetime(2023, 1, 1)
r = accu + adc
self.assertEqual(r.time_period, [datetime(2023, 1, 1), datetime(2023, 1, 5)])
def test_iadd_adc_without_job_date_sets_time_period_none(self):
"""empty accu += adc without job_date gets time_period None."""
accu = Accumulation()
adc = self.create_adc_result()
adc.job_date = None
accu += adc
self.assertIsNone(accu.time_period)
def test_get_accu_by_index_without_time_period(self):
"""get_accu_by_index without time information must not crash."""
accu = Accumulation()
adc = self.create_adc_result()
adc.index = [(0, 1), (1, 2)]
accu += adc
accu.time_period = None
r = accu.get_accu_by_index(0)
self.assertIsNone(r.time_period)
def test_write_to_tecmag_without_time_period_raises(self):
"""write_to_tecmag without time information must raise ValueError."""
accu = Accumulation()
accu += self.create_adc_result()
accu.time_period = None
self.assertRaises(ValueError, accu.write_to_tecmag, "/tmp/damaris_tecmag_no_time.tnt")
self.assertTrue(accu.lock.acquire(timeout=1), "lock leaked after ValueError in write_to_tecmag")
accu.lock.release()
if __name__ == "__main__":
unittest.main()