diff --git a/lfric_macros/apply_macros.py b/lfric_macros/apply_macros.py index 4047ce5f..d2af98ec 100755 --- a/lfric_macros/apply_macros.py +++ b/lfric_macros/apply_macros.py @@ -21,13 +21,14 @@ import yaml import networkx as nx from collections import defaultdict +from pathlib import Path BLACK_COMMAND = "black --line-length=80" CLASS_NAME_REGEX = r"vn\d+(_t\d+\w*)?" TAG_REGEX = r"\s*=\s*[\"']\s*(\S+)\s*[\"']" -def run_command(command, shell=False): +def run_command(command: str, shell: bool = False) -> subprocess.CompletedProcess: """ Run a subprocess command and return the result object Inputs: @@ -47,7 +48,7 @@ def run_command(command, shell=False): ) -def check_environment(): +def check_environment() -> None: """ Check that required dependencies are loaded in the current environment Cylc >= 8.0 @@ -72,14 +73,14 @@ def check_environment(): raise Exception("'isort' must be available to run this script") -def get_root_path(wc_path): +def get_root_path(wc_path: Path) -> Path: """ Given a path to a git clone, ensure the path and clone are both valid and return the path to the clone root directory Inputs: - wc_path, command line argument to a clone Outputs: - - str, path to the top level of the apps clone + - path to the top level of the apps clone """ command = f"git -C {wc_path} rev-parse --show-toplevel" @@ -94,14 +95,14 @@ def get_root_path(wc_path): # If no error, then search through output for the working copy root path # return the found path if result.stdout: - return result.stdout.strip() + return Path(result.stdout.strip()) raise Exception( "Couldn't extract the Git Clone Root path from the output of the " f"command '{command}'" ) -def apply_styling(filepath): +def apply_styling(filepath: Path) -> None: """ Run black on a given file Inputs: @@ -123,7 +124,7 @@ def apply_styling(filepath): ) -def read_versions_file(meta_dir): +def read_versions_file(meta_dir: Path) -> list[str]: """ Read in a versions.py and parse out blank lines Inputs: @@ -132,7 +133,7 @@ def read_versions_file(meta_dir): - a list of lines in the versions.py file with blank lines removed """ - version_file = os.path.join(meta_dir, "versions.py") + version_file = meta_dir / "versions.py" # Read in versions file and then remove all blank lines with open(version_file) as f: @@ -145,7 +146,7 @@ def read_versions_file(meta_dir): return file_parsed -def split_macros(parsed_versions): +def split_macros(parsed_versions: list[str]) -> list[str]: """ Read through a versions.py file and split macros into individual strings Inputs: @@ -177,7 +178,7 @@ def split_macros(parsed_versions): return macros -def deduplicate_list(lst): +def deduplicate_list(lst: list) -> list: """ Remove duplicate items from a list, keeping the first Need to preserve order so not using a set @@ -194,7 +195,7 @@ def deduplicate_list(lst): return deduplicated -def match_python_import(line): +def match_python_import(line: str) -> bool: """ Return true if string has form 'import *' or 'from * import *' Inputs: @@ -205,12 +206,12 @@ def match_python_import(line): return False -def read_python_imports(path): +def read_python_imports(path: Path) -> set[tuple]: """ Given a path to a python file, return a set containing info of all module imports in the file Inputs: - - path, path to a python file + - path to a python file Returns: - set containing data of python imports in given file """ @@ -233,7 +234,7 @@ def read_python_imports(path): return imports -def banner_print(message): +def banner_print(message: str) -> None: """Print a simple banner message""" print(f"\n{(len(message) + 4) * '*'}\n* {message} *\n{(len(message) + 4) * '*'}\n") @@ -243,112 +244,113 @@ class ApplyMacros: Object to hold data + methods to apply upgrade macros in lfric_apps """ - def __init__(self, tag, cname, version, apps, core, jules, testing=False): - self.tag = tag + def __init__( + self, + tag: str, + cname: str | None, + version: str, + apps: Path, + core: Path, + testing: bool = False, + ) -> None: + self.tag: str = tag if cname: - self.class_name = cname + self.class_name: str = cname else: # The default class name is the After Tag with the '.' # removed from the version - self.class_name = tag.replace(".", "") - self.temp_dirs = {} + self.class_name: str = tag.replace(".", "") + self.temp_dirs: dict = {} if testing: # Don't search for a git repo if testing - self.root_path = apps + self.root_path: Path = apps else: - self.root_path = get_root_path(apps) - self.core_source = self.get_dependency_paths(core, "lfric_core") - # The Jules source is temporarily ignored as Jules Shared metadata has a - # copy in LFRic, rather than using the Jules version. The LFRic build - # system needs modifying to enable this - # self.jules_source = self.get_dependency_paths(jules, "jules") + self.root_path: Path = get_root_path(apps) + self.core_source: Path = self.get_dependency_paths(core) self.set_rose_meta_path() if version is None: - self.version = re.search(r".*vn(\d+\.\d+)(_.*)?", tag).group(1) + self.version: str = re.search(r".*vn(\d+\.\d+)(_.*)?", tag).group(1) else: - self.version = version - self.author = None - self.ticket_number = None + self.version: str = version + self.author: str | None = None + self.ticket_number: str | None = None # All parsed macros per metadata section - self.parsed_macros = defaultdict(list) + self.parsed_macros: dict[list] = defaultdict(list) # Parsed macro with desired after tag, per metadata section - self.target_macros = {} - self.meta_dirs = set() - self.sections_with_macro = [] - self.python_imports = set() - self.upgraded_core = False + self.target_macros: dict = {} + self.meta_dirs: set = set() + self.sections_with_macro: list = [] + self.python_imports: set = set() + self.upgraded_core: bool = False - def set_rose_meta_path(self): + def set_rose_meta_path(self) -> None: """ - Set up the ROSE_META_PATH environment variable in order to use the Jules - and Core metadata. We also add the clone root path as this should - allow the script to be run from anywhere. - When Jules Shared from Jules is enabled, self.jules_source will need - adding here + Set up the ROSE_META_PATH environment variable in order to use the Core + metadata. We also add the clone root path as this should allow the script to be + run from anywhere. Edit 02/2026 - remove backwards compatibility support for pre central-metadata """ - rose_meta_path = ( - f"{os.path.join(self.root_path, 'rose-meta')}:" - f"{os.path.join(self.core_source, 'rose-meta')}" + rose_meta_path: str = ( + f"{self.root_path / 'rose-meta'}:{self.core_source / 'rose-meta'}" ) os.environ["ROSE_META_PATH"] = rose_meta_path - def parse_application_section(self, meta_dir): + def parse_application_section(self, meta_dir: Path) -> Path: """ Given a path to a metadata directory, parse out the application/science - section. Try to remove the apps, core and jules root paths. Then try to - remove trailing /HEAD or /versions.py + section. Try to remove the apps and core root paths. Then try to remove trailing + /HEAD or /versions.py Inputs: - meta_dir, path to a metadata dir Returns: - - path to the metadata directory with the root path removed + - path to the metadata directory with the root path and suffixes removed """ - meta_dir = meta_dir.removeprefix(self.root_path) - meta_dir = meta_dir.removeprefix(self.core_source) - # Reinstate when using Jules Shared from Jules - # meta_dir = meta_dir.removeprefix(self.jules_source) + + meta_dir = str(meta_dir) + meta_dir = meta_dir.removeprefix(str(self.root_path)) + meta_dir = meta_dir.removeprefix(str(self.core_source)) meta_dir = meta_dir.removeprefix("/") meta_dir = meta_dir.removesuffix("/HEAD") meta_dir = meta_dir.removesuffix("/versions.py") - return meta_dir + return Path(meta_dir) ############################################################################ # Get Working Copy Functions ############################################################################ - def get_dependency_paths(self, source, repo): + def get_dependency_paths(self, source: str | None) -> Path: """ - Parse the core or jules command line arguments to get the path to a git clone. + Parse the core command line arguments to get the path to a git clone. If the source isn't defined, first populate the source by reading the dependencies.yaml file. If the source is a remote GitHub source clone it to a temporary location Inputs: - - source, str, The command line argument for the source. If not set - this will be None - - repo, str, Either "lfric_core" or "jules" depending on which - source is being found + - source, The command line argument for the source. If not set this will be + None Outputs: - - str, The path to the source working copy to use + - The path to the source working copy to use """ + repo = "lfric_core" + # If source is None then read the dependencies.yaml file for the source if source is None: - source, ref = self.read_dependencies(repo) + source, ref = self.read_dependencies() + if ":" in str(source): + source_path = Path(source.split(":")[1]).expanduser() + else: + source_path = Path(source).expanduser() # If the source exists as a path then return as is - if os.path.exists(os.path.expanduser(source)): - return os.path.expanduser(source) - if ":" in source: - source_path = os.path.expanduser(source.split(":")[1]) - if os.path.exists(source_path): - return source_path + if source_path.exists(): + return source_path # Check that the source looks like a GitHub URL, raise an error if not if "git@github.com:" not in source and "https://github.com/" not in source: raise Exception( - f"The {repo} source: {source}, was not found as a working copy " + f"The {repo} source: {source}, was not found as a local clone " "and does not look like an GitHub URL. Please check the source." "If not set on the command then the dependencies.yaml file is " "being used." @@ -358,32 +360,29 @@ def get_dependency_paths(self, source, repo): source = self.git_clone_temp(source, ref, repo) return source - def read_dependencies(self, repo): + def read_dependencies(self, repo: str = "lfric_core") -> tuple[str, str]: """ Read through the dependencies.yaml file for the source of the repo defined by repo. Uses self.root_path to locate the dependencies.yaml file. - Inputs: - - repo, str, Either "lfric_core" or "jules" depending on which - source is being found. The function will work with - other repos, but not intended to within this script. Outputs: - - str, The source as defined by the dependencies.yaml file + - The source as defined by the dependencies.yaml file + - The ref as defined by the dependencies.yaml file """ - dependencies_path = os.path.join(self.root_path, "dependencies.yaml") + dependencies_path = self.root_path / "dependencies.yaml" with open(dependencies_path, "r") as f: dependencies = yaml.safe_load(f) return dependencies[repo]["source"], dependencies[repo]["ref"] - def git_clone_temp(self, source, ref, repo): + def git_clone_temp(self, source: str, ref: str, repo: str) -> Path: """ Given a github URL, extract a temporary clone Inputs: - - source, str, A github URL - - ref, str, A git ref to checkout, None will result in default branch - - repo, str, the name of the source being found + - source, A github URL + - ref, A git ref to checkout, None will result in default branch + - repo, the name of the source being found Outputs: - - str, The path to the temporary working copy + - The path to the temporary working copy """ print(f"Extracting {source} to a temporary directory") @@ -397,29 +396,29 @@ def git_clone_temp(self, source, ref, repo): f"Failed to clone from {source} into directory {tempdir} " f"with error message:\n\n{result.stderr}" ) - return tempdir + return Path(tempdir) ############################################################################ # Preprocess Macros Functions ############################################################################ - def find_meta_dirs(self, path): + def find_meta_dirs(self, path: Path) -> set[Path]: """ Searching from a working copy root path, return a list of paths to all - the rose-meta directories using os.walk(). Search by looking + the rose-meta directories using path.walk(). Search by looking for versions.py files Outputs: - - str, stdout of find command looking for versions.py files + - set containing paths of metadata directories """ meta_dirs = set() - for dirpath, dirnames, filenames in os.walk(path, followlinks=True): + for dirpath, dirnames, filenames in path.walk(follow_symlinks=True): dirnames[:] = [d for d in dirnames if d not in [".svn", ".git"]] if "versions.py" in filenames: meta_dirs.add(dirpath) return meta_dirs - def parse_macro(self, macro, meta_dir): + def parse_macro(self, macro: str, meta_dir: Path) -> dict: """ Given a macro string save the macro info in a dictionary Inputs: @@ -431,7 +430,7 @@ def parse_macro(self, macro, meta_dir): commands """ - version_file = os.path.join(meta_dir, "versions.py") + version_file = meta_dir / "versions.py" ticket_details = re.search(r"Upgrade .* (#\d+) by (\S+.*)", macro) try: @@ -485,7 +484,7 @@ def parse_macro(self, macro, meta_dir): "class_name": class_name, } - def remove_macro(self, contents, meta_dir): + def remove_macro(self, contents: list[str], meta_dir: Path) -> None: """ Rewrite the contents of a versions.py file without the newly added macro. Run black on the new file. @@ -494,8 +493,8 @@ def remove_macro(self, contents, meta_dir): - meta_dir, the path to the versions.py file being rewritten """ - filepath = os.path.join(meta_dir, "versions.py") - temppath = os.path.join(meta_dir, ".versions.py") + filepath = meta_dir / "versions.py" + temppath = meta_dir / ".versions.py" with open(temppath, "w") as f: in_new_macro = False @@ -510,23 +509,24 @@ def remove_macro(self, contents, meta_dir): apply_styling(temppath) - if not os.path.getsize(temppath) > 0: + if not temppath.stat().st_size > 0: raise Exception( f"The file modified at {filepath} has zero size, indicating " "something has gone wrong" ) - os.rename(temppath, filepath) + temppath.rename(filepath) - def find_last_macro(self, macros, meta_dir): + def find_last_macro(self, macros: list[str], meta_dir: Path) -> str: """ Given a list of macros, determine the after tag of the final macro in the chain. Start from assuming the first before tag is the Version Number. Inputs: - macros, a list of macro strings + - meta_dir, the path to the metadata directory Returns: - - str, the after tag of the final macro in the chain + - the after tag of the final macro in the chain """ after_tag = f"vn{self.version}" @@ -550,20 +550,19 @@ def find_last_macro(self, macros, meta_dir): ) return after_tag - def find_macro(self, meta_dir, macros): + def find_macro(self, meta_dir: Path, macros: list[str]) -> str: """ Read through a list of macros, trying to find the macro with a class name that matches the class name supplied (either from the tag or cname option). If this is present then return the macro. Inputs: - - meta_dir, str, The path to the rose metadata directory containing - these macros + - meta_dir, The path to the rose metadata directory containing these macros - macros, a list of macro strings Returns: - String containing the macro. Empty if the macro isn't found """ - version_file = os.path.join(meta_dir, "versions.py") + version_file = meta_dir / "versions.py" # Find the macro we're interested in for macro in macros: @@ -579,9 +578,9 @@ def find_macro(self, meta_dir, macros): return macro return "" - def get_full_import_path(self, imp): + def get_full_import_path(self, imp: str) -> Path: """ - Search through the Core, Jules and Apps working copies to get the full + Search through the Core and Apps working copies to get the full path to a metadata directory Inputs: - imp, the import statement without the full path @@ -590,26 +589,23 @@ def get_full_import_path(self, imp): not found """ - # TODO: Reinstate Jules checks when using Jules Metadata from Jules + core_imp = self.core_source / "rose-meta" / imp + apps_imp = self.root_path / "rose-meta" / imp - core_imp = os.path.join(self.core_source, "rose-meta", imp) - apps_imp = os.path.join(self.root_path, "rose-meta", imp) - - if os.path.exists(core_imp): + if core_imp.exists(): return core_imp - if os.path.exists(apps_imp): + if apps_imp.exists(): return apps_imp - if os.path.exists(os.path.dirname(core_imp)): + if core_imp.parent.exists(): return core_imp - if os.path.exists(os.path.dirname(apps_imp)): + if apps_imp.parent.exists(): return apps_imp raise Exception( - f"Couldn't find the import '{imp}' in any of the Apps, Core or " - "Jules sources." + f"Couldn't find the import '{imp}' in any of the Apps or Core Sources" ) - def read_meta_imports(self, meta_dir, flag="import"): + def read_meta_imports(self, meta_dir: Path, flag: str = "import") -> list[str]: """ Read a rose-meta.conf and record which other metadata files are imported by this metadata. @@ -618,10 +614,12 @@ def read_meta_imports(self, meta_dir, flag="import"): - flag, either 'import' or 'meta'. Searches for lines in the config file starting flag=. If 'meta', then will return the import statement on that line + Returns: + - list of metadata import strings """ if flag == "import": - meta_file = os.path.join(meta_dir, "HEAD", "rose-meta.conf") + meta_file = meta_dir / "HEAD" / "rose-meta.conf" else: meta_file = meta_dir @@ -647,7 +645,7 @@ def read_meta_imports(self, meta_dir, flag="import"): break return imports - def write_python_imports(self, meta_dir): + def write_python_imports(self, meta_dir: Path) -> None: """ Write out all required python module imports at the top of a versions.py file. New imports are written at the top of the current import @@ -656,8 +654,8 @@ def write_python_imports(self, meta_dir): - meta_dir, path to the metadata directory with a versions.py file """ - filepath = os.path.join(meta_dir, "versions.py") - temppath = os.path.join(meta_dir, ".versions.py") + filepath = meta_dir / "versions.py" + temppath = meta_dir / ".versions.py" # Work out where we need to insert the new imports # For simplicity, do this at the beginning of the existing imports @@ -687,9 +685,9 @@ def write_python_imports(self, meta_dir): for line in versions_file: f.write(line.strip("\n") + "\n") - os.rename(temppath, filepath) + temppath.rename(filepath) - def determine_import_order(self, app): + def determine_import_order(self, app: str) -> list[str]: """ Work out what order metadata is imported. This recursively works through import statements recorded in self.target_macros["imports"]. Produces a @@ -717,7 +715,7 @@ def determine_import_order(self, app): return deduplicate_list(import_list) - def combine_macros(self, import_order): + def combine_macros(self, import_order: list[str]) -> str: """ Combine macro commands, adding commands in the order determined by import_order. @@ -725,7 +723,7 @@ def combine_macros(self, import_order): - import_order, the metadata import order to match the order of marcro commands. Returns: - - string, combined macro commands + - combined macro commands in a single formatted string """ full_command = "" @@ -757,7 +755,7 @@ def combine_macros(self, import_order): ] return full_command - def write_new_macro(self, meta_dir, full_command, macro): + def write_new_macro(self, meta_dir: Path, full_command: str, macro: str) -> None: """ Write out the new macro with all relevant commands to the versions.py file @@ -767,8 +765,8 @@ def write_new_macro(self, meta_dir, full_command, macro): - macro, the parsed macro being written """ - filepath = os.path.join(meta_dir, "versions.py") - temppath = os.path.join(meta_dir, ".versions.py") + filepath = meta_dir / "versions.py" + temppath = meta_dir / ".versions.py" shutil.copy(filepath, temppath) author = macro["author"] or self.author @@ -788,9 +786,11 @@ def write_new_macro(self, meta_dir, full_command, macro): apply_styling(temppath) - os.rename(temppath, filepath) + temppath.rename(filepath) - def check_missing_macros(self, meta_dir, meta_imports): + def check_missing_macros( + self, meta_dir: Path, meta_imports: list[str] + ) -> list[str]: """ Check through macros of imported metadata sections, returning list of any that aren't in the current section (identified by the after tag) @@ -831,7 +831,9 @@ def check_missing_macros(self, meta_dir, meta_imports): return missing_macros - def combine_missing_macros(self, meta_imports, missing_macros): + def combine_missing_macros( + self, meta_imports: list[str], missing_macros: list[str] + ) -> dict: """ Combine missing macro commands Inputs: @@ -872,7 +874,7 @@ def combine_missing_macros(self, meta_imports, missing_macros): return new_macros - def fix_missing_macros(self, meta_dir, meta_imports): + def fix_missing_macros(self, meta_dir: Path, meta_imports: list[str]) -> str | None: """ Function to handle checking and fixing of missing upgrade macros Inputs: @@ -905,12 +907,13 @@ def fix_missing_macros(self, meta_dir, meta_imports): return None - def order_meta_dirs(self): + def order_meta_dirs(self) -> list[str]: """ Order the self.meta_dirs list by metadata import order, such that sections higher up the import tree come first Create a networkx ordered graph, with nodes as the import tree and edges as the - import statements. Then recreate list from this + import statements. + Return a list of networkx topological_sort result """ import_graph = nx.DiGraph() @@ -924,7 +927,7 @@ def order_meta_dirs(self): # guaranteed for valid rose metadata return list(nx.topological_sort(import_graph)) - def preprocess_macros(self): + def preprocess_macros(self) -> None: """ Overarching function to pre-process added macros Run before running any rose macro upgrade commands" @@ -937,10 +940,10 @@ def preprocess_macros(self): # Get list of versions files to check - in both core and apps self.meta_dirs = self.meta_dirs.union( - self.find_meta_dirs(os.path.join(self.root_path, "rose-meta")) + self.find_meta_dirs(self.root_path / "rose-meta") ) self.meta_dirs = self.meta_dirs.union( - self.find_meta_dirs(os.path.join(self.core_source, "rose-meta")) + self.find_meta_dirs(self.core_source / "rose-meta") ) for meta_dir in self.meta_dirs: @@ -983,9 +986,7 @@ def preprocess_macros(self): self.target_macros[meta_dir]["imports"] = self.read_meta_imports(meta_dir) # Read through the versions.py file for python import statements - self.python_imports.update( - read_python_imports(os.path.join(meta_dir, "versions.py")) - ) + self.python_imports.update(read_python_imports(meta_dir / "versions.py")) # Now reconstruct the macro for all applications which have the newly # added macro or import metadata with the new macro @@ -1022,7 +1023,7 @@ def preprocess_macros(self): # Upgrade Apps Functions ############################################################################ - def metadata_check(self, meta_dir): + def metadata_check(self, meta_dir: Path) -> None: """ " Note: Not currently run - see comment below Run rose metadata-check on rose metadata directories to check the @@ -1032,7 +1033,7 @@ def metadata_check(self, meta_dir): """ print(f"[INFO] Checking metadata in {meta_dir}") - command = f"rose metadata-check -C {os.path.join(meta_dir, 'HEAD')}" + command = f"rose metadata-check -C {meta_dir / 'HEAD'}" result = run_command(command) if result.returncode: print(f"[FAIL] The metadata at {meta_dir} failed to validate.") @@ -1042,21 +1043,21 @@ def metadata_check(self, meta_dir): ) print(f"[PASS] {meta_dir} validated") - def get_rose_apps(self): + def get_rose_apps(self) -> set[Path]: """ Return: - - list of paths to rose-stem apps in Apps, Core and Jules + - set of paths to rose-stem apps in Apps and Core """ apps_list = [] for item in (self.root_path, self.core_source): - app_dir = os.path.join(item, "rose-stem", "app") - if not os.path.exists(app_dir): + app_dir = item / "rose-stem" / "app" + if not app_dir.exists(): continue - apps_list += [os.path.join(app_dir, f) for f in os.listdir(app_dir)] + apps_list += [app_dir / f for f in os.listdir(app_dir)] return set(apps_list) - def apps_to_upgrade(self): + def apps_to_upgrade(self) -> list[Path]: """ Loop over rose-stem apps, finding ones using metadata with an upgrade macro. @@ -1069,29 +1070,27 @@ def apps_to_upgrade(self): for app_path in apps_list: # Ignore lfric_coupled_rivers as this is based on Jules-standalone # metadata which is not currently available - if "fcm_make" in app_path or "lfric_coupled_rivers" in app_path: + if "fcm_make" in str(app_path) or "lfric_coupled_rivers" in str(app_path): continue - if not os.path.isdir(app_path): + if not app_path.is_dir(): continue - meta_import = self.read_meta_imports( - os.path.join(app_path, "rose-app.conf"), "meta" - ) + meta_import = self.read_meta_imports(app_path / "rose-app.conf", "meta") # If there was a metadata import, it is the first value in the list # It includes the version directory, so remove this to compare with # self.sections_with_macro if meta_import: - if os.path.dirname(meta_import[0]) in self.sections_with_macro: + if meta_import[0].parent in self.sections_with_macro: upgradeable_apps.append(app_path) return upgradeable_apps - def run_app_upgrade(self, app_path): + def run_app_upgrade(self, app_path: Path) -> None: """ Run 'rose app-upgrade' on a particular rose-stem app Inputs: - app_path, the path to this app """ - app = os.path.basename(app_path) + app = app_path.name print(f"[INFO] Upgrading the rose-stem app {app}") command = f"rose app-upgrade -a -y -C {app_path} {self.tag}" result = run_command(command) @@ -1103,14 +1102,14 @@ def run_app_upgrade(self, app_path): ) print(f"[PASS] Upgraded rose-stem app {app} successfully") - def run_macro_fix(self, app_path): + def run_macro_fix(self, app_path: Path) -> None: """ Run 'rose macro --fix' on a particular rose-stem app to force metadata consistency Inputs: - app_path, the path to this app """ - app = os.path.basename(app_path) + app = app_path.name print(f"[INFO] Forcing metadata consistency in app {app}") command = f"rose macro --fix -y -C {app_path}" result = run_command(command) @@ -1122,7 +1121,7 @@ def run_macro_fix(self, app_path): ) print(f"[PASS] Successfully forced metadata consistency in {app}") - def upgrade_apps(self): + def upgrade_apps(self) -> None: """ Overarching function to run rose commands to apply upgrade macros to rose-stem apps. @@ -1150,11 +1149,11 @@ def upgrade_apps(self): for app_path in upgradeable_apps: self.run_app_upgrade(app_path) self.run_macro_fix(app_path) - if app_path.startswith(self.core_source): + if self.core_source in app_path.parents: self.upgraded_core = True -def check_tag(opt): +def check_tag(opt: str | None) -> str | None: """ Check that a command line supplied tag is of a valid format """ @@ -1167,7 +1166,7 @@ def check_tag(opt): return opt -def version_number(opt): +def version_number(opt: str | None) -> str | None: """ Check that the command line supplied version number is of a suitable format """ @@ -1181,7 +1180,7 @@ def version_number(opt): return opt -def parse_args(): +def parse_args() -> argparse.Namespace: """ Read command line args """ @@ -1210,7 +1209,8 @@ def parse_args(): parser.add_argument( "-a", "--apps", - default=".", + type=Path, + default=Path(".").absolute(), help="The path to the LFRic Apps working copy being used. Defaults to " "the location the script is being run from - this assumes you are in a " "working copy.", @@ -1223,25 +1223,23 @@ def parse_args(): "Either a path to a working copy or a git source." "If not set, will be read from the dependencies.yaml", ) - parser.add_argument( - "-j", - "--jules", - default=None, - help="The Jules source being used." - "Either a path to a working copy or a git source." - "If not set, will be read from the dependencies.yaml", - ) return parser.parse_args() -def apply_macros_main(tag, cname=None, version=None, apps=".", core=None, jules=None): +def apply_macros_main( + tag: str, + cname: str | None = None, + version: str | None = None, + apps: Path = Path(".").absolute(), + core: str | None = None, +) -> None: """ Main function for this program """ check_environment() - macro_object = ApplyMacros(tag, cname, version, apps, core, jules) + macro_object: ApplyMacros = ApplyMacros(tag, cname, version, apps, core) # Pre-process macros banner_print("Pre-Processing Macros") @@ -1264,14 +1262,10 @@ def apply_macros_main(tag, cname=None, version=None, apps=".", core=None, jules= shutil.rmtree(directory) # Run rose config-dump on rose-stem - config_dump_apps = ( - f"rose config-dump {os.path.join(macro_object.root_path), 'rose-stem'}" - ) + config_dump_apps = f"rose config-dump {macro_object.root_path / 'rose-stem'}" run_command(config_dump_apps) if __name__ == "__main__": args = parse_args() - apply_macros_main( - args.tag, args.cname, args.version, args.apps, args.core, args.jules - ) + apply_macros_main(args.tag, args.cname, args.version, args.apps, args.core) diff --git a/lfric_macros/check_macro_chains.py b/lfric_macros/check_macro_chains.py index df834837..3748efb3 100755 --- a/lfric_macros/check_macro_chains.py +++ b/lfric_macros/check_macro_chains.py @@ -14,11 +14,12 @@ import re import shutil import sys +from pathlib import Path -from apply_macros import ApplyMacros, run_command +from apply_macros import ApplyMacros -def find_upgradeable_apps(apps_dir): +def find_upgradeable_apps(apps_dir: Path, core_dir: Path) -> dict: """ Loop over rose-stem apps installed into the cylc_workflow and return a list of those with metadata imports and therefore available for rose upgrade @@ -28,33 +29,38 @@ def find_upgradeable_apps(apps_dir): Returns: - list of app names that can be upgraded """ - - valid_apps = [] - for app in os.listdir(apps_dir): - conf_path = os.path.join(apps_dir, app, "rose-app.conf") - if not os.path.isfile(conf_path): - continue - grep_com = f'grep -E "meta=.*" {conf_path}' - result = run_command(grep_com) - if "meta=" in result.stdout: - valid_apps.append(app) - + for start in (apps_dir, core_dir): + valid_apps = {} + for app in start.iterdir(): + conf_path = app / "rose-app.conf" + if not conf_path.is_file(): + continue + with open(conf_path) as f: + for line in f: + try: + version = re.search(r"\s*meta\s*=\s*([\w\.\-\/]+)", line).group( + 1 + ) + valid_apps[app] = version.split("/") + break + except AttributeError: + pass return valid_apps -def find_macro_tags(tag, path, errors): +def find_macro_tags(tag: str, path: Path, errors: list) -> set[str]: """ Find tags with format BEFORE_TAG= or AFTER_TAG= in a versions.py file. Inputs: tag - either before or after - path, path to the directory containing the versions.py file + path - path to the directory containing the versions.py file Returns: set of tags found in file """ found_tags = set() in_comment = False - with open(os.path.join(path, "versions.py")) as f: + with open(path / "versions.py") as f: for line in f: line = line.strip() # Check whether this is a comment @@ -75,13 +81,15 @@ def find_macro_tags(tag, path, errors): return found_tags -def compare_tags(before, after, path, errors): +def compare_tags(before: str, after: str, path: Path, errors: list) -> str | None: """ Check that the before and after tags form a continuous chain. This is done by ensuring that only the initial before tag (the version number) and the final after tag are not in both sets. Inputs: before/after, sets of the tags in a given file + Returns: + Final after tag in the macro chain """ # Tags in only one of before and after sets @@ -89,23 +97,29 @@ def compare_tags(before, after, path, errors): # There should be 2 single tags if len(single_tags) != 2 and len(single_tags) != 0: - errors.append( - f"[ERROR] - Found {len(single_tags)} unique before or after tags in " - f"{os.path.join(path, 'versions.py')} that were ONLY a before or " - "after tag.\nThere should be 2 of these - the beginning of the " - "chain and the end of the chain.\nThis is likely a typo in the tags in " - "the versions.py file. The identified tags were:\n" - ) + "\n".join(x for x in single_tags) + msg = ( + f"[ERROR] - Found {len(single_tags)} tags in {path / 'versions.py'} that " + "didn't have a 2nd instance.\nThere should be 2 of these - the beginning " + "of the chain and the end of the chain.\nThis is likely a typo in the tags " + "in the versions.py file. The identified tags were:\n" + ) + msg += "\n".join(x for x in single_tags) + errors.append(msg) + return None + + for item in single_tags: + if item in after: + return item -def check_fcm(): +def check_fcm() -> None: """ Check if this script is being run for a fcm working copy and fail gracefully if so. """ - dependency = os.path.join(os.environ["SOURCE_ROOT"], "apps", "dependencies.sh") + dependency = Path(os.environ["SOURCE_ROOT"]) / "apps" / "dependencies.sh" - if os.path.exists(dependency): + if dependency.exists(): raise Exception( "[ERROR] check_macro_chains.py no longer works with FCM sources. " "Please ignore this error until you have migrated your work " @@ -113,40 +127,63 @@ def check_fcm(): ) -def main(): +def main() -> None: """ Main function of the program """ check_fcm() - source_apps = os.path.join(os.environ["SOURCE_ROOT"], "lfric_apps") - source_core = os.path.join(os.environ["SOURCE_ROOT"], "lfric_core") + source_apps = Path(os.environ["SOURCE_ROOT"]) / "lfric_apps" + source_core = Path(os.environ["SOURCE_ROOT"]) / "lfric_core" - macro_object = ApplyMacros( - "vn0.0_t0", None, "vn0.0", source_apps, source_core, None + macro_object = ApplyMacros("vn0.0_t0", None, "vn0.0", source_apps, source_core) + apps_meta_dirs = macro_object.find_meta_dirs(macro_object.root_path / "rose-meta") + + rose_apps = find_upgradeable_apps( + source_apps / "rose-stem" / "app", source_core / "rose-stem" / "app" ) - macro_object.find_meta_dirs(os.path.join(macro_object.root_path, "applications")) - macro_object.meta_dirs + latest_meta = {} errors = [] - for meta_dir in macro_object.meta_dirs: + for meta_dir in apps_meta_dirs: before_tags = find_macro_tags("before", meta_dir, errors) after_tags = find_macro_tags("after", meta_dir, errors) - compare_tags(before_tags, after_tags, meta_dir, errors) - - # Remove temp directories - for _, directory in macro_object.temp_dirs.items(): - shutil.rmtree(directory) + latest = compare_tags(before_tags, after_tags, meta_dir, errors) + if latest: + latest_meta[meta_dir.name] = latest if errors: for item in errors: print(f"{item}\n\n\n\n", file=sys.stderr) print("[FAIL] - Found errors in macro chains - please check the job.err") exit(1) - print("[PASS] - Successfully checked all macro chains") + errors = [] + for app, version in rose_apps.items(): + if version[0] not in latest_meta: + continue + if version[1] != latest_meta[version[0]]: + msg = ( + f"The rose-stem app {app} is using a different macro tag " + f"'{version[0]}/{version[1]}' compared with the latest upgrade macro, " + f"'{latest_meta[version[0]]}'. This suggests a macro has not been " + "successfully applied." + ) + errors.append(msg) + + if errors: + for item in errors: + print(f"{item}\n\n\n\n", file=sys.stderr) + print("[FAIL] - Found errors in rose app versions - please check the job.err") + exit(1) + print("[PASS] - Successfully checked all rose-stem apps") + + # Remove temp directories + for _, directory in macro_object.temp_dirs.items(): + shutil.rmtree(directory) + if __name__ == "__main__": main() diff --git a/lfric_macros/release_lfric.py b/lfric_macros/release_lfric.py index a286fc4f..567dd96a 100755 --- a/lfric_macros/release_lfric.py +++ b/lfric_macros/release_lfric.py @@ -18,12 +18,12 @@ import argparse import getpass -import os import re import socket import subprocess import shutil import shlex +from pathlib import Path from apply_macros import ( ApplyMacros, @@ -47,7 +47,7 @@ def upgrade(self, config, meta_config=None): """ -def run_command(command, timelimit=120): +def run_command(command: str, timelimit: int = 120) -> subprocess.CompletedProcess: """ Run a subprocess command and return the result object Inputs: @@ -69,7 +69,7 @@ def run_command(command, timelimit=120): return result -def raise_exception(result, command): +def raise_exception(result: subprocess.CompletedProcess, command: str) -> None: """ Raise an exception if a subprocess command has failed """ @@ -77,7 +77,7 @@ def raise_exception(result, command): raise Exception(f"[FAIL] Error running command: '{command}'\n{result.stderr}") -def set_dependency_path(args): +def set_dependency_path(apps: Path, core: Path) -> None: """ Edit an LFRic Apps dependencies.sh file so that it points at the provided LFRic Core source @@ -86,7 +86,7 @@ def set_dependency_path(args): print("[INFO] Updating dependencies.yaml Core source") hostname = socket.gethostname() - dep_path = os.path.join(args.apps, "dependencies.yaml") + dep_path = apps / "dependencies.yaml" with open(dep_path) as f: lines = f.readlines() in_core = False @@ -95,7 +95,7 @@ def set_dependency_path(args): in_core = True elif in_core and "source:" in line: prefix, _, _ = line.partition("source:") - line = f"{prefix}source: {hostname}:{os.path.abspath(args.core)}\n" + line = f"{prefix}source: {hostname}:{core}\n" elif in_core and "ref:" in line: prefix, _, _ = line.partition("ref:") line = f"{prefix}ref:" @@ -106,25 +106,25 @@ def set_dependency_path(args): f.write("".join(x for x in lines)) -def find_meta_dirs(paths, exclude_dirs=()): +def find_meta_dirs(paths: list[Path], exclude_dirs: tuple[str] = ()) -> set[Path]: """ Return a set of rose-metadata directories that can be found in the apps and core sources. Done by seaching for rose-meta.conf files. Records the parent - directory of the current one, as rose-meta.conf files end up in HEAD/vnX.Y + directory of the current one, as rose-meta.conf files end up in 'HEAD' or 'vnX.Y' directories. """ dirs = set() for path in paths: print("[INFO] Finding rose metadata directories in", path) - for dirpath, dirnames, filenames in os.walk(path): + for dirpath, dirnames, filenames in path.walk(): dirnames[:] = [d for d in dirnames if d not in exclude_dirs] if "rose-meta.conf" in filenames: - dirs.add(os.path.dirname(dirpath)) + dirs.add(dirpath.parent) return dirs -def update_version_number(args): +def update_version_number(apps: Path, version: str) -> None: """ Update the "VN" variable number in the lfric_apps rose-suite.conf file, to be the new version number @@ -132,14 +132,14 @@ def update_version_number(args): print("[INFO] Updating rose-suite.conf version number") - fpath = os.path.join(args.apps, "rose-stem", "rose-suite.conf") + fpath = apps / "rose-stem" / "rose-suite.conf" with open(fpath, "r") as f: lines = f.readlines() for i, line in enumerate(lines): line = line.strip() if line.startswith("VN="): - line = f"VN='{args.version.removeprefix('vn')}'" + line = f"VN='{version.removeprefix('vn')}'" lines[i] = line break @@ -148,16 +148,16 @@ def update_version_number(args): f.write(line) -def update_variables_files(apps): +def update_variables_files(apps: Path) -> None: """ Edit meto variables_platforms.cylc files to remove any ticket updates """ - meto_path = os.path.join(apps, "rose-stem", "site", "meto") + meto_path = apps / "rose-stem" / "site" / "meto" variables_files = set() - for filename in os.listdir(meto_path): - if filename.startswith("variables_"): - variables_files.add(os.path.join(meto_path, filename)) + for filename in meto_path.iterdir(): + if str(filename).startswith("variables_"): + variables_files.add(meto_path / filename) for fpath in variables_files: with open(fpath, "r") as f: @@ -173,14 +173,20 @@ def update_variables_files(apps): f.write(line) -def get_user(): +def get_user() -> str: """ Return a str of username with .'s replaced by ' ' """ return getpass.getuser().replace(".", " ") -def add_new_upgrade_macro(meta_dirs, args, macro_object): +def add_new_upgrade_macro( + meta_dirs: list[Path], + old_version: str, + version: str, + ticket: str, + macro_object: ApplyMacros, +) -> None: """ Write out a new macro, updating to vnX.Y Use the template macro in the MACRO_TEMPLATE variable above @@ -192,14 +198,14 @@ def add_new_upgrade_macro(meta_dirs, args, macro_object): template_macro = MACRO_TEMPLATE # Replace Consistent Variables for all meta directories - class_name = f"{args.old_version.replace('.', '')}_t{args.ticket}" + class_name = f"{old_version.replace('.', '')}_t{ticket}" template_macro = template_macro.replace("CLASS_NAME", class_name) - template_macro = template_macro.replace("TICKET", args.ticket) + template_macro = template_macro.replace("TICKET", ticket) template_macro = template_macro.replace("AUTHOR", get_user()) - template_macro = template_macro.replace("AFTER_EDIT", args.version) + template_macro = template_macro.replace("AFTER_EDIT", version) for meta_dir in meta_dirs: - versions_file = os.path.join(meta_dir, "versions.py") + versions_file = meta_dir / "versions.py" macros = read_versions_file(meta_dir) macros = split_macros(macros) @@ -213,7 +219,7 @@ def add_new_upgrade_macro(meta_dirs, args, macro_object): f.write(f"\n{meta_dir_macro}") -def copy_head_meta(meta_dirs, args): +def copy_head_meta(meta_dirs: list[Path], apps: Path, core: Path, version: str) -> None: """ Copy the HEAD metadata to vnX.Y/ for all meta_dirs """ @@ -221,21 +227,21 @@ def copy_head_meta(meta_dirs, args): print("[INFO] Copying HEAD metadata") for meta_dir in meta_dirs: - head = os.path.join(meta_dir, "HEAD") - new = os.path.join(meta_dir, args.version) + head = meta_dir / "HEAD" + new = meta_dir / version shutil.copytree(head, new) - if args.core in new: - new = new.removeprefix(args.core) - new = new.lstrip("/") - command = f"git -C {args.core} add {new}" - elif args.apps in new: - new = new.removeprefix(args.apps) - new = new.lstrip("/") - command = f"git -C {args.apps} add {new}" + if core in new.parents: + new = new.relative_to(core) + print(new) + command = f"git -C {core} add {new}" + elif apps in new.parents: + new = new.relative_to(apps) + print(new) + command = f"git -C {apps} add {new}" _ = run_command(command) -def update_meta_import_path(meta_dirs, args): +def update_meta_import_path(meta_dirs: list[Path], version: str) -> None: """ Change HEAD to vnX.Y in meta import statements in the newly created vnX.Y/rose-meta.conf files @@ -244,7 +250,7 @@ def update_meta_import_path(meta_dirs, args): print("[INFO] Updating metadata import statements") for meta_dir in meta_dirs: - meta_file = os.path.join(meta_dir, args.version, "rose-meta.conf") + meta_file = meta_dir / version / "rose-meta.conf" with open(meta_file) as f: lines = f.readlines() @@ -255,7 +261,7 @@ def update_meta_import_path(meta_dirs, args): elif in_imports and not line.strip().startswith("="): break if in_imports: - line = line.replace("HEAD", args.version) + line = line.replace("HEAD", version) lines[i] = line with open(meta_file, "w") as f: @@ -263,39 +269,39 @@ def update_meta_import_path(meta_dirs, args): f.write(line) -def copy_versions_files(meta_dirs, args): +def copy_versions_files( + meta_dirs: list[Path], old_version: str, version: str, apps: Path, core: Path +) -> str: """ Copy versions.py files to versionAB_XY.py Returns the name of the AB->XY versions files. """ upgrade_name = ( - f"version{args.old_version.replace('.', '').replace('vn', '')}_" - f"{args.version.replace('.', '').replace('vn', '')}.py" + f"version{old_version.replace('.', '').replace('vn', '')}_" + f"{version.replace('.', '').replace('vn', '')}.py" ) print("[INFO] Copying versions.py files to versionAB_XY.py files") for meta_dir in meta_dirs: - versions_file = os.path.join(meta_dir, "versions.py") - upgrade_file = os.path.join(meta_dir, upgrade_name) - if not os.path.exists(versions_file): + versions_file = meta_dir / "versions.py" + upgrade_file = meta_dir / upgrade_name + if not versions_file.exists(): raise FileNotFoundError(f"The file {versions_file} doesn't exist") shutil.copyfile(versions_file, upgrade_file) - if args.core in upgrade_file: - upgrade_file = upgrade_file.removeprefix(args.core) - upgrade_file = upgrade_file.lstrip("/") - command = f"git -C {args.core} add {upgrade_file}" - elif args.apps in upgrade_file: - upgrade_file = upgrade_file.removeprefix(args.apps) - upgrade_file = upgrade_file.lstrip("/") - command = f"git -C {args.apps} add {upgrade_file}" + if core in upgrade_file.parents: + upgrade_file = upgrade_file.relative_to(core) + command = f"git -C {core} add {upgrade_file}" + elif apps in upgrade_file.parents: + upgrade_file = upgrade_file.relative_to(apps) + command = f"git -C {apps} add {upgrade_file}" _ = run_command(command) return upgrade_name -def add_new_import(versions_file, upgrade_name): +def add_new_import(versions_file: Path, upgrade_name: str) -> None: """ Read through a versions.py file, finding the line that imports MacroUpgrade from rose. Add the new `from .versionsAB_XY import *` import after that line @@ -330,7 +336,7 @@ def add_new_import(versions_file, upgrade_name): apply_styling(versions_file) -def update_versions_file(meta_dirs, upgrade_name): +def update_versions_file(meta_dirs: list[Path], upgrade_name: str) -> None: """ - Add import of versionAB_XY.py file to the template_versions.py - Replace old versions.py files with that file @@ -338,19 +344,15 @@ def update_versions_file(meta_dirs, upgrade_name): print("[INFO] Updating versions.py files") - template_path = os.path.join( - os.path.dirname(os.path.abspath(__file__)), - "files", - "template_versions.py", - ) + template_path = Path(__file__).absolute().parent / "files" / "template_versions.py" for meta_dir in meta_dirs: - versions_file = os.path.join(meta_dir, "versions.py") + versions_file = meta_dir / "versions.py" shutil.copyfile(template_path, versions_file) add_new_import(versions_file, upgrade_name) -def ticket_number(opt): +def ticket_number(opt: str) -> str: """ Check that the command line supplied ticket number is of a suitable format """ @@ -362,7 +364,7 @@ def ticket_number(opt): return opt -def parse_args(): +def parse_args() -> argparse.Namespace: """ Read command line args """ @@ -392,7 +394,8 @@ def parse_args(): parser.add_argument( "-a", "--apps", - default=".", + default=Path(".").absolute(), + type=Path, help="The path to the LFRic Apps working copy being used. Defaults to " "the location the script is being run from - this assumes you are in a " "working copy.", @@ -401,20 +404,20 @@ def parse_args(): "-c", "--core", required=True, + type=Path, help="Path to the LFRic Core working copy being used.", ) args = parser.parse_args() - args.apps = os.path.abspath(args.apps) args.apps = get_root_path(args.apps) - args.core = os.path.abspath(args.core) + args.core = args.core.expanduser().absolute() args.version = f"vn{args.version}" args.old_version = f"vn{args.old_version}" return args -def main(): +def main() -> None: args = parse_args() macro_object = ApplyMacros( @@ -423,12 +426,12 @@ def main(): args.old_version.removeprefix("vn"), args.apps, args.core, - None, ) - set_dependency_path(args) + set_dependency_path(args.apps, args.core) - # Find all metadata directories, excluing jules shared and lfric inputs as these have metadata but no macros. + # Find all metadata directories, excluing jules shared and lfric inputs as these + # have metadata but no macros. exclude_dirs = ( ".svn", "rose-stem", @@ -438,18 +441,25 @@ def main(): ) meta_dirs = find_meta_dirs([args.apps, args.core], exclude_dirs) - # Find JULES shared metadata directories and combine with all other metadirs for where they are handled differently - jules_meta_path = os.path.join( - args.apps, "interfaces", "jules_interface", "rose-meta", "lfric-jules-shared" + # Find JULES shared metadata directories and combine with all other metadirs for + # where they are handled differently + jules_meta_path = ( + args.apps + / "interfaces" + / "jules_interface" + / "rose-meta" + / "lfric-jules-shared" ) jules_shared_meta_dirs = find_meta_dirs([jules_meta_path]) meta_dirs_plus_jules = meta_dirs.union(jules_shared_meta_dirs) - update_version_number(args) + update_version_number(args.apps, args.version) update_variables_files(args.apps) - add_new_upgrade_macro(meta_dirs, args, macro_object) + add_new_upgrade_macro( + meta_dirs, args.old_version, args.version, args.ticket, macro_object + ) # Run the apply_macros script apply_macros_main( @@ -458,15 +468,16 @@ def main(): args.old_version, args.apps, args.core, - None, ) print("\n[INFO] Successfully upgraded apps") - copy_head_meta(meta_dirs_plus_jules, args) + copy_head_meta(meta_dirs_plus_jules, args.apps, args.core, args.version) - update_meta_import_path(meta_dirs, args) + update_meta_import_path(meta_dirs, args.version) - upgrade_file_name = copy_versions_files(meta_dirs, args) + upgrade_file_name = copy_versions_files( + meta_dirs, args.old_version, args.version, args.apps, args.core + ) update_versions_file(meta_dirs, upgrade_file_name) diff --git a/lfric_macros/tests/test_apply_macros.py b/lfric_macros/tests/test_apply_macros.py index c6928d4f..c4e3f78c 100644 --- a/lfric_macros/tests/test_apply_macros.py +++ b/lfric_macros/tests/test_apply_macros.py @@ -7,7 +7,7 @@ Unit tests for apply_macros """ -from os import path +from pathlib import Path import pytest from ..apply_macros import ( @@ -20,11 +20,9 @@ ) # Commonly used paths -TEST_APPS_DIR = path.join( - path.dirname(path.dirname(path.abspath(__file__))), "tests", "test_lfric_apps_dir" -) -TEST_META_DIR = path.join(TEST_APPS_DIR, "rose-meta") -TEST_ROSE_STEM = path.join(TEST_APPS_DIR, "rose-stem", "app") +TEST_APPS_DIR = Path(__file__).absolute().parent / "test_lfric_apps_dir" +TEST_META_DIR = TEST_APPS_DIR / "rose-meta" +TEST_ROSE_STEM = TEST_APPS_DIR / "rose-stem" / "app" # A macro that we want to find for these tests @@ -72,7 +70,7 @@ def __repr__(self): # Create an instance of the apply_macros class # Use /tmp for Core and Jules as these are not required for testing -applymacros = ApplyMacros("vn0.0_t001", None, None, TEST_APPS_DIR, "/tmp", "/tmp", True) +applymacros = ApplyMacros("vn0.0_t001", None, None, TEST_APPS_DIR, Path("/tmp"), True) def test_read_versions_file(): @@ -80,19 +78,20 @@ def test_read_versions_file(): Test read_versions_file """ - assert read_versions_file( - path.join(TEST_APPS_DIR, "rose-meta", "lfric-gungho") - ) == ["# line 1\n", "# line 2\n"] + assert read_versions_file(TEST_APPS_DIR / "rose-meta" / "lfric-gungho") == [ + "# line 1\n", + "# line 2\n", + ] def test_find_meta_dirs(): - result = applymacros.find_meta_dirs(path.join(TEST_APPS_DIR, "rose-meta")) + result = applymacros.find_meta_dirs(TEST_APPS_DIR / "rose-meta") expected = set() - expected.add(path.join(TEST_META_DIR, "lfric-lfric_atm")) - expected.add(path.join(TEST_META_DIR, "lfric-gungho")) - expected.add(path.join(TEST_META_DIR, "lfric-driver")) - expected.add(path.join(TEST_META_DIR, "um-iau")) - expected.add(path.join(TEST_META_DIR, "lfric-transport")) + expected.add(TEST_META_DIR / "lfric-lfric_atm") + expected.add(TEST_META_DIR / "lfric-gungho") + expected.add(TEST_META_DIR / "lfric-driver") + expected.add(TEST_META_DIR / "um-iau") + expected.add(TEST_META_DIR / "lfric-transport") assert result == expected applymacros.meta_dirs = result @@ -136,7 +135,7 @@ def test_read_python_imports(): Test read_python_imports """ - test_file = path.join(TEST_APPS_DIR, "test_read_python_imports.py") + test_file = TEST_APPS_DIR / "test_read_python_imports.py" expected = set() imports = ( (("shlex",), ("split",), None), @@ -151,11 +150,13 @@ def test_read_python_imports(): def test_find_macro(): - assert applymacros.find_macro("meta_dir", expected_split_macros) == desired_macro - assert applymacros.find_macro("meta_dir", [existing_macro]) == "" + assert ( + applymacros.find_macro(Path("meta_dir"), expected_split_macros) == desired_macro + ) + assert applymacros.find_macro(Path("meta_dir"), [existing_macro]) == "" expected_error = r".*meta_dir/versions.py.*" with pytest.raises(Exception, match=expected_error): - applymacros.find_macro("meta_dir", [""]) + applymacros.find_macro(Path("meta_dir"), [""]) def test_find_last_macro(): @@ -167,7 +168,7 @@ def test_find_last_macro(): def test_parse_macro(): for macro in (existing_macro, desired_macro): applymacros.parsed_macros["meta_dir"].append( - applymacros.parse_macro(macro, "meta_dir") + applymacros.parse_macro(macro, Path("meta_dir")) ) expected_macros_list = [ { @@ -197,25 +198,23 @@ def test_parse_macro(): ] assert applymacros.parsed_macros["meta_dir"] == expected_macros_list with pytest.raises(Exception, match=r".*failed/versions.py"): - applymacros.parse_macro("", "failed") + applymacros.parse_macro("", Path("failed")) def test_check_missing_macro(): macros = applymacros.parsed_macros["meta_dir"] - applymacros.parsed_macros[path.join(TEST_META_DIR, "lfric-gungho")] = macros + applymacros.parsed_macros[TEST_META_DIR / "lfric-gungho"] = macros missing = applymacros.check_missing_macros( - path.join(TEST_META_DIR, "lfric-lfric_atm"), ["lfric-gungho"] + TEST_META_DIR / "lfric-lfric_atm", ["lfric-gungho"] ) assert missing == ["vn0.0_t000"] - missing = applymacros.check_missing_macros( - path.join(TEST_META_DIR, "lfric-gungho"), [] - ) + missing = applymacros.check_missing_macros(TEST_META_DIR / "lfric-gungho", []) assert missing == [] def test_combine_missing_macros(): combined = applymacros.combine_missing_macros( - [path.join(TEST_META_DIR, "lfric-gungho")], ["vn0.0_t000"] + [TEST_META_DIR / "lfric-gungho"], ["vn0.0_t000"] ) expected_combined = { "vn0.0_t000": { @@ -237,20 +236,18 @@ def test_combine_missing_macros(): def test_read_meta_imports(): applymacros.parsed_macros[TEST_APPS_DIR] = {} applymacros.parsed_macros[TEST_APPS_DIR]["imports"] = applymacros.read_meta_imports( - path.join(TEST_APPS_DIR, "rose-meta", "lfric-gungho") + TEST_APPS_DIR / "rose-meta" / "lfric-gungho" ) expected_imports = [ - path.join(applymacros.root_path, "rose-meta", "lfric-driver"), - path.join(applymacros.root_path, "rose-meta", "um-iau"), + applymacros.root_path / "rose-meta" / "lfric-driver", + applymacros.root_path / "rose-meta" / "um-iau", ] assert applymacros.parsed_macros[TEST_APPS_DIR]["imports"] == expected_imports - expected_meta = [ - path.join(applymacros.root_path, "rose-meta", "lfric-lfric_atm", "vn0.0") - ] + expected_meta = [applymacros.root_path / "rose-meta" / "lfric-lfric_atm" / "vn0.0"] assert ( applymacros.read_meta_imports( - path.join(TEST_APPS_DIR, "rose-stem", "app", "lfric_atm", "rose-app.conf"), + TEST_APPS_DIR / "rose-stem" / "app" / "lfric_atm" / "rose-app.conf", "meta", ) == expected_meta @@ -290,26 +287,18 @@ def test_combine_macros(): def test_parse_application_section(): - assert ( - applymacros.parse_application_section(path.join("meta_dir", "HEAD")) - == "meta_dir" - ) - assert ( - applymacros.parse_application_section(path.join("meta_dir", "versions.py")) - == "meta_dir" - ) - assert ( - applymacros.parse_application_section( - path.join(applymacros.root_path, "meta_dir") - ) - == "meta_dir" - ) - assert ( - applymacros.parse_application_section( - path.join(applymacros.core_source, "meta_dir") - ) - == "meta_dir" + assert applymacros.parse_application_section(Path("meta_dir") / "HEAD") == Path( + "meta_dir" ) + assert applymacros.parse_application_section( + Path("meta_dir") / "versions.py" + ) == Path("meta_dir") + assert applymacros.parse_application_section( + applymacros.root_path / "meta_dir" + ) == Path("meta_dir") + assert applymacros.parse_application_section( + applymacros.core_source / "meta_dir" + ) == Path("meta_dir") def test_read_dependencies(): @@ -318,26 +307,26 @@ def test_read_dependencies(): def test_order_meta_dirs(): applymacros.target_macros = { - path.join(TEST_META_DIR, "lfric-driver"): {"imports": []}, - path.join(TEST_META_DIR, "lfric-gungho"): { + TEST_META_DIR / "lfric-driver": {"imports": []}, + TEST_META_DIR / "lfric-gungho": { "imports": [ - path.join(TEST_META_DIR, "lfric-driver"), - path.join(TEST_META_DIR, "um-iau"), + TEST_META_DIR / "lfric-driver", + TEST_META_DIR / "um-iau", ] }, - path.join(TEST_META_DIR, "um-iau"): {"imports": []}, - path.join(TEST_META_DIR, "lfric-lfric_atm"): { - "imports": [path.join(TEST_META_DIR, "lfric-gungho")] + TEST_META_DIR / "um-iau": {"imports": []}, + TEST_META_DIR / "lfric-lfric_atm": { + "imports": [TEST_META_DIR / "lfric-gungho"] }, - path.join(TEST_META_DIR, "lfric-transport"): { - "imports": [path.join(TEST_META_DIR, "lfric-gungho")] + TEST_META_DIR / "lfric-transport": { + "imports": [TEST_META_DIR / "lfric-gungho"] }, } order = applymacros.order_meta_dirs() - gungho = order.index(path.join(TEST_META_DIR, "lfric-gungho")) - lfric_atm = order.index(path.join(TEST_META_DIR, "lfric-lfric_atm")) - driver = order.index(path.join(TEST_META_DIR, "lfric-driver")) - um = order.index(path.join(TEST_META_DIR, "um-iau")) + gungho = order.index(TEST_META_DIR / "lfric-gungho") + lfric_atm = order.index(TEST_META_DIR / "lfric-lfric_atm") + driver = order.index(TEST_META_DIR / "lfric-driver") + um = order.index(TEST_META_DIR / "um-iau") assert gungho > driver assert gungho > um assert lfric_atm > gungho @@ -345,19 +334,19 @@ def test_order_meta_dirs(): def test_get_rose_apps(): expected = set() - expected.add(path.join(TEST_ROSE_STEM, "gungho")) - expected.add(path.join(TEST_ROSE_STEM, "lfric_atm")) - expected.add(path.join(TEST_ROSE_STEM, "transport")) + expected.add(TEST_ROSE_STEM / "gungho") + expected.add(TEST_ROSE_STEM / "lfric_atm") + expected.add(TEST_ROSE_STEM / "transport") assert applymacros.get_rose_apps() == expected def test_apps_to_upgrade(): applymacros.sections_with_macro = [ - path.join(TEST_META_DIR, "lfric-gungho"), - path.join(TEST_META_DIR, "lfric-lfric_atm"), + TEST_META_DIR / "lfric-gungho", + TEST_META_DIR / "lfric-lfric_atm", ] expected = ( - [path.join(TEST_ROSE_STEM, "gungho"), path.join(TEST_ROSE_STEM, "lfric_atm")], - [path.join(TEST_ROSE_STEM, "lfric_atm"), path.join(TEST_ROSE_STEM, "gungho")], + [TEST_ROSE_STEM / "gungho", TEST_ROSE_STEM / "lfric_atm"], + [TEST_ROSE_STEM / "lfric_atm", TEST_ROSE_STEM / "gungho"], ) assert applymacros.apps_to_upgrade() in expected diff --git a/lfric_macros/validate_rose_meta.py b/lfric_macros/validate_rose_meta.py index 7ea21f70..c7f0f2e2 100755 --- a/lfric_macros/validate_rose_meta.py +++ b/lfric_macros/validate_rose_meta.py @@ -12,6 +12,7 @@ import sys import subprocess import argparse +from pathlib import Path # A list of invalid metadata sections. Most are invalid as they are imported by # lfric-gungho but also use the files namelist contained there, creating a circular @@ -43,7 +44,9 @@ ] -def run_command(command, shell=False, env=None): +def run_command( + command: str, shell: bool = False, env: dict = None +) -> subprocess.CompletedProcess: """ Run a subprocess command and return the result object Inputs: @@ -66,7 +69,7 @@ def run_command(command, shell=False, env=None): ) -def check_rose_metadata(rose_meta_path, source_path): +def check_rose_metadata(rose_meta_path: str, source_path: Path) -> bool: """ Auto find rose-meta sections from the top level rose-meta directory and run `rose metadata-check` on each @@ -79,12 +82,11 @@ def check_rose_metadata(rose_meta_path, source_path): my_env = os.environ.copy() my_env["ROSE_META_PATH"] = rose_meta_path - start_dir = os.path.join(source_path, "rose-meta") - dirs = os.listdir(start_dir) - for section in dirs: - if section in INVALID_METADATA: + start_dir = source_path / "rose-meta" + for section in start_dir.iterdir(): + if str(section.relative_to(start_dir)) in INVALID_METADATA: continue - meta_dir = os.path.join(start_dir, section, "HEAD") + meta_dir = start_dir / section / "HEAD" command = f"rose metadata-check --verbose -C {meta_dir}" result = run_command(command, env=my_env) if result.returncode: @@ -100,7 +102,7 @@ def check_rose_metadata(rose_meta_path, source_path): return failed -def parse_suite_controlled(err_msg): +def parse_suite_controlled(err_msg: str) -> list: """ Remove any app validation error messages resulting from suite_controlled option configs @@ -124,7 +126,7 @@ def parse_suite_controlled(err_msg): return [] -def check_rose_stem_apps(meta_paths, source_path): +def check_rose_stem_apps(meta_paths: str, source_path: Path) -> bool: """ Auto find rose-stem apps that use rose metadata and validate these using 'rose macro --validate' @@ -133,13 +135,12 @@ def check_rose_stem_apps(meta_paths, source_path): print("\n\n[INFO] - Validating rose-stem apps\n\n") failed = False - start_dir = os.path.join(source_path, "rose-stem", "app") - apps = os.listdir(start_dir) - for app in apps: - if app in INVALID_APPS: + start_dir = source_path / "rose-stem" / "app" + for app in start_dir.iterdir(): + if str(app.relative_to(start_dir)) in INVALID_APPS: continue - app_dir = os.path.join(start_dir, app) - conf_file = os.path.join(app_dir, "rose-app.conf") + app_dir = start_dir / app + conf_file = app_dir / "rose-app.conf" with open(conf_file, "r") as f: for line in f: if line.startswith("meta="): @@ -166,7 +167,7 @@ def check_rose_stem_apps(meta_paths, source_path): return failed -def parse_args(): +def parse_args() -> argparse.Namespace: """ Read command line args """ @@ -200,14 +201,14 @@ def parse_args(): ) if args.apps: - args.apps = os.path.expanduser(args.apps) + args.apps = Path(args.apps).absolute().expanduser() if args.core: - args.core = os.path.expanduser(args.core) + args.core = Path(args.core).absolute().expanduser() return args -def main(): +def main() -> None: """ main function for this script """ @@ -217,17 +218,17 @@ def main(): rose_meta_path = "" if args.apps: source_path = args.apps - meta_paths += f"-M {os.path.join(args.apps, 'rose-meta')} " - rose_meta_path += f"{os.path.join(args.apps, 'rose-meta')}" + meta_paths += f"-M {args.apps / 'rose-meta'} " + rose_meta_path += f"{args.apps / 'rose-meta'}" if args.core: - meta_paths += f"-M {os.path.join(args.core, 'rose-meta')} " + meta_paths += f"-M {args.core / 'rose-meta'} " if rose_meta_path: # Apps has already started this - rose_meta_path += f":{os.path.join(args.core, 'rose-meta')}" + rose_meta_path += f":{args.core / 'rose-meta'}" else: # Apps hasn't been set source_path = args.core - rose_meta_path = f"{os.path.join(args.core, 'rose-meta')}" + rose_meta_path = f"{args.core / 'rose-meta'}" if check_rose_metadata(rose_meta_path, source_path) or check_rose_stem_apps( meta_paths, source_path