Coverage for src/CSET/extract_workflow.py: 95%
103 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:03 +0000
1# © Crown copyright, Met Office (2022-2026) and CSET contributors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Extract the CSET cylc workflow for use."""
17import importlib.metadata
18import importlib.resources
19import logging
20import os
21import re
22import shutil
23import stat
24import subprocess
25import tempfile
26from pathlib import Path
28import CSET.cset_workflow
30logger = logging.getLogger(__name__)
33def make_script_executable(p: Path):
34 """Make a script file (starting with a shebang) executable."""
35 if p.is_file():
36 try:
37 with open(p, "rb") as fd:
38 shebang = fd.read(14)
39 except PermissionError:
40 # Skip files that can't be read.
41 logger.debug("Unreadable file: %s", p)
42 return
43 # Assume the first 14 bytes of a script are #!/usr/bin/env
44 if shebang == b"#!/usr/bin/env":
45 logger.debug("Changing file mode to executable: %s", p)
46 mode = p.stat().st_mode
47 # User must be able to read if we read the file.
48 mode |= stat.S_IXUSR
49 # Make executable by group and/or others if they can read.
50 if mode & stat.S_IRGRP:
51 mode |= stat.S_IXGRP
52 if mode & stat.S_IROTH:
53 mode |= stat.S_IXOTH
54 p.chmod(mode)
57def install_workflow(location: Path) -> Path:
58 """Install the workflow's files and link the conda environment.
60 Parameters
61 ----------
62 location: Path
63 A directory where the workflow files are to be installed to. A
64 sub-directory named "cset-workflow-vX.Y.Z" will be created under here.
66 Returns
67 -------
68 workflow_dir: Path
69 Path to newly created workflow directory.
70 """
71 # Check location's parents exist.
72 if not location.is_dir():
73 raise OSError(f"{location} should exist and be a directory.")
74 workflow_dir = location / f"cset-workflow-v{importlib.metadata.version('CSET')}"
76 # Write workflow content into workflow_dir.
77 workflow_files = importlib.resources.files(CSET.cset_workflow)
78 with importlib.resources.as_file(workflow_files) as w:
79 logger.info("Copying workflow files into place.")
80 try:
81 shutil.copytree(w, workflow_dir)
82 except FileExistsError as err:
83 raise FileExistsError(f"Refusing to overwrite {workflow_dir}") from err
85 # Make scripts executable.
86 logger.info("Changing mode of scripts to be executable.")
87 for dirpath, _, filenames in os.walk(workflow_dir):
88 for filename in filenames:
89 make_script_executable(Path(dirpath) / filename)
91 # Create link to conda environment.
92 conda_prefix = os.getenv("CONDA_PREFIX")
93 if conda_prefix is not None:
94 logger.info("Linking workflow conda environment to %s", conda_prefix)
95 (workflow_dir / "conda-environment").symlink_to(Path(conda_prefix).resolve())
96 else:
97 logger.warning("CONDA_PREFIX not defined. Skipping linking environment.")
99 print(f"Workflow written to {workflow_dir}")
100 return workflow_dir
103def list_refs(url: str) -> list[str]:
104 """List release refs for the repository.
106 This serves both as an access check and to find an appropriate ref.
107 """
108 # Disable interactively asking for authentication.
109 env = os.environ.copy()
110 env["GIT_TERMINAL_PROMPT"] = "false"
112 # List release references.
113 cmd = ("git", "ls-remote", "--quiet", "--refs", url, "releases/**", "v*")
114 logger.debug("Running %s", " ".join(cmd))
115 try:
116 p = subprocess.run(cmd, env=env, check=True, capture_output=True, text=True)
117 except subprocess.CalledProcessError as err:
118 raise ValueError(f"Cannot access Git repository at {url}") from err
120 # Reduce to just ref names.
121 release_refs = [
122 line.split(maxsplit=1)[-1]
123 .removeprefix("refs/heads/")
124 .removeprefix("refs/tags/")
125 for line in p.stdout.splitlines()
126 ]
127 return release_refs
130def clone_ref(ref: str, url: str, location: str):
131 """Clone the specified ref."""
132 # Disable interactively asking for authentication.
133 env = os.environ.copy()
134 env["GIT_TERMINAL_PROMPT"] = "false"
136 cmd = ("git", "clone", "--quiet", "--depth", "1", "--branch", ref, url, location)
137 logger.debug("Running %s", " ".join(cmd))
138 try:
139 subprocess.run(cmd, env=env, check=True, capture_output=True)
140 except subprocess.CalledProcessError as err:
141 raise ValueError(
142 f"Cannot access ref {ref} from Git repository at {url}"
143 ) from err
146def install_restricted_files(workflow_dir: Path, alternative_url: str | None = None):
147 """Install restricted site-specific restricted files into CSET workflow.
149 Parameters
150 ----------
151 workflow_dir: Path
152 The workflow directory into which the restricted files will be copied.
153 alternative_url: str, optional
154 Alternative Git URL to fetch the restricted files from. If omitted,
155 defaults to trying to clone first from 'localmirrors:', then from GitHub
156 via SSH and HTTPS.
158 Notes
159 -----
160 Requires Git to be installed to function.
161 """
162 # Basic check workflow_dir is correct.
163 if not (workflow_dir / "flow.cylc").is_file():
164 raise ValueError(f"{workflow_dir} should be a CSET workflow directory.")
166 # Determine target tag/branch from version.
167 version_tag = f"v{importlib.metadata.version('CSET')}"
168 logger.debug("Running for CSET %s", version_tag)
169 m = re.match(r"v\d+\.\d+", version_tag)
170 base_version = m.group(0) if m else version_tag
171 if m is None: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 logger.warning("Cannot determine major version from %s", version_tag)
173 release_branch = f"releases/{base_version}"
175 # Default URLs to try in order, or use alternative if provided.
176 urls = (
177 (
178 "localmirrors:MetOffice/CSET-restricted-files.git",
179 "git@github.com:MetOffice/CSET-restricted-files.git",
180 "https://github.com/MetOffice/CSET-restricted-files.git",
181 )
182 if alternative_url is None
183 else (alternative_url,)
184 )
186 # Find first working URL and list its refs.
187 for url in urls:
188 try:
189 refs = list_refs(url)
190 break
191 except ValueError:
192 continue
193 else:
194 url = (
195 alternative_url
196 if alternative_url is not None
197 else "https://github.com/MetOffice/CSET-restricted-files.git"
198 )
199 raise ValueError(
200 f"Could not read from restricted repository. Have you got access to {url}"
201 )
203 # Use most specific ref, falling back to main.
204 logger.debug("Release refs: %s", refs)
205 if version_tag in refs: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 ref = version_tag
207 elif release_branch in refs: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 ref = release_branch
209 else:
210 ref = "main"
211 logger.info("Fetching restricted files from ref %s of %s", ref, url)
213 # Checkout git repository to temporary location.
214 with tempfile.TemporaryDirectory() as tempdir:
215 # Clone restrict file repository.
216 logger.debug("Cloning to %s", tempdir)
217 clone_ref(ref, url, tempdir)
219 # Delete unwanted top-level README.
220 (Path(tempdir) / "README.md").unlink(missing_ok=True)
222 # Copy remaining files, skipping hidden files.
223 shutil.copytree(
224 tempdir,
225 workflow_dir,
226 ignore=shutil.ignore_patterns(".*"),
227 symlinks=True,
228 dirs_exist_ok=True,
229 )
230 print(f"Installed site-specific restricted files into {workflow_dir}.")