Repository: lyft/cartography

Issue:
Failure during the cleanup job of EC2 snapshots
I've experienced a new failure during the cleanup job of EC2 snapshots:

```
neobolt.exceptions.ClientError: Unable to load NODE with id <redacted>.
```

Full log:

```


ERROR:cartography.graph.job:Unhandled error while executing statement in job 'cleanup EBS Snapshots': Unable to load NODE with id <redacted>.
ERROR:cartography.sync:Unhandled exception during sync stage 'aws'
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/cartography/sync.py", line 73, in run
    stage_func(neo4j_session, config)
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 66, in timed
    return method(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/__init__.py", line 202, in start_aws_ingestion
    _sync_multiple_accounts(
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/__init__.py", line 139, in _sync_multiple_accounts
    _sync_one_account(
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/__init__.py", line 57, in _sync_one_account
    RESOURCE_FUNCTIONS[func_name](**sync_args)
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 66, in timed
    return method(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/ec2/snapshots.py", line 117, in sync_ebs_snapshots
    cleanup_snapshots(neo4j_session, common_job_parameters)
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 66, in timed
    return method(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/ec2/snapshots.py", line 99, in cleanup_snapshots
    run_cleanup_job(
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 32, in run_cleanup_job
    GraphJob.run_from_json(
  File "/usr/local/lib/python3.9/site-packages/cartography/graph/job.py", line 95, in run_from_json
    job.run(neo4j_session)
  File "/usr/local/lib/python3.9/site-packages/cartography/graph/job.py", line 45, in run
    stm.run(neo4j_session)
  File "/usr/local/lib/python3.9/site-packages/cartography/graph/statement.py", line 48, in run
    tx: neo4j.Transaction = session.begin_transaction()
  File "/usr/local/lib/python3.9/site-packages/neo4j/__init__.py", line 614, in begin_transaction
    self._open_transaction(metadata=metadata, timeout=timeout)
  File "/usr/local/lib/python3.9/site-packages/neo4j/__init__.py", line 619, in _open_transaction
    self._connect(access_mode)
  File "/usr/local/lib/python3.9/site-packages/neo4j/__init__.py", line 381, in _connect
    self._connection.sync()
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 527, in sync
    detail_delta, summary_delta = self.fetch()
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 419, in fetch
    return self._fetch()
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 461, in _fetch
    response.on_failure(summary_metadata or {})
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 755, in on_failure
    raise CypherError.hydrate(**metadata)
neobolt.exceptions.ClientError: Unable to load NODE with id <redacted>.
Traceback (most recent call last):
  File "/usr/local/bin/cartography", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.9/site-packages/cartography/cli.py", line 441, in main
    return CLI(default_sync, prog='cartography').main(argv)
  File "/usr/local/lib/python3.9/site-packages/cartography/cli.py", line 421, in main
    return cartography.sync.run_with_config(self.sync, config)
  File "/usr/local/lib/python3.9/site-packages/cartography/sync.py", line 150, in run_with_config
    return sync.run(neo4j_driver, config)
  File "/usr/local/lib/python3.9/site-packages/cartography/sync.py", line 73, in run
    stage_func(neo4j_session, config)
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 66, in timed
    return method(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/__init__.py", line 202, in start_aws_ingestion
    _sync_multiple_accounts(
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/__init__.py", line 139, in _sync_multiple_accounts
    _sync_one_account(
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/__init__.py", line 57, in _sync_one_account
    RESOURCE_FUNCTIONS[func_name](**sync_args)
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 66, in timed
    return method(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/ec2/snapshots.py", line 117, in sync_ebs_snapshots
    cleanup_snapshots(neo4j_session, common_job_parameters)
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 66, in timed
    return method(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/cartography/intel/aws/ec2/snapshots.py", line 99, in cleanup_snapshots
    run_cleanup_job(
  File "/usr/local/lib/python3.9/site-packages/cartography/util.py", line 32, in run_cleanup_job
    GraphJob.run_from_json(
  File "/usr/local/lib/python3.9/site-packages/cartography/graph/job.py", line 95, in run_from_json
    job.run(neo4j_session)
  File "/usr/local/lib/python3.9/site-packages/cartography/graph/job.py", line 45, in run
    stm.run(neo4j_session)
  File "/usr/local/lib/python3.9/site-packages/cartography/graph/statement.py", line 48, in run
    tx: neo4j.Transaction = session.begin_transaction()
  File "/usr/local/lib/python3.9/site-packages/neo4j/__init__.py", line 614, in begin_transaction
    self._open_transaction(metadata=metadata, timeout=timeout)
  File "/usr/local/lib/python3.9/site-packages/neo4j/__init__.py", line 619, in _open_transaction
    self._connect(access_mode)
  File "/usr/local/lib/python3.9/site-packages/neo4j/__init__.py", line 381, in _connect
    self._connection.sync()
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 527, in sync
    detail_delta, summary_delta = self.fetch()
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 419, in fetch
    return self._fetch()
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 461, in _fetch
    response.on_failure(summary_metadata or {})
  File "/usr/local/lib/python3.9/site-packages/neobolt/direct.py", line 755, in on_failure
    raise CypherError.hydrate(**metadata)
neobolt.exceptions.ClientError: Unable to load NODE with id <redacted>.
```

Discussion:
achantavy: @marco-lancini @jg10 - I'm interested in seeing if this helps with your 20+ minute times for cleanup jobs.
marco-lancini: Thanks for tackling this, @achantavy! 🙏 

Patch:
diff --git a/cartography/graph/job.py b/cartography/graph/job.py
index f1684abc..73afb1c8 100644
--- a/cartography/graph/job.py
+++ b/cartography/graph/job.py
@@ -1,8 +1,15 @@
 import json
 import logging
+from pathlib import Path
+from typing import Dict
+from typing import List
 
+import neo4j
+
+from cartography.graph.statement import get_job_shortname
 from cartography.graph.statement import GraphStatement
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -24,18 +31,21 @@ class GraphJob:
     A job that will run against the cartography graph. A job is a sequence of statements which execute sequentially.
     """
 
-    def __init__(self, name, statements):
+    def __init__(self, name: str, statements: List[GraphStatement], short_name: str = None):
+        # E.g. "Okta intel module cleanup"
         self.name = name
-        self.statements = statements
+        self.statements: List[GraphStatement] = statements
+        # E.g. "okta_import_cleanup"
+        self.short_name = short_name
 
-    def merge_parameters(self, parameters):
+    def merge_parameters(self, parameters: Dict) -> None:
         """
         Merge parameters for all job statements.
         """
         for s in self.statements:
             s.merge_parameters(parameters)
 
-    def run(self, neo4j_session):
+    def run(self, neo4j_session: neo4j.Session):
         """
         Run the job. This will execute all statements sequentially.
         """
@@ -50,52 +60,58 @@ class GraphJob:
                     e,
                 )
                 raise
-        logger.debug("Finished job '%s'.", self.name)
+        log_msg = f"Finished job {self.short_name}" if self.short_name else f"Finished job {self.name}"
+        logger.info(log_msg)
 
-    def as_dict(self):
+    def as_dict(self) -> Dict:
         """
         Convert job to a dictionary.
         """
         return {
             "name": self.name,
             "statements": [s.as_dict() for s in self.statements],
+            "short_name": self.short_name,
         }
 
     @classmethod
-    def from_json(cls, blob):
+    def from_json(cls, blob: str, short_name: str = None):
         """
         Create a job from a JSON blob.
         """
-        data = json.loads(blob)
+        data: Dict = json.loads(blob)
         statements = _get_statements_from_json(data)
         name = data["name"]
-        return cls(name, statements)
+        return cls(name, statements, short_name)
 
     @classmethod
-    def from_json_file(cls, file_path):
+    def from_json_file(cls, file_path: Path):
         """
         Create a job from a JSON file.
         """
         with open(file_path) as j_file:
-            data = json.load(j_file)
-        statements = _get_statements_from_json(data)
-        name = data["name"]
-        return cls(name, statements)
+            data: Dict = json.load(j_file)
+
+        job_shortname: str = get_job_shortname(file_path)
+        statements: List[GraphStatement] = _get_statements_from_json(data, job_shortname)
+        name: str = data["name"]
+        return cls(name, statements, job_shortname)
 
     @classmethod
-    def run_from_json(cls, neo4j_session, blob, parameters=None):
+    def run_from_json(
+        cls, neo4j_session: neo4j.Session, blob: str, parameters: Dict = None, short_name: str = None,
+    ) -> None:
         """
         Run a job from a JSON blob. This will deserialize the job and execute all statements sequentially.
         """
         if not parameters:
             parameters = {}
 
-        job = cls.from_json(blob)
+        job: GraphJob = cls.from_json(blob, short_name)
         job.merge_parameters(parameters)
         job.run(neo4j_session)
 
     @classmethod
-    def run_from_json_file(cls, file_path, neo4j_session, parameters=None):
+    def run_from_json_file(cls, file_path: Path, neo4j_session: neo4j.Session, parameters: Dict = None) -> None:
         """
         Run a job from a JSON file. This will deserialize the job and execute all statements sequentially.
         """
@@ -103,17 +119,18 @@ class GraphJob:
             parameters = {}
 
         job = cls.from_json_file(file_path)
+
         job.merge_parameters(parameters)
         job.run(neo4j_session)
 
 
-def _get_statements_from_json(blob):
+def _get_statements_from_json(blob: Dict, short_job_name: str = None) -> List[GraphStatement]:
     """
     Deserialize all statements from the JSON blob.
     """
-    statements = []
-    for statement_data in blob["statements"]:
-        statement = GraphStatement.create_from_json(statement_data)
+    statements: List[GraphStatement] = []
+    for i, statement_data in enumerate(blob["statements"]):
+        statement: GraphStatement = GraphStatement.create_from_json(statement_data, short_job_name, i)
         statements.append(statement)
 
     return statements
diff --git a/cartography/graph/statement.py b/cartography/graph/statement.py
index fc5e5ba0..9c83e675 100644
--- a/cartography/graph/statement.py
+++ b/cartography/graph/statement.py
@@ -1,9 +1,17 @@
 import json
 import logging
+import os
+from pathlib import Path
+from typing import Dict
+from typing import Union
 
 import neo4j
 
+from cartography.stats import get_stats_client
+
+
 logger = logging.getLogger(__name__)
+stat_handler = get_stats_client(__name__)
 
 
 class GraphStatementJSONEncoder(json.JSONEncoder):
@@ -19,20 +27,32 @@ class GraphStatementJSONEncoder(json.JSONEncoder):
             return json.JSONEncoder.default(self, obj)
 
 
+# TODO move this cartography.util after we move util.run_*_job to cartography.graph.job.
+def get_job_shortname(file_path: Union[Path, str]) -> str:
+    # Return filename without path and extension
+    return os.path.splitext(file_path)[0]
+
+
 class GraphStatement:
     """
     A statement that will run against the cartography graph. Statements can query or update the graph.
     """
 
-    def __init__(self, query, parameters=None, iterative=False, iterationsize=0):
+    def __init__(
+        self, query: str, parameters: Dict = None, iterative: bool = False, iterationsize: int = 0,
+        parent_job_name: str = None, parent_job_sequence_num: int = None,
+    ):
         self.query = query
-        self.parameters = parameters
+        self.parameters: Dict = parameters
         if not parameters:
             self.parameters = {}
         self.iterative = iterative
         self.iterationsize = iterationsize
         self.parameters["LIMIT_SIZE"] = self.iterationsize
 
+        self.parent_job_name = parent_job_name if parent_job_name else None
+        self.parent_job_sequence_num = parent_job_sequence_num if parent_job_sequence_num else None
+
     def merge_parameters(self, parameters):
         """
         Merge given parameters with existing parameters.
@@ -41,17 +61,15 @@ class GraphStatement:
         tmp.update(parameters)
         self.parameters = tmp
 
-    def run(self, session) -> None:
+    def run(self, session: neo4j.Session) -> None:
         """
         Run the statement. This will execute the query against the graph.
         """
-        tx: neo4j.Transaction = session.begin_transaction()
         if self.iterative:
-            self._run_iterative(tx)
+            self._run_iterative(session)
         else:
-            data: neo4j.StatementResult = self._run(tx)
-            data.consume()
-        tx.commit()
+            session.write_transaction(self._run_noniterative).consume()
+        logger.info(f"Completed {self.parent_job_name} statement #{self.parent_job_sequence_num}")
 
     def as_dict(self):
         """
@@ -64,13 +82,32 @@ class GraphStatement:
             "iterationsize": self.iterationsize,
         }
 
-    def _run(self, tx: neo4j.Transaction) -> neo4j.StatementResult:
+    def _run_noniterative(self, tx: neo4j.Transaction) -> neo4j.StatementResult:
         """
         Non-iterative statement execution.
         """
-        return tx.run(self.query, self.parameters)
+        result: neo4j.StatementResult = tx.run(self.query, self.parameters)
+
+        # Handle stats
+        summary: neo4j.BoltStatementResultSummary = result.summary()
+        objects_changed: int = (
+            summary.counters.constraints_added +
+            summary.counters.constraints_removed +
+            summary.counters.indexes_added +
+            summary.counters.indexes_removed +
+            summary.counters.labels_added +
+            summary.counters.labels_removed +
+            summary.counters.nodes_created +
+            summary.counters.nodes_deleted +
+            summary.counters.properties_set +
+            summary.counters.relationships_created +
+            summary.counters.relationships_deleted
+        )
+        stat_handler.incr(f'{self.parent_job_name}-{self.parent_job_sequence_num}-objects_changed', objects_changed)
+
+        return result
 
-    def _run_iterative(self, tx: neo4j.Transaction) -> None:
+    def _run_iterative(self, session: neo4j.Session) -> None:
         """
         Iterative statement execution.
 
@@ -79,20 +116,17 @@ class GraphStatement:
         self.parameters["LIMIT_SIZE"] = self.iterationsize
 
         while True:
-            result: neo4j.StatementResult = self._run(tx)
-            record: neo4j.Record = result.single()
-
-            # TODO: use the BoltStatementResultSummary object to determine the number of items processed
-            total_completed = int(record['TotalCompleted'])
-            logger.debug("Processed %d items", total_completed)
+            result: neo4j.StatementResult = session.write_transaction(self._run_noniterative)
 
-            # Ensure network buffers are cleared
-            result.consume()
-            if total_completed == 0:
+            # Exit if we have finished processing all items
+            if not result.summary().counters.contains_updates:
+                # Ensure network buffers are cleared
+                result.consume()
                 break
+            result.consume()
 
     @classmethod
-    def create_from_json(cls, json_obj):
+    def create_from_json(cls, json_obj: Dict, short_job_name: str = None, job_sequence_num: int = None):
         """
         Create a statement from a JSON blob.
         """
@@ -101,14 +135,16 @@ class GraphStatement:
             json_obj.get("parameters", {}),
             json_obj.get("iterative", False),
             json_obj.get("iterationsize", 0),
+            short_job_name,
+            job_sequence_num,
         )
 
     @classmethod
-    def create_from_json_file(cls, file_path):
+    def create_from_json_file(cls, file_path: Path):
         """
         Create a statement from a JSON file.
         """
         with open(file_path) as json_file:
             data = json.load(json_file)
 
-        return cls.create_from_json(data)
+        return cls.create_from_json(data, get_job_shortname(file_path))
diff --git a/cartography/util.py b/cartography/util.py
index 4fdb3f61..cd8fdc7c 100644
--- a/cartography/util.py
+++ b/cartography/util.py
@@ -5,8 +5,10 @@ from typing import Dict
 from typing import Optional
 
 import botocore
+import neo4j
 
 from cartography.graph.job import GraphJob
+from cartography.graph.statement import get_job_shortname
 from cartography.stats import get_stats_client
 
 if sys.version_info >= (3, 7):
@@ -25,10 +27,14 @@ def run_analysis_job(filename, neo4j_session, common_job_parameters, package='ca
             filename,
         ),
         common_job_parameters,
+        get_job_shortname(filename),
     )
 
 
-def run_cleanup_job(filename, neo4j_session, common_job_parameters, package='cartography.data.jobs.cleanup'):
+def run_cleanup_job(
+    filename: str, neo4j_session: neo4j.Session, common_job_parameters: Dict,
+    package: str = 'cartography.data.jobs.cleanup',
+) -> None:
     GraphJob.run_from_json(
         neo4j_session,
         read_text(
@@ -36,6 +42,7 @@ def run_cleanup_job(filename, neo4j_session, common_job_parameters, package='car
             filename,
         ),
         common_job_parameters,
+        get_job_shortname(filename),
     )
 
 


Test patch:
diff --git a/tests/data/jobs/__init__.py b/tests/data/jobs/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/data/jobs/sample.py b/tests/data/jobs/sample.py
new file mode 100644
index 00000000..1d4e334b
--- /dev/null
+++ b/tests/data/jobs/sample.py
@@ -0,0 +1,19 @@
+SAMPLE_CLEANUP_JOB = """
+{
+  "statements": [
+    {
+      "query": "MATCH(:TypeA)-[r:REL]->(:TypeB) WHERE r.lastupdated <> {UPDATE_TAG} WITH r LIMIT {LIMIT_SIZE} DELETE r",
+      "iterative": true,
+      "iterationsize": 100
+    },{
+      "query": "MATCH (n:TypeA) WHERE n.lastupdated <> {UPDATE_TAG} WITH n LIMIT {LIMIT_SIZE} DETACH DELETE (n)",
+      "iterative": true,
+      "iterationsize": 100
+    },{
+      "query": "MATCH (n:TypeB) WHERE n.lastupdated <> {UPDATE_TAG} WITH n LIMIT {LIMIT_SIZE} DETACH DELETE (n)",
+      "iterative": true,
+      "iterationsize": 100
+    }],
+  "name": "cleanup stale resources"
+}
+"""
diff --git a/tests/integration/cartography/data/jobs/test_cleanup_jobs.py b/tests/integration/cartography/data/jobs/test_cleanup_jobs.py
new file mode 100644
index 00000000..d20fbfca
--- /dev/null
+++ b/tests/integration/cartography/data/jobs/test_cleanup_jobs.py
@@ -0,0 +1,133 @@
+from unittest import mock
+
+import cartography.util
+from cartography.util import run_cleanup_job
+
+
+UPDATE_TAG_T1 = 111111
+UPDATE_TAG_T2 = 222222
+UPDATE_TAG_T3 = 333333
+UPDATE_TAG_T4 = 444444
+
+
+SAMPLE_CLEANUP_JOB = """
+{
+  "statements": [
+    {
+      "query": "MATCH(:TypeA)-[r:REL]->(:TypeB) WHERE r.lastupdated <> {UPDATE_TAG} WITH r LIMIT {LIMIT_SIZE} DELETE r",
+      "iterative": true,
+      "iterationsize": 100
+    },{
+      "query": "MATCH (n:TypeA) WHERE n.lastupdated <> {UPDATE_TAG} WITH n LIMIT {LIMIT_SIZE} DETACH DELETE (n)",
+      "iterative": true,
+      "iterationsize": 100
+    },{
+      "query": "MATCH (n:TypeB) WHERE n.lastupdated <> {UPDATE_TAG} WITH n LIMIT {LIMIT_SIZE} DETACH DELETE (n)",
+      "iterative": true,
+      "iterationsize": 100
+    }],
+  "name": "cleanup stale resources"
+}
+"""
+
+SAMPLE_JOB_FILENAME = '/path/to/this/cleanupjob/mycleanupjob.json'
+
+
+@mock.patch.object(cartography.util, 'read_text', return_value=SAMPLE_CLEANUP_JOB)
+def test_run_cleanup_job_on_relationships(mock_read_text: mock.MagicMock, neo4j_session):
+    # Arrange: nodes id1 and id2 are connected to each other at time T2 via stale RELship r
+    neo4j_session.run(
+        """
+        MERGE (a:TypeA{id:"id1", lastupdated:{UPDATE_TAG_T2}})-[r:REL{lastupdated:{UPDATE_TAG_T1}}]->
+              (b:TypeB{id:"id2", lastupdated:{UPDATE_TAG_T2}})
+        """,
+        UPDATE_TAG_T1=UPDATE_TAG_T1,
+        UPDATE_TAG_T2=UPDATE_TAG_T2,
+    )
+
+    # Act: delete all nodes and rels where `lastupdated` != UPDATE_TAG_T2
+    job_parameters = {'UPDATE_TAG': UPDATE_TAG_T2}
+    run_cleanup_job(SAMPLE_JOB_FILENAME, neo4j_session, job_parameters)
+
+    # Assert 1: Node id1 is no longer attached to Node id2
+    nodes = neo4j_session.run(
+        """
+        MATCH (a:TypeA)
+        OPTIONAL MATCH (a)-[r:REL]->(b:TypeB)
+        RETURN a.id, r.lastupdated, b.id
+        """,
+    )
+    actual_nodes = {(n['a.id'], n['r.lastupdated'], n['b.id']) for n in nodes}
+    expected_nodes = {
+        ('id1', None, None),
+    }
+    assert actual_nodes == expected_nodes
+
+    # Assert 2: Node id2 still exists
+    nodes = neo4j_session.run(
+        """
+        MATCH (b:TypeB) RETURN b.id, b.lastupdated
+        """,
+    )
+    actual_nodes = {(n['b.id'], n['b.lastupdated']) for n in nodes}
+    expected_nodes = {
+        ('id2', UPDATE_TAG_T2),
+    }
+    assert actual_nodes == expected_nodes
+    mock_read_text.assert_called_once()
+
+
+@mock.patch.object(cartography.util, 'read_text', return_value=SAMPLE_CLEANUP_JOB)
+def test_run_cleanup_job_on_nodes(mock_read_text: mock.MagicMock, neo4j_session):
+    # Arrange: we are now at time T3, and node id1 exists but node id2 no longer exists
+    neo4j_session.run(
+        """
+        MATCH (a:TypeA{id:"id1"}) SET a.lastupdated={UPDATE_TAG_T3}
+        """,
+        UPDATE_TAG_T3=UPDATE_TAG_T3,
+    )
+
+    # Act: delete all nodes and rels where `lastupdated` != UPDATE_TAG_T3
+    job_parameters = {'UPDATE_TAG': UPDATE_TAG_T3}
+    run_cleanup_job(SAMPLE_JOB_FILENAME, neo4j_session, job_parameters)
+
+    # Assert: Node id1 is the only node that still exists
+    nodes = neo4j_session.run(
+        """
+        MATCH (n) RETURN n.id, n.lastupdated
+        """,
+    )
+    actual_nodes = {(n['n.id'], n['n.lastupdated']) for n in nodes}
+    expected_nodes = {
+        ('id1', UPDATE_TAG_T3),
+    }
+    assert actual_nodes == expected_nodes
+    mock_read_text.assert_called_once()
+
+
+@mock.patch.object(cartography.util, 'read_text', return_value=SAMPLE_CLEANUP_JOB)
+def test_run_cleanup_job_iterative_multiple_batches(mock_read_text: mock.MagicMock, neo4j_session):
+    # Arrange: load 300 nodes to the graph
+    for i in range(300):
+        neo4j_session.run(
+            """
+            MATCH (a:TypeA{id:{Id}}) SET a.lastupdated={UPDATE_TAG_T3}
+            """,
+            Id=i,
+            UPDATE_TAG_T3=UPDATE_TAG_T3,
+        )
+
+    # Act: delete all nodes and rels where `lastupdated` != UPDATE_TAG_T4.
+    job_parameters = {'UPDATE_TAG': UPDATE_TAG_T4}
+    run_cleanup_job(SAMPLE_JOB_FILENAME, neo4j_session, job_parameters)
+
+    # Assert: There are no nodes of label :TypeA in the graph, as the job iteratively removed them.
+    nodes = neo4j_session.run(
+        """
+        MATCH (n:TypeA) RETURN n.id
+        """,
+    )
+    actual_nodes = {n['n.id'] for n in nodes}
+    expected_nodes = set()
+    assert actual_nodes == expected_nodes
+    mock_read_text.assert_called_once()
diff --git a/tests/unit/cartography/data/test_graphjob.py b/tests/unit/cartography/data/test_graphjob.py
new file mode 100644
index 00000000..4618729f
--- /dev/null
+++ b/tests/unit/cartography/data/test_graphjob.py
@@ -0,0 +1,12 @@
+from cartography.graph.job import GraphJob
+from tests.data.jobs.sample import SAMPLE_CLEANUP_JOB
+
+
+def test_graphjob_from_json():
+    # Act
+    job: GraphJob = GraphJob.from_json(SAMPLE_CLEANUP_JOB)
+
+    # Assert that the job was created from json string contents correctly
+    assert job.name == "cleanup stale resources"
+    assert len(job.statements) == 3
+    assert job.short_name is None