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

1# © Crown copyright, Met Office (2022-2025) 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 logging 

19import os 

20import shutil 

21import stat 

22import sys 

23from pathlib import Path 

24 

25import CSET.cset_workflow 

26 

27logger = logging.getLogger(__name__) 

28 

29# The as_file interface only supports directories from python 3.12. 

30if sys.version_info.minor < 12: 

31 import importlib_resources 

32else: 

33 import importlib.resources as importlib_resources 

34 

35 

36def make_script_executable(p: Path): 

37 """Make a script file (starting with a shebang) executable.""" 

38 if p.is_file(): 

39 try: 

40 with open(p, "rb") as fd: 

41 shebang = fd.read(14) 

42 except PermissionError: 

43 # Skip files that can't be read. 

44 logger.debug("Unreadable file: %s", p) 

45 return 

46 # Assume the first 14 bytes of a script are #!/usr/bin/env 

47 if shebang == b"#!/usr/bin/env": 

48 logger.debug("Changing file mode to executable: %s", p) 

49 mode = p.stat().st_mode 

50 # User must be able to read if we read the file. 

51 mode |= stat.S_IXUSR 

52 # Make executable by group and/or others if they can read. 

53 if mode & stat.S_IRGRP: 

54 mode |= stat.S_IXGRP 

55 if mode & stat.S_IROTH: 

56 mode |= stat.S_IXOTH 

57 p.chmod(mode) 

58 

59 

60def install_workflow(location: Path): 

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

62 

63 Arguments 

64 --------- 

65 location: Path 

66 A directory where the workflow files are to be installed to. A 

67 sub-directory named "cset-workflow-vX.Y.Z" will be created under here. 

68 """ 

69 # Check location's parents exist. 

70 if not location.is_dir(): 

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

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

73 

74 # Write workflow content into workflow_dir. 

75 workflow_files = importlib_resources.files(CSET.cset_workflow) 

76 with importlib_resources.as_file(workflow_files) as w: 

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

78 try: 

79 shutil.copytree(w, workflow_dir) 

80 except FileExistsError as err: 

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

82 

83 # Make scripts executable. 

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

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

86 for filename in filenames: 

87 make_script_executable(Path(dirpath) / filename) 

88 

89 # Create link to conda environment. 

90 conda_prefix = os.getenv("CONDA_PREFIX") 

91 if conda_prefix is not None: 

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

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

94 else: 

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

96 

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