fix tavern test codes

Author: ryoya-hayase

test/tavern/create_item_type_template.py

@@ -163,9 +163,10 @@ def create_template(json):
     text_type = ['text', 'textarea', 'select', 'radios']
     result = {}
     for k, v in json.items():
-        if v.get('title') == 'Identifier Registration'\
-            or v.get('title').startswith('Persistent Identifier')\
-            or v.get('title') == 'File Information':
+        if v.get('title') \
+            and (v.get('title') == 'Identifier Registration'\
+                 or v.get('title').startswith('Persistent Identifier')\
+                 or v.get('title') == 'File Information'):
             continue
         if v.get('format'):
             if v.get('format') == 'datetime':
@@ -210,7 +211,10 @@ def create_random_template(json):
     result = {}
     has_uri = False
     for k, v in json.items():
-        if v.get('title') == 'Identifier Registration' or v.get('title').startswith('Persistent Identifier') or v.get('title') == 'File Information':
+        if v.get('title') \
+            and (v.get('title') == 'Identifier Registration' \
+                 or v.get('title').startswith('Persistent Identifier') \
+                 or v.get('title') == 'File Information'):
             continue
         if v.get('format'):
             if v.get('format') == 'datetime':

test/tavern/generator/json/generate_json.py

@@ -359,7 +359,11 @@ def generate_semi_pair(keys, pairs, properties, number):
         dict: The semi-pair of values.
     """
     result = {}
+    if not pairs:
+        raise ValueError("The 'pairs' list is empty.")
     for i in range(len(keys)):
+        if i >= len(pairs[number % len(pairs)]):
+            raise IndexError(f"Index {i} is out of range for 'pairs' element with length {len(pairs[number % len(pairs)])}.")
         result[keys[i]] = pairs[number % len(pairs)][i]
         if not result[keys[i]] and mode > 1:
             result[keys[i]] = pairs[number % len(pairs) + 1][i]

test/tavern/generator/json/generate_prepare_index.py

@@ -78,7 +78,7 @@ def create_query(self):
 
 
 def create_index(
-    depth, num, current_level = 1, parent_id = 0, before_id = 0
+    depth, num, current_level = 1, parent_id = 0, before_id = 0, adjust_position = -1
 ):
     """Recursively create a list of Index instances.
 
@@ -88,6 +88,7 @@ def create_index(
         current_level(int, optional): Current level in the hierarchy. Defaults to 1.
         parent_id(int, optional): Parent index ID. Defaults to 0.
         before_id(int, optional): ID of the last created index. Defaults to 0.
+        adjust_position(int, optional): Adjustment for the position value. Defaults to -1.
 
     Returns:
         list[Index]: List of created Index instances.
@@ -101,7 +102,7 @@ def create_index(
         params = {
             "id": id,
             "parent": parent_id,
-            "position": i + 2,
+            "position": i + adjust_position,
             "index_name": f"Index_L{current_level}_N{i}",
             "index_name_english": f"Index_L{current_level}_N{i}_EN",
             "index_link_name": f"Index_L{current_level}_N{i}_Link",
@@ -122,7 +123,7 @@ def main(hierarchy_depth, num_indices_per_level):
         hierarchy_depth(int): Depth of the index hierarchy.
         num_indices_per_level(int): Number of indices per level.
     """
-    indices = create_index(hierarchy_depth, num_indices_per_level, before_id=100)
+    indices = create_index(hierarchy_depth, num_indices_per_level, before_id=100, adjust_position=2)
 
     with open("prepare_data/prepare_index.sql", "w", encoding="utf-8") as f:
         query = [index.create_query() for index in indices]

test/tavern/generator/json/generate_request_body.py

@@ -540,10 +540,18 @@ def create_error_files(required_types, item_type_id, output_dir):
         }
         save_body(body_noid, output_dir, f"mapping_noid_{mapping_type}.json")
 
+    duplicate_output_dir = output_dir.replace("error", "duplicate")
+    mapping = {}
+    if os.path.exists(duplicate_output_dir):
+        duplicate_files = [f for f in os.listdir(duplicate_output_dir) if f.endswith(".json")]
+        if duplicate_files:
+            with open(os.path.join(duplicate_output_dir, duplicate_files[0]), "r", encoding="utf-8") as f:
+                duplicate_data = json.load(f)
+            mapping = duplicate_data.get("mapping", {})
     # missing mapping_type key
     body_noid = {
         "item_type_id": item_type_id,
-        "mapping": {}
+        "mapping": mapping
     }
     save_body(body_noid, output_dir, f"mapping_no_mapping_type.json")
 
@@ -632,6 +640,8 @@ def set_db_keys(k):
         else:
             return (4, k)
     db_item_keys = sorted(list(mapping_raw.keys()), key=set_db_keys)
+    if meta_ids and len(meta_ids) == 2:
+        db_item_keys = [id for id in meta_ids if id in db_item_keys] + [k for k in db_item_keys if k not in meta_ids]
 
     for role in test_roles:
         print(f"{role}のテストデータの生成を開始します。")

test/tavern/helper/common/item_create_helper.py

@@ -31,6 +31,7 @@ def create_item(
     is_doi=False,
     prepare_edit=False,
     id=2000001,
+    replace_title=True,
 ):
     """Create items based on the provided creation information.
 
@@ -44,6 +45,7 @@ def create_item(
         is_doi(bool, optional): Boolean indicating whether to assign a DOI to the item.
         prepare_edit(bool, optional): Boolean indicating whether to prepare the item for editing after creation.
         id(int, optional): Optional ID to assign to the created item.
+        replace_title(bool, optional): Boolean indicating whether to replace the title of the item.
 
     Raises:
         Exception: If any step in the item creation process fails.
@@ -63,18 +65,21 @@ def create_item(
     with open(create_info["data_file"], "r") as f:
         data = json.loads(f.read())
 
-    url = data[create_info["file_key"]][0]["url"]["url"]
-    replaced_url = url.replace("{id}", str(id))
-    data[create_info["file_key"]][0]["url"]["url"] = replaced_url
-
-    title_key = create_info["title_key"].split(".")
-    title = data[title_key[0]][int(title_key[1])][title_key[2]]
-    replaced_title = title + f"_{id}"
-    data[title_key[0]][int(title_key[1])][title_key[2]] = replaced_title
-
-    identifier = data[create_info["identifier_key"]][0]["subitem_identifier_uri"]
-    replaced_identifier = identifier + f"_{id}"
-    data[create_info["identifier_key"]][0]["subitem_identifier_uri"] = replaced_identifier
+    if create_info.get("file_key") and data.get(create_info["file_key"]) is not None:
+        url = data[create_info["file_key"]][0]["url"]["url"]
+        replaced_url = url.replace("{id}", str(id))
+        data[create_info["file_key"]][0]["url"]["url"] = replaced_url
+
+    if replace_title:
+        title_key = create_info["title_key"].split(".")
+        title = data[title_key[0]][int(title_key[1])][title_key[2]]
+        replaced_title = title + f"_{id}"
+        data[title_key[0]][int(title_key[1])][title_key[2]] = replaced_title
+
+    if create_info.get("identifier_key") and data.get(create_info["identifier_key"]) is not None:
+        identifier = data[create_info["identifier_key"]][0]["subitem_identifier_uri"]
+        replaced_identifier = identifier + f"_{id}"
+        data[create_info["identifier_key"]][0]["subitem_identifier_uri"] = replaced_identifier
 
     for _ in range(creation_count):
         # create activity

test/tavern/helper/common/item_import_helper.py

@@ -7,14 +7,15 @@
 from box import Box
 
 
-def import_item(response, host, file_path, target_index):
+def import_item(response, host, file_path, target_index, csrf_token):
     """Import item via admin API.
 
     Args:
         response(Response): Response object from previous request to get headers.
         host(str): Host URL.
         file_path(str): Path to the file to be imported.
         target_index(int): Target index ID to replace in the import data.
+        csrf_token(str): CSRF token for authentication.
 
     Returns:
         Box: A Box object containing import task details.
@@ -23,7 +24,7 @@ def import_item(response, host, file_path, target_index):
     header = {
         "Referer": host,
         "Cookie": request_headers.get("Cookie", ""),
-        "X-CSRFToken": request_headers.get("X-CSRFToken", ""),
+        "X-CSRFToken": csrf_token,
     }
 
     session = requests.Session()
@@ -38,12 +39,10 @@ def import_item(response, host, file_path, target_index):
     # Get check status
     while True:
         check_result = get_check_status(host, session, check_import_task_id)
-
-        if check_result["end_date"]:
+        if check_result.get("end_date"):
             break
         sleep(1)
 
-    print("Import Check Result:", check_result)
     # Import
     import_response = start_import(
         host, session, check_result["data_path"], check_result["list_record"]

test/tavern/helper/common/request_helper.py

@@ -14,6 +14,8 @@ def request_create_validate_param(data, file_metadata=None):
             item_id (str): item id
             data (dict): register data
     """
+    if isinstance(data, str):
+        data = json.loads(data)
     item_id = data['$schema'].split('/')[-1]
     params = {
         'item_id': item_id,
@@ -22,10 +24,8 @@ def request_create_validate_param(data, file_metadata=None):
     if file_metadata is not None:
         metadata = json.loads(file_metadata)
         keys = list(metadata.keys())
-        for key, value in metadata[keys[0]][0].items():
-            if key not in params['data'][keys[0]][0]:
-                params['data'][keys[0]][0][key] = value
-    print(params['data'][keys[0]])
+        if keys[0] not in params['data'].keys():
+            params['data'][keys[0]] = metadata[keys[0]]
     return params
 
 
@@ -98,9 +98,8 @@ def request_create_save_param(data, url=None, file_upload_info=None, file_metada
     if file_metadata is not None:
         metadata = json.loads(file_metadata)
         keys = list(metadata.keys())
-        for key, value in metadata[keys[0]][0].items():
-            if key not in params['metainfo'][keys[0]][0]:
-                params['metainfo'][keys[0]][0][key] = value
+        if keys[0] not in params['metainfo'].keys():
+            params['metainfo'][keys[0]] = metadata[keys[0]]
     return params
 
 def request_create_deposits_items_param(data=None):
@@ -159,7 +158,7 @@ def request_create_deposits_redirect_param(data, title_key, file_metadata=None):
         '$schema': data['$schema'],
         'lang': 'ja',
         'pubdate': data['pubdate'],
-        'shared_user_id': data['shared_user_id'],
+        'shared_user_ids': data['shared_user_ids'],
         'title': title
     }
 

test/tavern/helper/common/response_helper.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timezone
 import json
+import os
 import random
 import string
 from urllib.parse import urlparse
@@ -247,7 +248,10 @@ def response_save_url_described_in_email(_, target, save_key):
     Raises:
         ValueError: If no URL is found in the email file.
     """
-    with open('mail/' + target) as f:
+    target_folder = f'mail/{target}/new'
+    files = [f for f in os.listdir(target_folder) if os.path.isfile(os.path.join(target_folder, f))]
+    files.sort(reverse=True)
+    with open(os.path.join(target_folder, files[0]), 'r') as f:
         lines = f.readlines()
     url = ''
     for line in lines:
@@ -362,6 +366,13 @@ def response_save_register_data_with_change(_, file_name, key_dict_file, change_
     elif change_type == 6:
         # change creator name
         register_data[creator_key][0]['creatorNames'][0]['creatorName'] = 'たかはし さぶろう'
+        register_data[creator_key][0]['nameIdentifiers'] = [
+            {
+                "nameIdentifier": "1",
+                "nameIdentifierScheme": "WEKO"
+            }
+        ]
+        del register_data[creator_key][1]
 
     return Box({'register_data': json.dumps(register_data)})
 

…/tavern/helper/verify_database_helper.py …/helper/common/verify_database_helper.pytest/tavern/helper/verify_database_helper.py renamed to test/tavern/helper/common/verify_database_helper.py test/tavern/helper/verify_database_helper.py renamed to test/tavern/helper/common/verify_database_helper.py

@@ -1,29 +1,10 @@
 from datetime import datetime, timedelta
 import json
 import os
-import psycopg2
 
-from helper.config import DATABASE, REPLACEMENT_DECISION_STRING
+from helper.config import REPLACEMENT_DECISION_STRING
 
 
-def connect_db():
-    """Connect to database
-    
-    Args:
-        None
-        
-    Returns:
-        psycopg2.extensions.connection: connection to database
-    """
-    conn = psycopg2.connect(
-        dbname=DATABASE['dbname'],
-        user=DATABASE['user'],
-        password=DATABASE['password'],
-        host=DATABASE['host'],
-        port=DATABASE['port'],
-    )
-    return conn
-
 def compare_db_data(cursor, folder_path, replace_params = {}, type_conversion_params = {}, datetime_columns=[]):
     """Compare data in database with data in excel file
 

test/tavern/helper/index/data_transformation_helper.py

@@ -63,7 +63,7 @@ def get_parents(cursor, index_id):
                 "position": index_data["position"],
                 "public_date": datetime.strftime(
                     index_data["public_date"], "%Y-%m-%dT%H:%M:%S")
-                    if index_data["public_date"] else "",
+                    if index_data["public_date"] else None,
                 "public_state": index_data["public_state"],
                 "recursive_coverpage_check": index_data["recursive_coverpage_check"],
                 "settings": {