mark_as_restored

oc_ocdm/graph/graph_entity.py

@@ -205,6 +205,7 @@ def __init__(self, g: Graph, g_set: GraphSet, res: URIRef = None, res_type: URIR
         # FLAGS
         self._to_be_deleted: bool = False
         self._was_merged: bool = False
+        self._is_restored: bool = False
 
         # If res was not specified, create from scratch the URI reference for this entity,
         # otherwise use the provided one
@@ -250,6 +251,23 @@ def was_merged(self) -> bool:
     def merge_list(self) -> Tuple[GraphEntity]:
         return self._merge_list
 
+    @property 
+    def is_restored(self) -> bool:
+        """Indicates if this entity was restored after being deleted."""
+        return self._is_restored
+
+    def mark_as_restored(self) -> None:
+        """
+        Marks an entity as being restored after deletion.
+                
+        This state signals to the provenance system that:
+        - No new invalidation time should be generated for the previous snapshot
+        - The original deletion snapshot's invalidation time should be preserved
+        - The entity should be treated as restored rather than newly created
+        """
+        self._to_be_deleted = False
+        self._is_restored = True
+
     def mark_as_to_be_deleted(self) -> None:
         # Here we must REMOVE triples pointing
         # to 'self' [THIS CANNOT BE UNDONE]:
@@ -332,6 +350,7 @@ def commit_changes(self):
         else:
             for triple in self.g.triples((self.res, None, None)):
                 self.preexisting_graph.add(triple)
+        self._is_restored = False
         self._to_be_deleted = False
         self._was_merged = False
         self._merge_list = tuple()

oc_ocdm/prov/prov_set.py

@@ -189,6 +189,17 @@ def generate_provenance(self, c_time: float = None) -> set:
                     cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been deleted.")
                     cur_snapshot.has_update_action(update_query)
                     modified_entities.add(cur_subj.res)
+                elif cur_subj.is_restored:
+                    # RESTORATION SNAPSHOT
+                    last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
+                    # Don't set invalidation time on previous snapshot for restorations
+
+                    cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
+                    cur_snapshot.derives_from(last_snapshot)
+                    cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been restored.")
+                    if update_query:
+                        cur_snapshot.has_update_action(update_query)
+                    modified_entities.add(cur_subj.res)
                 elif was_modified:
                     # MODIFICATION SNAPSHOT
                     last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)

oc_ocdm/test/prov/test_prov_set.py

@@ -52,7 +52,6 @@ def test_creation_merged_entity(self):
         a.merge(b, prefer_self=True)
 
         result = self.prov_set.generate_provenance(self.cur_time)
-        print(a.res, b.res, URIRef(a.res + '/prov/se/1'), self.prov_set.get_entity(URIRef(a.res + '/prov/se/1')), self.prov_set.res_to_entity)
         se_a = self.prov_set.get_entity(URIRef(a.res + '/prov/se/1'))
         self.assertIsNotNone(se_a)
         self.assertIsInstance(se_a, SnapshotEntity)
@@ -248,6 +247,42 @@ def test_retrieve_last_snapshot(self):
         prov_subject = URIRef('https://w3id.org/oc/corpus/br/abc')
         self.assertRaises(ValueError, self.prov_set._retrieve_last_snapshot, prov_subject)
 
+    def test_restore_deleted_entity(self):
+        # Create and delete an entity first
+        a = self.graph_set.add_br(self.resp_agent)
+
+        # Generate provenance for dcreation
+        self.prov_set.generate_provenance(self.cur_time)
+
+        a.mark_as_to_be_deleted()
+
+        # Generate provenance for deletion
+        self.prov_set.generate_provenance(self.cur_time)
+        deletion_time = self.cur_time_str
+
+        # Get the deletion snapshot
+        se_a_2: SnapshotEntity = self.prov_set.get_entity(URIRef(a.res + '/prov/se/2'))
+        self.assertIsNotNone(se_a_2)
+        self.assertEqual(deletion_time, se_a_2.get_generation_time())
+        self.assertEqual(deletion_time, se_a_2.get_invalidation_time())
+
+        # Now restore the entity
+        a.mark_as_restored()
+        a.has_title("Restored Title")  # Add some modification
+
+        # Generate provenance after restoration
+        restoration_time = "2020-12-08T21:17:39+00:00"
+        result = self.prov_set.generate_provenance(1607462259.846196)  # One day later
+
+        # Check the restoration snapshot
+        se_a_3: SnapshotEntity = self.prov_set.get_entity(URIRef(a.res + '/prov/se/3'))
+        self.assertIsNotNone(se_a_3)
+        self.assertEqual(restoration_time, se_a_3.get_generation_time())
+        self.assertIsNone(se_a_3.get_invalidation_time())  # No invalidation time for restoration
+        self.assertEqual(f"The entity '{a.res}' has been restored.", se_a_3.get_description())
+        self.assertSetEqual({se_a_2}, set(se_a_3.get_derives_from()))
+        self.assertIsNotNone(se_a_3.get_update_action())
+
 class TestProvSetWorkflow(unittest.TestCase):
     def setUp(self):
         self.test_dir = os.path.join('oc_ocdm', 'test', 'prov', 'provset_workflow_data') + os.sep
@@ -348,5 +383,6 @@ def test_full_workflow(self):
         # self.assertEqual(3, len(derived_from))
         # self.assertTrue(all(se.res in derived_from for se in [se_a, se_b, se_c]))
 
+
 if __name__ == '__main__':
     unittest.main()