Extend Conditions POST test to use new JSON interface.

[plomtask] / plomtask / versioned_attributes.py

diff --git a/plomtask/versioned_attributes.py b/plomtask/versioned_attributes.py
index ab39df098059e416c7627b5fcefecae2668d7c06..cc42bbc40f761c35715f11b0ca4f61ba67dbe579 100644 (file)
--- a/plomtask/versioned_attributes.py
+++ b/plomtask/versioned_attributes.py
@@ -4,6 +4,7 @@ from typing import Any
 from sqlite3 import Row
 from time import sleep
 from plomtask.db import DatabaseConnection
+from plomtask.exceptions import HandledException, BadFormatException
 
 TIMESTAMP_FMT = '%Y-%m-%d %H:%M:%S.%f'
 
@@ -18,6 +19,18 @@ class VersionedAttribute:
         self.default = default
         self.history: dict[str, str | float] = {}
 
+    def __hash__(self) -> int:
+        history_tuples = tuple((k, v) for k, v in self.history.items())
+        hashable = (self.parent.id_, self.table_name, self.default,
+                    history_tuples)
+        return hash(hashable)
+
+    @property
+    def as_dict(self) -> dict[str, object]:
+        """Return self as (json.dumps-coompatible) dict."""
+        d = {'parent_id': self.parent.id_, 'history': self.history}
+        return d
+
     @property
     def _newest_timestamp(self) -> str:
         """Return most recent timestamp."""
@@ -30,6 +43,32 @@ class VersionedAttribute:
             return self.default
         return self.history[self._newest_timestamp]
 
+    def reset_timestamp(self, old_str: str, new_str: str) -> None:
+        """Rename self.history key (timestamp) old to new.
+
+        Chronological sequence of keys must be preserved, i.e. cannot move
+        key before earlier or after later timestamp.
+        """
+        try:
+            new = datetime.strptime(new_str, TIMESTAMP_FMT)
+            old = datetime.strptime(old_str, TIMESTAMP_FMT)
+        except ValueError as exc:
+            raise BadFormatException('Timestamp of illegal format.') from exc
+        timestamps = list(self.history.keys())
+        if old_str not in timestamps:
+            raise HandledException(f'Timestamp {old} not found in history.')
+        sorted_timestamps = sorted([datetime.strptime(t, TIMESTAMP_FMT)
+                                    for t in timestamps])
+        expected_position = sorted_timestamps.index(old)
+        sorted_timestamps.remove(old)
+        sorted_timestamps += [new]
+        sorted_timestamps.sort()
+        if sorted_timestamps.index(new) != expected_position:
+            raise HandledException('Timestamp not respecting chronology.')
+        value = self.history[old_str]
+        del self.history[old_str]
+        self.history[new_str] = value
+
     def set(self, value: str | float) -> None:
         """Add to self.history if and only if not same value as newest one.
 
@@ -51,6 +90,8 @@ class VersionedAttribute:
 
     def at(self, queried_time: str) -> str | float:
         """Retrieve value of timestamp nearest queried_time from the past."""
+        if len(queried_time) == 10:
+            queried_time += ' 23:59:59.999'
         sorted_timestamps = sorted(self.history.keys())
         if 0 == len(sorted_timestamps):
             return self.default
@@ -66,3 +107,7 @@ class VersionedAttribute:
         db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_,
                                   [[item[0], item[1]]
                                    for item in self.history.items()])
+
+    def remove(self, db_conn: DatabaseConnection) -> None:
+        """Remove from DB."""
+        db_conn.delete_where(self.table_name, 'parent', self.parent.id_)