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 importlib.resources 

19import logging 

20import os 

21import shutil 

22import stat 

23from pathlib import Path 

24 

25import CSET.cset_workflow 

26 

27logger = logging.getLogger(__name__) 

28 

29 

30def make_script_executable(p: Path): 

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

32 if p.is_file(): 

33 try: 

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

35 shebang = fd.read(14) 

36 except PermissionError: 

37 # Skip files that can't be read. 

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

39 return 

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

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

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

43 mode = p.stat().st_mode 

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

45 mode |= stat.S_IXUSR 

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

47 if mode & stat.S_IRGRP: 

48 mode |= stat.S_IXGRP 

49 if mode & stat.S_IROTH: 

50 mode |= stat.S_IXOTH 

51 p.chmod(mode) 

52 

53 

54def install_workflow(location: Path): 

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

56 

57 Arguments 

58 --------- 

59 location: Path 

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

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

62 """ 

63 # Check location's parents exist. 

64 if not location.is_dir(): 

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

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

67 

68 # Write workflow content into workflow_dir. 

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

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

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

72 try: 

73 shutil.copytree(w, workflow_dir) 

74 except FileExistsError as err: 

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

76 

77 # Make scripts executable. 

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

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

80 for filename in filenames: 

81 make_script_executable(Path(dirpath) / filename) 

82 

83 # Create link to conda environment. 

84 conda_prefix = os.getenv("CONDA_PREFIX") 

85 if conda_prefix is not None: 

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

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

88 else: 

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

90 

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