Coverage for src/CSET/extract_workflow.py: 55%

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. 

14 

15"""Extract the CSET cylc workflow for use.""" 

16 

17import importlib.metadata 

18import importlib.resources 

19import logging 

20import os 

21import re 

22import shutil 

23import stat 

24import subprocess 

25import tempfile 

26from pathlib import Path 

27 

28import CSET.cset_workflow 

29 

30logger = logging.getLogger(__name__) 

31 

32 

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) 

55 

56 

57def install_workflow(location: Path) -> Path: 

58 """Install the workflow's files and link the conda environment. 

59 

60 Arguments 

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. 

65 """ 

66 # Check location's parents exist. 

67 if not location.is_dir(): 

68 raise OSError(f"{location} should exist and be a directory.") 

69 workflow_dir = location / f"cset-workflow-v{importlib.metadata.version('CSET')}" 

70 

71 # Write workflow content into workflow_dir. 

72 workflow_files = importlib.resources.files(CSET.cset_workflow) 

73 with importlib.resources.as_file(workflow_files) as w: 

74 logger.info("Copying workflow files into place.") 

75 try: 

76 shutil.copytree(w, workflow_dir) 

77 except FileExistsError as err: 

78 raise FileExistsError(f"Refusing to overwrite {workflow_dir}") from err 

79 

80 # Make scripts executable. 

81 logger.info("Changing mode of scripts to be executable.") 

82 for dirpath, _, filenames in os.walk(workflow_dir): 

83 for filename in filenames: 

84 make_script_executable(Path(dirpath) / filename) 

85 

86 # Create link to conda environment. 

87 conda_prefix = os.getenv("CONDA_PREFIX") 

88 if conda_prefix is not None: 

89 logger.info("Linking workflow conda environment to %s", conda_prefix) 

90 (workflow_dir / "conda-environment").symlink_to(Path(conda_prefix).resolve()) 

91 else: 

92 logger.warning("CONDA_PREFIX not defined. Skipping linking environment.") 

93 

94 print(f"Workflow written to {workflow_dir}") 

95 return workflow_dir 

96 

97 

98def list_refs(url: str) -> list[str]: 

99 """List release refs for the repository. 

100 

101 This serves both as an access check and to find an appropriate ref. 

102 """ 

103 # Disable interactively asking for authentication. 

104 env = os.environ.copy() 

105 env["GIT_TERMINAL_PROMPT"] = "false" 

106 

107 cmd = ("git", "ls-remote", "--quiet", url, "releases/**", "v*") 

108 logger.debug("Running %s", " ".join(cmd)) 

109 try: 

110 p = subprocess.run(cmd, env=env, check=True, capture_output=True, text=True) 

111 except subprocess.CalledProcessError as err: 

112 raise ValueError(f"Cannot access Git repository at {url}") from err 

113 

114 return [line.split(maxsplit=1)[-1] for line in p.stdout.splitlines()] 

115 

116 

117def clone_ref(ref: str, url: str, location: str): 

118 """Clone the specified ref.""" 

119 # Disable interactively asking for authentication. 

120 env = os.environ.copy() 

121 env["GIT_TERMINAL_PROMPT"] = "false" 

122 

123 cmd = ("git", "clone", "--quiet", "--depth", "1", "--branch", ref, url, location) 

124 logger.debug("Running %s", " ".join(cmd)) 

125 try: 

126 subprocess.run(cmd, env=env, check=True, capture_output=True) 

127 except subprocess.CalledProcessError as err: 

128 raise ValueError(f"Cannot access Git repository at {url}") from err 

129 

130 

131def install_restricted_files(workflow_dir: Path, alternative_url: str | None = None): 

132 """Install restricted site-specific restricted files into CSET workflow. 

133 

134 Parameters 

135 ---------- 

136 workflow_dir: str 

137 The workflow directory into which the restricted files will be copied. 

138 alternative_url: str, optional 

139 Alternative Git URL to fetch the restricted files from. If omitted, 

140 defaults to trying to clone first from 'localmirrors:', then from GitHub 

141 via SSH and HTTPS. 

142 

143 Notes 

144 ----- 

145 Requires Git to be installed to function. 

146 """ 

147 # Basic check workflow_dir is correct. 

148 if not (workflow_dir / "flow.cylc").is_file(): 

149 raise ValueError("workflow_dir should be a CSET workflow directory.") 

150 

151 # Determine target tag/branch from version. 

152 version_tag = f"v{importlib.metadata.version('CSET')}" 

153 logger.debug("Running for CSET %s", version_tag) 

154 m = re.match(r"v\d+\.\d+", version_tag) 

155 if m is None: 

156 logger.error("Unknown CSET version: %s", version_tag) 

157 m = [version_tag] 

158 release_branch = f"releases/{m[0]}" 

159 

160 # Default URLs to try in order, or use alternative if provided. 

161 urls = ( 

162 ( 

163 "localmirrors:MetOffice/CSET-restricted-files.git", 

164 "git@github.com:MetOffice/CSET-restricted-files.git", 

165 "https://github.com/MetOffice/CSET-restricted-files.git", 

166 ) 

167 if alternative_url is None 

168 else (alternative_url,) 

169 ) 

170 

171 # Find first working URL and list its refs. 

172 for url in urls: 

173 try: 

174 refs = list_refs(url) 

175 break 

176 except ValueError: 

177 continue 

178 else: 

179 raise ValueError( 

180 "Could not read from restricted repository. Have you got access?" 

181 ) 

182 

183 # Use most specific ref, falling back to main. 

184 logger.debug("Release refs: %s", refs) 

185 if version_tag in refs: 

186 ref = version_tag 

187 elif release_branch in refs: 

188 ref = release_branch 

189 else: 

190 ref = "main" 

191 logger.info("Fetching restricted files from ref %s of %s", ref, url) 

192 

193 # Checkout git repository to temporary location. 

194 with tempfile.TemporaryDirectory() as tempdir: 

195 # Clone restrict file repository. 

196 logger.debug("Cloning to %s", tempdir) 

197 clone_ref(ref, url, tempdir) 

198 

199 # Delete unwanted top-level README. 

200 os.unlink(f"{tempdir}/README.md") 

201 

202 # Copy remaining files, skipping hidden files. 

203 shutil.copytree( 

204 tempdir, 

205 workflow_dir, 

206 ignore=shutil.ignore_patterns(".*"), 

207 symlinks=True, 

208 dirs_exist_ok=True, 

209 ) 

210 logger.info("Installation complete.")