From 0078761dc94c33c785557df09bb86411ef8f6f46 Mon Sep 17 00:00:00 2001
From: Ayush Chauhan <57655135+ayush-san@users.noreply.github.com>
Date: Thu, 30 Jul 2020 23:51:45 +0530
Subject: [PATCH] fix: cypher statement param issue in
 Neo4jStalenessRemovalTask (#307)

* fix: cypher statement param issue

* test: update existing test cases according to change in marker param
---
 .../task/neo4j_staleness_removal_task.py       |  4 ++--
 .../task/test_neo4j_staleness_removal_task.py  | 18 +++++++++---------
 2 files changed, 11 insertions(+), 11 deletions(-)

diff --git a/databuilder/task/neo4j_staleness_removal_task.py b/databuilder/task/neo4j_staleness_removal_task.py
index 583251c51..a0cf4b932 100644
--- a/databuilder/task/neo4j_staleness_removal_task.py
+++ b/databuilder/task/neo4j_staleness_removal_task.py
@@ -91,7 +91,7 @@ def init(self, conf):
             self.ms_to_expire = conf.get_int(MS_TO_EXPIRE)
             if self.ms_to_expire < conf.get_int(MIN_MS_TO_EXPIRE):
                 raise Exception('{} is too small'.format(MS_TO_EXPIRE))
-            self.marker = '(timestamp() - {})'.format(conf.get_int(MS_TO_EXPIRE))
+            self.marker = self.ms_to_expire
         else:
             self.marker = conf.get_string(JOB_PUBLISH_TAG)
 
@@ -144,7 +144,7 @@ def _decorate_staleness(self, statement):
         """
         if self.ms_to_expire:
             return statement.format(textwrap.dedent("""
-            n.publisher_last_updated_epoch_ms < ${marker}
+            n.publisher_last_updated_epoch_ms < (timestamp() - ${marker})
             OR NOT EXISTS(n.publisher_last_updated_epoch_ms)""".format(marker=MARKER_VAR_NAME)))
 
         return statement.format(textwrap.dedent("""
diff --git a/tests/unit/task/test_neo4j_staleness_removal_task.py b/tests/unit/task/test_neo4j_staleness_removal_task.py
index afd1c2ddc..18cf5e74f 100644
--- a/tests/unit/task/test_neo4j_staleness_removal_task.py
+++ b/tests/unit/task/test_neo4j_staleness_removal_task.py
@@ -136,7 +136,7 @@ def test_marker(self):
 
             task.init(job_config)
             self.assertIsNotNone(task.ms_to_expire)
-            self.assertEqual(task.marker, '(timestamp() - 86400000)')
+            self.assertEqual(task.marker, 86400000)
 
     def test_validation_statement_publish_tag(self):
         with patch.object(GraphDatabase, 'driver'), patch.object(Neo4jStalenessRemovalTask, '_execute_cypher_query') \
@@ -215,22 +215,22 @@ def test_validation_statement_ms_to_expire(self):
             RETURN head(node) as type, count
             """))
 
-            mock_execute.assert_any_call(param_dict={'marker': '(timestamp() - 9876543210)'},
+            mock_execute.assert_any_call(param_dict={'marker': 9876543210},
                                          statement=textwrap.dedent("""
             MATCH (n)
             WHERE 
-            n.publisher_last_updated_epoch_ms < $marker
+            n.publisher_last_updated_epoch_ms < (timestamp() - $marker)
             OR NOT EXISTS(n.publisher_last_updated_epoch_ms)
             WITH DISTINCT labels(n) as node, count(*) as count
             RETURN head(node) as type, count
             """))
 
             task._validate_relation_staleness_pct()
-            mock_execute.assert_any_call(param_dict={'marker': '(timestamp() - 9876543210)'},
+            mock_execute.assert_any_call(param_dict={'marker': 9876543210},
                                          statement=textwrap.dedent("""
             MATCH ()-[n]-()
             WHERE 
-            n.publisher_last_updated_epoch_ms < $marker
+            n.publisher_last_updated_epoch_ms < (timestamp() - $marker)
             OR NOT EXISTS(n.publisher_last_updated_epoch_ms)
             RETURN type(n) as type, count(*) as count
             """))
@@ -313,11 +313,11 @@ def test_delete_statement_ms_to_expire(self):
             task._delete_stale_relations()
 
             mock_execute.assert_any_call(dry_run=False,
-                                         param_dict={'marker': '(timestamp() - 9876543210)', 'batch_size': 100},
+                                         param_dict={'marker': 9876543210, 'batch_size': 100},
                                          statement=textwrap.dedent("""
             MATCH (n:Foo)
             WHERE 
-            n.publisher_last_updated_epoch_ms < $marker
+            n.publisher_last_updated_epoch_ms < (timestamp() - $marker)
             OR NOT EXISTS(n.publisher_last_updated_epoch_ms)
             WITH n LIMIT $batch_size
             DETACH DELETE (n)
@@ -325,11 +325,11 @@ def test_delete_statement_ms_to_expire(self):
             """))
 
             mock_execute.assert_any_call(dry_run=False,
-                                         param_dict={'marker': '(timestamp() - 9876543210)', 'batch_size': 100},
+                                         param_dict={'marker': 9876543210, 'batch_size': 100},
                                          statement=textwrap.dedent("""
             MATCH ()-[n:BAR]-()
             WHERE 
-            n.publisher_last_updated_epoch_ms < $marker
+            n.publisher_last_updated_epoch_ms < (timestamp() - $marker)
             OR NOT EXISTS(n.publisher_last_updated_epoch_ms)
             WITH n LIMIT $batch_size
             DELETE n