Develop w2025 16 1.3

modules/invenio-oaiserver/tests/test_response.py

@@ -198,7 +198,7 @@ def test_getrecord(app, db, item_type, mocker):
                 "item_custom_sort": {"2": 1}
             }
         private_index = Index(**private_index_metadata)
-        mapping = Mapping.create(
+        mapping = Mapping.create_or_update(
             item_type_id=item_type.id,
             mapping={}
         )
@@ -449,7 +449,7 @@ def test_getrecord_future_item(app,records,item_type,mock_execute,db,mocker):
             "item_custom_sort": {"2": 1}
         }
         index = Index(**index_metadata)
-        mapping = Mapping.create(
+        mapping = Mapping.create_or_update(
             item_type_id=item_type.id,
             mapping={}
         )
@@ -538,7 +538,7 @@ def test_listidentifiers(es_app,records,item_type,mock_execute,db,mocker):
             "item_custom_sort":{"2":1}
         }
         index = Index(**index_metadata)
-        mapping = Mapping.create(
+        mapping = Mapping.create_or_update(
             item_type_id=item_type.id,
             mapping={}
         )
@@ -883,7 +883,7 @@ def test_listrecords(es_app,records,item_type,mock_execute,db,mocker):
             "item_custom_sort":{"2":1}
         }
         index = Index(**index_metadata)
-        mapping = Mapping.create(
+        mapping = Mapping.create_or_update(
             item_type_id=item_type.id,
             mapping={}
         )
@@ -2036,7 +2036,7 @@ def test_issue34851_listrecords(es_app, records, item_type, mock_execute,db,mock
             "item_custom_sort":{"2":1}
         }
         index = Index(**index_metadata)
-        mapping = Mapping.create(
+        mapping = Mapping.create_or_update(
             item_type_id=item_type.id,
             mapping={}
         )
@@ -2159,7 +2159,7 @@ def test_issue34851_listidentifiers(es_app, records, item_type, mock_execute,db,
             "item_custom_sort":{"2":1}
         }
         index = Index(**index_metadata)
-        mapping = Mapping.create(
+        mapping = Mapping.create_or_update(
             item_type_id=item_type.id,
             mapping={}
         )

modules/weko-gridlayout/tests/conftest.py

@@ -493,7 +493,7 @@ def item_type(db):
     with db.session.begin_nested():
         db.session.add(item_type)
         db.session.add(item_type_property)
-    mappin = Mapping.create(
+    mappin = Mapping.create_or_update(
         item_type.id,
         mapping = json_data("data/item_type/item_type_mapping.json")
     )

modules/weko-itemtypes-ui/tests/test_admin.py

@@ -1166,7 +1166,7 @@ def test_mapping_register(self,client,admin_view,users,item_type,mocker):
             assert  Mapping.get_record(1) == {"key":"test_mapping"}
 
             # raise Exception
-            with patch("weko_itemtypes_ui.admin.Mapping.create",side_effect=BaseException):
+            with patch("weko_itemtypes_ui.admin.Mapping.create_or_update",side_effect=BaseException):
                 res = client.post(url,json=data)
                 assert json.loads(res.data)["msg"] == "Unexpected error occurred."
 #     def schema_list(self, SchemaName=None):

modules/weko-itemtypes-ui/weko_itemtypes_ui/admin.py

@@ -274,11 +274,11 @@ def register(self, item_type_id=0):
             ]
 
             if not upgrade_version:
-                Mapping.create(item_type_id=record.model.id,
+                Mapping.create_or_update(item_type_id=record.model.id,
                                mapping=table_row_map.get('mapping'))
             # Just update Mapping when create new record
             elif record.model.id != item_type_id:
-                Mapping.create(item_type_id=record.model.id,
+                Mapping.create_or_update(item_type_id=record.model.id,
                                mapping=table_row_map.get('mapping'))
                 workflow = WorkFlow()
                 workflow_list = workflow.get_workflow_by_itemtype_id(
@@ -649,7 +649,7 @@ def item_type_import(self):
                 'WEKO_ITEMTYPES_UI_UPGRADE_VERSION_ENABLED'
             ]
             item_type_id=record.model.id
-            Mapping.create(
+            Mapping.create_or_update(
                 item_type_id=item_type_id,
                 mapping=import_data['ItemTypeMapping'].get('mapping')
             )
@@ -950,7 +950,7 @@ def mapping_register(self):
             return jsonify(duplicate=True, err_items=lst_duplicate,
                            msg=_('Duplicate mapping as below:'))
         try:
-            Mapping.create(item_type_id=data.get('item_type_id'),
+            Mapping.create_or_update(item_type_id=data.get('item_type_id'),
                            mapping=data_mapping)
             db.session.commit()
             UserActivityLogger.info(

modules/weko-records/tests/conftest.py

@@ -639,7 +639,7 @@ def item_type_mapping(app, db):
             }
         }
     }
-    return Mapping.create(1, _mapping)
+    return Mapping.create_or_update(1, _mapping)
 
 @pytest.fixture()
 def item_type2(app, db):
@@ -706,7 +706,7 @@ def item_type_mapping2(app, db):
             }
         }
     }
-    return Mapping.create(2, _mapping)
+    return Mapping.create_or_update(2, _mapping)
 
 @pytest.fixture()
 def item_type3(app, db):
@@ -987,7 +987,7 @@ def item_type_mapping_with_form(app, db, item_type_with_form):
             }
         }
     }
-    return Mapping.create(item_type_with_form.id, _mapping)
+    return Mapping.create_or_update(item_type_with_form.id, _mapping)
 
 
 @pytest.fixture()

modules/weko-records/tests/test_api.py

@@ -20,7 +20,6 @@
 
 """Module tests."""
 
-from re import T
 import uuid
 import pytest
 from datetime import datetime, timedelta
@@ -39,7 +38,9 @@
 from weko_records.api import FeedbackMailList, RequestMailList, ItemApplication, FilesMetadata, ItemLink, \
     ItemsMetadata, ItemTypeEditHistory, ItemTypeNames, ItemTypeProps, \
     ItemTypes, Mapping, JsonldMapping, SiteLicense, RecordBase, WekoRecord
-from weko_records.models import ItemReference, ItemTypeJsonldMapping, ItemTypeName, ItemTypeProperty, ItemType
+from weko_records.models import ItemReference, ItemTypeMapping, ItemTypeJsonldMapping, ItemTypeName, ItemTypeProperty, ItemType
+from jsonschema.validators import Draft4Validator
+from datetime import datetime, timedelta
 
 # class RecordBase(dict):
 # .tox/c1/bin/pytest --cov=weko_records tests/test_api.py::test_recordbase -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-records/.tox/c1/tmp
@@ -1072,22 +1073,58 @@ def test_item_type_edit_history(app, db, user):
 #     def create(cls, item_type_id=None, mapping=None):
 # .tox/c1/bin/pytest --cov=weko_records tests/test_api.py::test_mapping_create -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-records/.tox/c1/tmp
 def test_mapping_create(app, db):
-    mapping = Mapping.create()
-    assert mapping.id==1
-    assert mapping.model.item_type_id==None
-    assert mapping.model.mapping=={}
+    with patch("weko_records.api.before_record_insert") as mock_before_record_insert, \
+            patch("weko_records.api.after_record_insert") as mock_after_record_insert:
+        mapping = Mapping.create_or_update()
+        db.session.commit()
+        obj = ItemTypeMapping.query.first()
+        assert obj.id == 1
+        assert obj.item_type_id == None
+        assert obj.mapping == {}
+        mock_before_record_insert.send.assert_called_once()
+        mock_after_record_insert.send.assert_called_once()
+
+    with patch("weko_records.api.before_record_insert") as mock_before_record_insert, \
+            patch("weko_records.api.after_record_insert") as mock_after_record_insert:
+        mapping = Mapping.create_or_update(1, {'mapping': 'create'})
+        db.session.commit()
+        obj = ItemTypeMapping.query.filter_by(item_type_id=1).first()
+        assert obj.id == 2
+        assert obj.item_type_id == 1
+        assert obj.mapping == {'mapping': 'create'}
+        assert obj.version_id == 1
+        mock_before_record_insert.send.assert_called_once()
+        mock_after_record_insert.send.assert_called_once()
+
+    with patch("weko_records.api.before_record_update") as mock_before_record_update, \
+            patch("weko_records.api.after_record_update") as mock_after_record_update:
+        Mapping.create_or_update(1, {'mapping': 'update'})
+        db.session.commit()
+        obj = ItemTypeMapping.query.filter_by(item_type_id=1).one()
+        assert obj.id == 2
+        assert obj.item_type_id == 1
+        assert obj.mapping == {'mapping': 'update'}
+        assert obj.version_id == 2
+        mock_before_record_update.send.assert_called_once()
+        mock_after_record_update.send.assert_called_once()
 
-    mapping = Mapping.create(1, {'mapping': 'test'})
-    assert mapping.id==2
-    assert mapping.model.item_type_id==1
-    assert mapping.model.mapping=={'mapping': 'test'}
+        versions = obj.versions.all()
+        assert len(versions) == 2
+        assert versions[0].id == obj.id
+        assert versions[0].item_type_id == 1
+        assert versions[0].mapping == {'mapping': 'create'}
+        assert versions[0].version_id == 1
+        assert versions[1].id == obj.id
+        assert versions[1].item_type_id == 1
+        assert versions[1].mapping == {'mapping': 'update'}
+        assert versions[1].version_id == 2
 
 # class Mapping(RecordBase):
 #     def get_record(cls, item_type_id, with_deleted=False):
 # .tox/c1/bin/pytest --cov=weko_records tests/test_api.py::test_mapping_get_record -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-records/.tox/c1/tmp
 def test_mapping_get_record(app, db):
-    Mapping.create(1, {'mapping': 'test'})
-    Mapping.create(2)
+    Mapping.create_or_update(1, {'mapping': 'test'})
+    Mapping.create_or_update(2)
 
     mapping = Mapping.get_record(0)
     assert mapping==None
@@ -1132,8 +1169,8 @@ def test_patch_Mapping(app):
 #     def commit(self, **kwargs):
 # .tox/c1/bin/pytest --cov=weko_records tests/test_api.py::test_mapping_commit -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-records/.tox/c1/tmp
 def test_mapping_commit(app, db):
-    mapping1 = Mapping.create(1)
-    mapping2 = Mapping.create(2)
+    mapping1 = Mapping.create_or_update(1)
+    mapping2 = Mapping.create_or_update(2)
 
     mapping1.model = None
     with pytest.raises(Exception) as e:
@@ -1149,30 +1186,30 @@ def test_mapping_commit(app, db):
 #     def delete(self, force=False):
 # .tox/c1/bin/pytest --cov=weko_records tests/test_api.py::test_mapping_delete -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-records/.tox/c1/tmp
 def test_mapping_delete(app, db):
-    mapping1 = Mapping.create(1)
-    mapping2 = Mapping.create(2)
-    mapping3 = Mapping.create(3)
+    mapping1 = Mapping.create_or_update(1)
+    mapping2 = Mapping.create_or_update(2)
+    mapping3 = Mapping.create_or_update(3)
 
     mapping1.model = None
     with pytest.raises(Exception) as e:
-        Mapping.delete(mapping1)
+        mapping1.delete()
     assert e.type==MissingModelError
 
-    mapping2 = Mapping.delete(mapping2, False)
+    mapping2 = mapping2.delete(force=False)
     assert mapping2.id==2
     assert mapping2.model.item_type_id==2
     assert mapping2.model.mapping=={}
 
     # need to fix
-    mapping3 = Mapping.delete(mapping3, True)
-    assert mapping2=={}
+    mapping3 = mapping3.delete(force=True)
+    assert mapping3=={}
 
 # class Mapping(RecordBase):
 #     def revert(self, revision_id):
 # .tox/c1/bin/pytest --cov=weko_records tests/test_api.py::test_mapping_revert -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-records/.tox/c1/tmp
 def test_mapping_revert(app, db):
-    mapping1 = Mapping.create(1)
-    mapping2 = Mapping.create(2)
+    mapping1 = Mapping.create_or_update(1)
+    mapping2 = Mapping.create_or_update(2)
 
     mapping1.model = None
     with pytest.raises(Exception) as e:
@@ -1208,8 +1245,8 @@ def dummy_func():
 #     def get_mapping_by_item_type_ids(cls, item_type_ids: list) -> list:
 # .tox/c1/bin/pytest --cov=weko_records tests/test_api.py::test_mapping_get_mapping_by_item_type_ids -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-records/.tox/c1/tmp
 def test_mapping_get_mapping_by_item_type_ids(app, db):
-    Mapping.create(1)
-    Mapping.create(2)
+    Mapping.create_or_update(1)
+    Mapping.create_or_update(2)
 
     mappings = Mapping.get_mapping_by_item_type_ids([0])
     assert len(mappings)==0

modules/weko-records/tests/test_serializers_utils.py

@@ -143,7 +143,7 @@ def fetcher(obj_uuid, data):
         form=json_data(form),
         tag=1
     )
-    _item_type_mapping = Mapping.create(
+    _item_type_mapping = Mapping.create_or_update(
         item_type_id=_item_type.id,
         mapping=json_data(mapping)
     )

modules/weko-records/weko_records/alembic/e0dd9fb514cf_add_constraints_to_mapping.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 National Institute of Informatics.
+#
+# WEKO-Records is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Add constraints to mapping"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'e0dd9fb514cf'
+down_revision = 'e3b07ec6e628'
+branch_labels = ()
+depends_on = None
+
+
+def upgrade():
+    """Upgrade database."""
+    with op.batch_alter_table('item_type_mapping') as batch_op:
+        batch_op.create_unique_constraint(
+            'uq_item_type_mapping_item_type_id', ['item_type_id']
+        )
+        batch_op.create_foreign_key(
+            'fk_item_type_mapping_item_type_id_item_type',
+            'item_type', ['item_type_id'], ['id'],
+            ondelete="CASCADE"
+        )
+
+    # Drop indexes from DDL.
+    op.execute("DROP INDEX IF EXISTS idx_created_item_type_mapping")
+    op.execute("DROP INDEX IF EXISTS idx_item_type_id_item_type_mapping")
+
+
+
+def downgrade():
+    """Downgrade database."""
+    with op.batch_alter_table('item_type_mapping') as batch_op:
+        batch_op.drop_constraint(
+            'fk_item_type_mapping_item_type_id_item_type', type_='foreignkey'
+        )
+        batch_op.drop_constraint(
+            'uq_item_type_mapping_item_type_id', type_='unique'
+        )
+
+    # Recreate indexes from DDL.
+    op.execute("""
+        CREATE INDEX IF NOT EXISTS idx_created_item_type_mapping
+        ON item_type_mapping USING BTREE (created)
+    """)
+    op.execute("""
+        CREATE INDEX IF NOT EXISTS idx_item_type_id_item_type_mapping
+        ON item_type_mapping USING BTREE (item_type_id)
+    """)