Verify incremental coverage rejects a concurrent parent update

tests/test_postgres_ingestion.py

@@ -135,6 +135,34 @@ class PostgreSQLIngestionTests(unittest.TestCase):
             self.assertEqual(tuple(coverage.values),('partial',2,2))
             self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
 
+    def test_incremental_postgres_does_not_overwrite_a_concurrently_changed_parent(self):
+        listing='https://example.org/uai/2001'
+        rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
+              for key in ('concurrent1','concurrent2','concurrent3')]
+        first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json')
+        value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value))
+        self.ingest(first)
+        path=self.file('racing',[rows[2]]);target=path.with_suffix('.manifest.json')
+        value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental',
+            coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=4)
+        target.write_text(json.dumps(value))
+        original=Store.record
+        def change_parent(store,*args,**kwargs):
+            result=original(store,*args,**kwargs)
+            # A separate committed SQL writer need not honor the native
+            # advisory lock. Its update must survive our rejected file.
+            with connect(self.service) as other:
+                other.execute('SET search_path TO '+quote(self.schema)+',pg_catalog')
+                other.execute('UPDATE edition_sources SET expected_count=99 WHERE list_url=%s',(listing,))
+            return result
+        with patch.object(Store,'record',change_parent):
+            with self.assertRaisesRegex(ValueError,'parent changed during import'):self.ingest(path)
+        with closing(ReadConnection(self.service,self.schema)) as c:
+            coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
+            self.assertEqual(tuple(coverage.values),('partial',99,2))
+            self.assertEqual(c.execute('SELECT count(*) FROM source_records WHERE external_id=?',('concurrent3',)).fetchone()[0],0)
+            self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
+
     def test_native_writes_keep_nul_historical_ids_and_later_acceptance(self):
         store = self.store()
         try: