Fix tests against qcodes main while keeping 0.58.0 support

src/qcodes_loop/loops.py

@@ -43,7 +43,7 @@
 import logging
 import time
 from collections.abc import Sequence
-from datetime import datetime
+from datetime import UTC, datetime
 from typing import Optional
 
 import numpy as np
@@ -69,6 +69,17 @@
 _tprint_times: dict[str, float] = {}
 
 
+def _timestamp_now() -> str:
+    """
+    Current local time as a timezone aware ISO formatted string.
+
+    The format matches the one QCoDeS uses for the ``ts`` entry of a
+    parameter snapshot so that timestamps stored in the metadata of a
+    :class:`qcodes_loop.data.data_set.DataSet` are directly comparable.
+    """
+    return datetime.now(UTC).astimezone().isoformat(sep=" ", timespec="seconds")
+
+
 def wait_secs(finish_clock: float) -> float:
     """
     Calculate the number of seconds until a given clock time.
@@ -795,7 +806,7 @@ def run(
         # information about the loop definition is in its snapshot
         data_set.add_metadata({"loop": self.snapshot()})
         # then add information about how and when it was run
-        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        ts = _timestamp_now()
         data_set.add_metadata(
             {
                 "loop": {
@@ -870,7 +881,7 @@ def _run_wrapper(self, *args, **kwargs):
                 # TODO (giulioungaretti) WTF?
                 # somehow this does not show up in the data_set returned by
                 # run(), but it is saved to the metadata
-                ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+                ts = _timestamp_now()
                 self.data_set.add_metadata({"loop": {"ts_end": ts}})
                 self.data_set.finalize()
 

src/qcodes_loop/tests/test_loop.py

@@ -1,5 +1,5 @@
 import os
-from datetime import datetime
+from datetime import UTC, datetime
 from unittest import TestCase
 from unittest.mock import patch
 
@@ -17,6 +17,20 @@
 from qcodes_loop.sweep_values import Sweeper
 
 
+def _timestamp_variants():
+    """
+    Both spellings of "now" that may show up in a snapshot.
+
+    QCoDeS >= 0.59 renders parameter snapshot timestamps as timezone aware
+    ISO strings, older versions used a naive ``%Y-%m-%d %H:%M:%S`` string.
+    """
+    now = datetime.now(UTC).astimezone()
+    return (
+        now.isoformat(sep=" ", timespec="seconds"),
+        now.strftime("%Y-%m-%d %H:%M:%S"),
+    )
+
+
 class NanReturningParameter(MultiParameter):
     def __init__(self, name, instrument, names=("first", "second"), shapes=((), ())):
 
@@ -387,7 +401,7 @@ def g():
             g_calls.append(1)
 
         breaker = BreakIf(lambda: self.p1() >= 3)
-        ts1 = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        ts1 = _timestamp_variants()
         # evaluate param snapshots now since later value will change
         p1snap = self.p1.snapshot()
         self.p2.set(2)
@@ -400,7 +414,7 @@ def g():
             .then(Task(self.p1.set, 2), Wait(0.01), Task(f))
             .run_temp()
         )
-        ts2 = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        ts2 = _timestamp_variants()
 
         self.assertEqual(repr(data.p1.tolist()), repr([1.0, 2.0, 3.0, nan, nan]))
         self.assertEqual(self.p1.get(), 2)
@@ -411,10 +425,11 @@ def g():
         loopmeta = data.metadata["loop"]
         # assuming the whole loop takes < 1 sec, all timestamps
         # should each be the same as one of the bounding times
-        self.check_snap_ts(loopmeta, "ts_start", (ts1, ts2))
-        self.check_snap_ts(loopmeta, "ts_end", (ts1, ts2))
-        self.check_snap_ts(loopmeta["sweep_values"]["parameter"], "ts", (ts1, ts2))
-        self.check_snap_ts(loopmeta["actions"][0], "ts", (ts1, ts2))
+        ts_set = (*ts1, *ts2)
+        self.check_snap_ts(loopmeta, "ts_start", ts_set)
+        self.check_snap_ts(loopmeta, "ts_end", ts_set)
+        self.check_snap_ts(loopmeta["sweep_values"]["parameter"], "ts", ts_set)
+        self.check_snap_ts(loopmeta["actions"][0], "ts", ts_set)
         del p1snap["ts"], p2snap["ts"], p3snap["ts"]
 
         self.assertEqual(
@@ -503,7 +518,13 @@ def test_halt(self):
         [self.res.append(float("nan")) for i in range(0, abort_after - 1)]
 
         p1 = AbortingGetter(
-            "p1", count=abort_after, vals=Numbers(-10, 10), set_cmd=None
+            "p1",
+            count=abort_after,
+            vals=Numbers(-10, 10),
+            set_cmd=None,
+            # prime the cache so that the snapshots taken while setting up the
+            # loop are satisfied from the cache and do not consume the counter
+            initial_cache_value=0,
         )
         loop = Loop(Sweeper(p1).sweep(0, abort_after, 1), 0.005).each(p1)
         # we want to test what's in data, so get it ahead of time

src/qcodes_loop/tests/test_measure.py

@@ -28,8 +28,8 @@ def test_simple_scalar(self):
         self.assertEqual(len(meta["actions"]), 1)
         self.assertFalse(meta["use_threads"])
 
-        ts_start = datetime.strptime(meta["ts_start"], "%Y-%m-%d %H:%M:%S")
-        ts_end = datetime.strptime(meta["ts_end"], "%Y-%m-%d %H:%M:%S")
+        ts_start = datetime.fromisoformat(meta["ts_start"])
+        ts_end = datetime.fromisoformat(meta["ts_end"])
         self.assertGreaterEqual(ts_end, ts_start)
 
     def test_simple_array(self):