tests/i: forward port test changes from cylc#3500

tests/integration/test_data_store_mgr.py

@@ -58,10 +58,10 @@ def test_generate_graph_elements(harness):
 def test_get_data_elements(harness):
     schd, data = harness
     flow_msg = schd.data_store_mgr.get_data_elements(TASK_PROXIES)
-    assert len(flow_msg.deltas) == len(data[TASK_PROXIES])
+    assert len(flow_msg.added) == len(data[TASK_PROXIES])
 
     flow_msg = schd.data_store_mgr.get_data_elements(WORKFLOW)
-    assert flow_msg.last_updated == data[WORKFLOW].last_updated
+    assert flow_msg.added.last_updated == data[WORKFLOW].last_updated
 
     none_msg = schd.data_store_mgr.get_data_elements('fraggle')
     assert len(none_msg.ListFields()) == 0
@@ -103,7 +103,7 @@ def test_prune_points(harness):
 
 def test_update_data_structure(harness):
     """Test update_data_structure. This method will generate and
-    apply deltas/updates given."""
+    apply adeltas/updates given."""
     schd, data = harness
     assert len(collect_states(data, TASK_PROXIES)) == 1
     update_tasks = schd.pool.get_all_tasks()

tests/integration/test_job_pool.py

@@ -93,36 +93,35 @@ def int_id(_):
 
 def test_insert_job(myflow):
     """Test method that adds a new job to the pool."""
-    assert len(myflow.job_pool.updates) == 0
+    assert len(myflow.job_pool.added) == 0
     myflow.job_pool.insert_job(job_config(myflow))
-    assert len(myflow.job_pool.updates) == 1
-    assert ext_id(myflow) in myflow.job_pool.updates
+    assert len(myflow.job_pool.added) == 1
+    assert ext_id(myflow) in myflow.job_pool.added
 
 
 def test_insert_db_job(myflow, job_db_row):
     """Test method that adds a new job to the pool."""
-    assert len(myflow.job_pool.updates) == 0
+    assert len(myflow.job_pool.added) == 0
     myflow.job_pool.insert_db_job(0, job_db_row)
-    assert len(myflow.job_pool.updates) == 1
-    assert ext_id(myflow) in myflow.job_pool.updates
+    assert len(myflow.job_pool.added) == 1
+    assert ext_id(myflow) in myflow.job_pool.added
 
 
 def test_add_job_msg(myflow):
     """Test method adding messages to job element."""
     myflow.job_pool.insert_job(job_config(myflow))
-    job = myflow.job_pool.updates[ext_id(myflow)]
-    old_stamp = copy(job.stamp)
-    assert len(job.messages) == 0
+    job_added = myflow.job_pool.added[ext_id(myflow)]
+    assert len(job_added.messages) == 0
     myflow.job_pool.add_job_msg(int_id(myflow), 'The Atomic Age')
-    assert old_stamp != job.stamp
-    assert len(job.messages) == 1
+    job_updated = myflow.job_pool.updated[ext_id(myflow)]
+    assert len(job_updated.messages) == 1
 
 
 def test_reload_deltas(myflow):
     """Test method reinstatiating job pool on reload"""
     assert myflow.job_pool.updates_pending is False
     myflow.job_pool.insert_job(job_config(myflow))
-    myflow.job_pool.pool = {e.id: e for e in myflow.job_pool.updates.values()}
+    myflow.job_pool.pool = {e.id: e for e in myflow.job_pool.added.values()}
     myflow.job_pool.reload_deltas()
     assert myflow.job_pool.updates_pending
 
@@ -145,41 +144,43 @@ def test_remove_task_jobs(myflow):
     assert len(pruned) == 0
     myflow.job_pool.remove_task_jobs('NotTaskID')
     assert len(pruned) == 0
-    task_id = myflow.job_pool.updates[ext_id(myflow)].task_proxy
+    task_id = myflow.job_pool.added[ext_id(myflow)].task_proxy
     myflow.job_pool.remove_task_jobs(task_id)
     assert len(pruned) == 1
 
 
 def test_set_job_attr(myflow):
     """Test method setting job attribute value."""
     myflow.job_pool.insert_job(job_config(myflow))
-    job = myflow.job_pool.updates[ext_id(myflow)]
-    old_exit_script = copy(job.exit_script)
-    assert job.exit_script == old_exit_script
+    job_added = myflow.job_pool.added[ext_id(myflow)]
     myflow.job_pool.set_job_attr(int_id(myflow), 'exit_script', 'rm -v *')
-    assert old_exit_script != job.exit_script
+    assert job_added.exit_script != (
+        myflow.job_pool.updated[ext_id(myflow)].exit_script
+    )
 
 
 def test_set_job_state(myflow):
     """Test method setting the job state."""
     myflow.job_pool.insert_job(job_config(myflow))
-    job = myflow.job_pool.updates[ext_id(myflow)]
-    old_state = copy(job.state)
-    myflow.job_pool.set_job_state(int_id(myflow), 'waiting')
-    assert job.state == old_state
+    job_added = myflow.job_pool.added[ext_id(myflow)]
+    myflow.job_pool.set_job_state(int_id(myflow), JOB_STATUSES_ALL[1])
+    job_updated = myflow.job_pool.updated[ext_id(myflow)]
+    state_two = copy(job_updated.state)
+    assert job_added.state != state_two
     myflow.job_pool.set_job_state(int_id(myflow), JOB_STATUSES_ALL[-1])
-    assert old_state != job.state
+    assert state_two != job_updated.state
 
 
 def test_set_job_time(myflow):
     """Test method setting event time."""
     event_time = get_current_time_string()
     myflow.job_pool.insert_job(job_config(myflow))
-    job = myflow.job_pool.updates[ext_id(myflow)]
-    old_time = copy(job.submitted_time)
-    assert job.submitted_time == old_time
+    job_added = myflow.job_pool.added[ext_id(myflow)]
     myflow.job_pool.set_job_time(int_id(myflow), 'submitted', event_time)
-    assert old_time != job.submitted_time
+    job_updated = myflow.job_pool.updated[ext_id(myflow)]
+    with pytest.raises(ValueError):
+        job_updated.HasField('jumped_time')
+    assert job_added.submitted_time != job_updated.submitted_time
 
 
 def test_parse_job_item(myflow):

tests/integration/test_publisher.py

@@ -42,4 +42,4 @@ async def test_publisher(flow, scheduler, run, one_conf, port_range):
             btopic, msg = await subscriber.socket.recv_multipart()
 
         _, delta = process_delta_msg(btopic, msg, None)
-        assert schd.id == delta.id
+        assert schd.id == delta.added.id

tests/integration/test_resolvers.py

@@ -69,7 +69,7 @@ async def flow(mod_flow, mod_scheduler, mod_run):
     ret.name = ret.schd.suite
     ret.id = list(ret.schd.data_store_mgr.data.keys())[0]
     ret.resolvers = Resolvers(
-        ret.schd.data_store_mgr.data,
+        ret.schd.data_store_mgr,
         schd=ret.schd
     )
     ret.data = ret.schd.data_store_mgr.data[ret.id]

tests/integration/test_server.py

@@ -65,7 +65,7 @@ def test_pb_data_elements(myflow):
             element_type
         )
     )
-    assert data.id == myflow.id
+    assert data.added.id == myflow.id
 
 
 def test_pb_entire_workflow(myflow):