misc: keep track of stored object maximum identifier (#45919)

tests/test_admin_pages.py

@@ -470,7 +470,7 @@ def test_workflows_export_import_create_role(pub):
     assert Workflow.get(3).name == 'Copy of foo (2)'
     assert Role.count() == 1
     assert Role.select()[0].name == 'PLOP'
-    assert Workflow.get(2).possible_status[0].items[0].by == [Role.select()[0].id]
+    assert Workflow.get(3).possible_status[0].items[0].by == [Role.select()[0].id]
 
     # don't create role if they are managed by the identity provider
     Role.wipe()

tests/test_datasource.py

@@ -620,6 +620,24 @@ def test_data_source_slug_name():
     assert data_source.slug == 'foo_bar'
 
 
+def test_data_source_new_id():
+    NamedDataSource.wipe()
+    data_source = NamedDataSource(name='foo bar')
+    data_source.store()
+    assert data_source.id == '1'
+    data_source = NamedDataSource(name='foo bar2')
+    data_source.store()
+    assert data_source.id == '2'
+    data_source.remove_self()
+    data_source = NamedDataSource(name='foo bar3')
+    data_source.store()
+    assert data_source.id == '3'
+    NamedDataSource.wipe()
+    data_source = NamedDataSource(name='foo bar4')
+    data_source.store()
+    assert data_source.id == '1'
+
+
 def test_optional_item_field_with_data_source():
     NamedDataSource.wipe()
     data_source = NamedDataSource(name='foobar')

tests/test_workflow_import.py

@@ -514,9 +514,6 @@ def test_complex_dispatch_action(pub):
     role4.name = 'Test Role 2'
     role4.store()
 
-    role1.remove_self()
-    role2.remove_self()
-
     xml_export_orig = export_to_indented_xml(wf, include_id=True)
     wf2 = Workflow.import_from_xml_tree(xml_export_orig)
     assert wf2.possible_status[0].items[0].variable == dispatch.variable

wcs/formdef.py

@@ -330,15 +330,9 @@ class FormDef(StorableObject):
 
     @classmethod
     def get_new_id(cls, create=False):
-        keys = cls.keys()
-        if not keys:
-            id = 1
-        else:
-            id = max([lax_int(x) for x in keys]) + 1
-            if id == 0:
-                id = len(keys)+1
+        id = super().get_new_id(create=False)
         if get_publisher().is_using_postgresql():
-            id = cls.get_sql_new_id(id_start=id)
+            id = cls.get_sql_new_id(id_start=int(id))
         if create:
             objects_dir = cls.get_objects_dir()
             object_filename = os.path.join(objects_dir, fix_key(id))

wcs/qommon/storage.py

@@ -378,21 +378,27 @@ class StorableObject(object):
 
     @classmethod
     def get_new_id(cls, create=False):
+        objects_dir = cls.get_objects_dir()
+        try:
+            max_id = int(open(os.path.join(objects_dir, '.max_id')).read())
+        except (IOError, OSError, ValueError):
+            max_id = 0
         keys = cls.keys()
         if not keys:
-            id = 1
+            id = max_id + 1
         else:
-            id = max([lax_int(x) for x in keys]) + 1
+            id = max([lax_int(x) for x in keys] + [max_id]) + 1
             if id == 0:
                 id = len(keys)+1
         if create:
-            objects_dir = cls.get_objects_dir()
             object_filename = os.path.join(objects_dir, fix_key(id))
             try:
                 fd = os.open(object_filename, os.O_CREAT | os.O_EXCL)
             except OSError:
                 return cls.get_new_id(create=True)
             os.close(fd)
+            with open(os.path.join(objects_dir, '.max_id'), 'w') as fd:
+                fd.write(str(id))
         return str(id)
 
     @classmethod