Run pre-commit

amuser/am_api_ability.py

@@ -3,6 +3,7 @@
 This module contains the ``ArchivematicaAPIAbility`` class, which represents a
 user's ability to use Archivematica's APIs to interact with Archivematica.
 """
+
 import logging
 import os
 import time
@@ -11,7 +12,6 @@
 
 from . import base
 
-
 logger = logging.getLogger("amuser.api")
 
 

amuser/am_browser_ability.py

@@ -5,6 +5,7 @@
 Archivematica. This class provides an interface to Selenium for opening browser
 windows and interacting with Archivematica's GUIs.
 """
+
 import logging
 import time
 
@@ -24,7 +25,6 @@
 from . import base
 from . import constants as c
 
-
 logger = logging.getLogger("amuser.browser")
 
 
@@ -183,8 +183,8 @@ def initiate_reingest(self, aip_uuid, reingest_type="metadata-only"):
         }.get(reingest_type)
         if not type_selector:
             raise ArchivematicaBrowserAbilityError(
-                "Unable to initiate a reingest of type {} on AIP"
-                " {}".format(reingest_type, aip_uuid)
+                f"Unable to initiate a reingest of type {reingest_type} on AIP"
+                f" {aip_uuid}"
             )
         while True:
             type_input_el = self.driver.find_element(By.CSS_SELECTOR, type_selector)

amuser/am_browser_auth_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Authentication Ability"""
+
 import logging
 
 from selenium.common.exceptions import TimeoutException
@@ -8,7 +9,6 @@
 
 from . import selenium_ability
 
-
 logger = logging.getLogger("amuser.authentication")
 
 

amuser/am_browser_file_explorer_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Browser File Explorer Ability"""
+
 import logging
 import time
 
@@ -13,7 +14,6 @@
 from . import constants as c
 from . import selenium_ability
 
-
 logger = logging.getLogger("amuser.fileexplorer")
 
 
@@ -245,20 +245,18 @@ def get_xpath_matches_folder_text(folder_text):
     """
     return (
         "div[contains(@class, 'tree-label') and"
-        " descendant::span[starts-with(normalize-space(text()), '{0}') and"
+        f" descendant::span[starts-with(normalize-space(text()), '{folder_text}') and"
         " starts-with(normalize-space(substring-after("
         "normalize-space(text()),"
-        " '{0}')), '(')]]".format(folder_text)
+        f" '{folder_text}')), '(')]]"
     )
 
 
 def folder_label2icon_xpath(folder_label_xpath):
     """Given XPATH for TS folder label, return XPATH for its folder
     icon.
     """
-    return "{}/preceding-sibling::i[@class='tree-branch-head']".format(
-        folder_label_xpath
-    )
+    return f"{folder_label_xpath}/preceding-sibling::i[@class='tree-branch-head']"
 
 
 def folder_label2children_xpath(folder_label_xpath):

amuser/am_browser_ingest_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Ingest Tab Ability"""
+
 import logging
 import os
 import tempfile
@@ -105,7 +106,7 @@ def get_mets(self, transfer_name, sip_uuid=None, parse_xml=True):
             self.am_url, sip_uuid
         )
         self.navigate(aip_preview_url)
-        mets_path = "storeAIP/{0}-{1}/METS.{1}.xml".format(transfer_name, sip_uuid)
+        mets_path = f"storeAIP/{transfer_name}-{sip_uuid}/METS.{sip_uuid}.xml"
         handles_before = self.driver.window_handles
         self.navigate_to_aip_directory_and_click(mets_path)
         self.wait_for_new_window(handles_before)
@@ -116,10 +117,8 @@ def get_mets(self, transfer_name, sip_uuid=None, parse_xml=True):
         while self.driver.current_url.strip() == "about:blank":
             if attempts > self.max_check_mets_loaded_attempts:
                 msg = (
-                    "Exceeded maximum allowable attempts ({}) for checking"
-                    " if the METS file has loaded.".format(
-                        self.max_check_mets_loaded_attempts
-                    )
+                    f"Exceeded maximum allowable attempts ({self.max_check_mets_loaded_attempts}) for checking"
+                    " if the METS file has loaded."
                 )
                 logger.warning(msg)
                 raise ArchivematicaBrowserMETSAbilityError(msg)

amuser/am_browser_jobs_tasks_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Browser Jobs & Tasks Ability"""
+
 import logging
 import sys
 import time
@@ -10,7 +11,6 @@
 from . import selenium_ability
 from . import utils
 
-
 logger = logging.getLogger("amuser.jobstasks")
 
 
@@ -197,9 +197,7 @@ def get_job_uuid(
                         # The job is taking a long time to complete. Half the
                         # amount of checking to avoid stack-overflow.
                         logger.warning(
-                            "Recursion limit close to being reached: level: {} <= {}".format(
-                                level, sys.getrecursionlimit()
-                            )
+                            f"Recursion limit close to being reached: level: {level} <= {sys.getrecursionlimit()}"
                         )
                         time.sleep(self.quick_wait)
                     else:

amuser/am_browser_preservation_planning_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Browser Preservation Planning Ability"""
+
 import logging
 
 from selenium.common.exceptions import NoSuchElementException
@@ -7,7 +8,6 @@
 
 from . import selenium_ability
 
-
 logger = logging.getLogger("amuser.preservationplanning")
 
 
@@ -157,8 +157,8 @@ def save_policy_check_command(self, policy_command, description):
                 option.click()
                 break
         self.driver.find_element(By.ID, "id_description").send_keys(description)
-        js_script = 'document.getElementById("id_command").value =' " `{}`;".format(
-            policy_command
+        js_script = (
+            'document.getElementById("id_command").value =' f" `{policy_command}`;"
         )
         self.driver.execute_script(js_script)
         self.driver.find_element(By.ID, "id_script_type").send_keys("Python")
@@ -213,9 +213,7 @@ def search_for_fpr_rule(self, purpose, format_, command_description):
         description. Uses the FPR asynchronous search input.
         """
         terse_format = format_.split(":")[2].strip()
-        search_term = '"{}" "{}" "{}"'.format(
-            purpose, terse_format, command_description
-        )
+        search_term = f'"{purpose}" "{terse_format}" "{command_description}"'
         self.search_rules(search_term)
 
     def ensure_fpr_rule_enabled(self, purpose, format_, command_description):
@@ -234,17 +232,13 @@ def ensure_fpr_rule_enabled(self, purpose, format_, command_description):
         ]
         if not disabled_rules:
             logger.info(
-                'Tried to enable FPR rule with purpose "{}" that runs command "{}"'
-                ' against files with format "{}" but did not find it'.format(
-                    purpose, command_description, format_
-                )
+                f'Tried to enable FPR rule with purpose "{purpose}" that runs command "{command_description}"'
+                f' against files with format "{format_}" but did not find it'
             )
             return
         assert len(disabled_rules) == 1, (
-            'Expected to enable one FPR rule with purpose "{}" that runs command "{}"'
-            ' against files with format "{}" but found {} disabled rules'.format(
-                purpose, command_description, format_, len(disabled_rules)
-            )
+            f'Expected to enable one FPR rule with purpose "{purpose}" that runs command "{command_description}"'
+            f' against files with format "{format_}" but found {len(disabled_rules)} disabled rules'
         )
         rule = disabled_rules[0]
         rule.find_element(By.CSS_SELECTOR, "td:nth-child(6) a:nth-child(3)").click()

amuser/am_browser_ss_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Browser Storage Service Ability"""
+
 import logging
 import pprint
 import time
@@ -11,7 +12,6 @@
 from . import selenium_ability
 from . import utils
 
-
 logger = logging.getLogger("amuser.ss")
 
 
@@ -43,8 +43,8 @@ def approve_aip_delete_request(self, aip_uuid):
                 matching_rows.append(row_el)
         if len(matching_rows) != 1:
             raise ArchivematicaBrowserStorageServiceAbilityError(
-                "More than one delete request row {} matches AIP"
-                " {}".format(len(matching_rows), aip_uuid)
+                f"More than one delete request row {len(matching_rows)} matches AIP"
+                f" {aip_uuid}"
             )
         matching_rows[0].find_element(By.TAG_NAME, "textarea").send_keys("Cuz wanna")
         matching_rows[0].find_element(By.CSS_SELECTOR, 'input[name="approve"]').click()
@@ -318,9 +318,9 @@ def add_replicator_to_default_aip_stor_loc(self, replicator_location_uuid):
                 break
         if not found_replicator:
             raise ArchivematicaBrowserStorageServiceAbilityError(
-                "Unable to find replicator location {} as a possible replicator"
+                f"Unable to find replicator location {replicator_location_uuid} as a possible replicator"
                 " for the default AIP Storage"
-                " location".format(replicator_location_uuid)
+                " location"
             )
         self.driver.find_element(By.CSS_SELECTOR, "input[type=submit]").click()
 

amuser/am_browser_transfer_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Transfer Tab Ability"""
+
 import logging
 import os
 import time
@@ -11,7 +12,6 @@
 from . import constants as c
 from . import selenium_ability
 
-
 logger = logging.getLogger("amuser.transfer")
 
 
@@ -249,9 +249,7 @@ def approve_transfer(
         re-click the micro-service <div> to make the hidden <select> visible
         again.
         """
-        approve_transfer_option_selector = "option[value='{}']".format(
-            approve_option_uuid
-        )
+        approve_transfer_option_selector = f"option[value='{approve_option_uuid}']"
         while True:
             try:
                 approve_transfer_option = transfer_div_elem.find_element(

amuser/am_browser_transfer_ingest_ability.py

@@ -1,4 +1,5 @@
 """Archivematica Transfer & Ingest Tabs Ability"""
+
 import logging
 import sys
 import time
@@ -15,7 +16,6 @@
 from . import selenium_ability
 from . import utils
 
-
 logger = logging.getLogger("amuser.transferingest")
 
 
@@ -116,10 +116,8 @@ def assert_no_option(
             assert f'Unable to select choice "{choice_text}"' == str(exc)
         else:
             raise AssertionError(
-                'We were able to select choice "{}" at decision point "{}" even'
-                " though we expected this not to be possible.".format(
-                    choice_text, decision_point
-                )
+                f'We were able to select choice "{choice_text}" at decision point "{decision_point}" even'
+                " though we expected this not to be possible."
             )
 
     @selenium_ability.recurse_on_stale
@@ -142,9 +140,7 @@ def wait_for_microservice_visibility(
             # The job is taking a long time to complete. Half the
             # amount of checking to avoid stack-overflow.
             logger.warning(
-                "Recursion limit close to being reached: level: {} <= {}".format(
-                    level, sys.getrecursionlimit()
-                )
+                f"Recursion limit close to being reached: level: {level} <= {sys.getrecursionlimit()}"
             )
             time.sleep(self.micro_wait)
         else:
@@ -186,9 +182,9 @@ def wait_for_transfer_micro_service_group(self, group_name, transfer_uuid):
         while True:
             if attempts > max_attempts:
                 msg = (
-                    "Exceeded maxumim allowable attempts ({}) for checking"
-                    " whether micro-service group {} of transfer {} is"
-                    " visible.".format(max_attempts, group_name, transfer_uuid)
+                    f"Exceeded maxumim allowable attempts ({max_attempts}) for checking"
+                    f" whether micro-service group {group_name} of transfer {transfer_uuid} is"
+                    " visible."
                 )
                 logger.warning(msg)
                 raise ArchivematicaBrowserTransferIngestAbilityError(msg)

amuser/am_docker_ability.py

@@ -4,14 +4,14 @@
 ability of an Archivematica user to use Docker to interact configure and deploy
 Archivematica.
 """
+
 import logging
 import os
 import shlex
 import subprocess
 
 from . import base
 
-
 logger = logging.getLogger("amuser.docker")
 
 
@@ -28,9 +28,7 @@ def recreate_archivematica(self, capture_output=False):
         """Recreate the docker-compose deploy of Archivematica by calling
         docker-compose's ``up`` subcommand.
         """
-        dc_recreate_cmd = "docker-compose -f {} up -d".format(
-            self.docker_compose_file_path
-        )
+        dc_recreate_cmd = f"docker-compose -f {self.docker_compose_file_path} up -d"
         capture_output = {True: "true"}.get(capture_output, "false")
         env = dict(os.environ, AM_CAPTURE_CLIENT_SCRIPT_OUTPUT=capture_output)
         subprocess.check_output(shlex.split(dc_recreate_cmd), env=env)
@@ -54,12 +52,10 @@ def get_tasks_from_sip_uuid(
             " TIMEDIFF(t.endTime,t.startTime) as duration"
             " FROM Tasks t"
             " INNER JOIN Files f ON f.fileUUID=t.fileUUID"
-            " WHERE f.sipUUID='{}'"
-            " ORDER by endTime-startTime, exec;".format(sip_uuid)
-        )
-        cmd_str = 'docker-compose exec mysql mysql -u {} -p{} MCP -e "{}"'.format(
-            mysql_user, mysql_password, sql_query
+            f" WHERE f.sipUUID='{sip_uuid}'"
+            " ORDER by endTime-startTime, exec;"
         )
+        cmd_str = f'docker-compose exec mysql mysql -u {mysql_user} -p{mysql_password} MCP -e "{sql_query}"'
         res = subprocess.check_output(
             shlex.split(cmd_str), cwd=self.docker_compose_path
         ).decode("utf8")
@@ -114,9 +110,7 @@ def cp_server_file_to_local(self, server_file_path):
         local_path = os.path.join(self.tmp_path, filename)
         subprocess.check_output(
             shlex.split(
-                "docker cp {}:{} {}".format(
-                    docker_container_id, server_file_path, local_path
-                )
+                f"docker cp {docker_container_id}:{server_file_path} {local_path}"
             ),
             cwd=self.docker_compose_path,
         ).decode("utf8").strip()
@@ -136,9 +130,7 @@ def cp_server_dir_to_local(self, server_dir_path):
         local_path = os.path.join(self.tmp_path, dirname)
         subprocess.check_output(
             shlex.split(
-                "docker cp {}:{} {}".format(
-                    docker_container_id, server_dir_path, local_path
-                )
+                f"docker cp {docker_container_id}:{server_dir_path} {local_path}"
             ),
             cwd=self.docker_compose_path,
         ).decode("utf8").strip()

amuser/am_localfs_ability.py

@@ -3,11 +3,11 @@
 This module contains the ``ArchivematicaLocalFSAbility`` class, which encodes
 the ability to transfer files when they're locally available.
 """
+
 import logging
 
 from . import base
 
-
 logger = logging.getLogger("amuser.localfs")