Coverage for structured_tutorials / runners / vagrant.py: 100%

1# Copyright (c) 2026 Mathias Ertl 

2# Licensed under the MIT License. See LICENSE file for details. 

3 

4"""Class for running the VagrantRunner.""" 

5 

6import io 

7import logging 

8import os.path 

9import shlex 

10import shutil 

11import subprocess 

12import tempfile 

13from collections.abc import Sequence 

14from pathlib import Path 

15from subprocess import CompletedProcess 

16from typing import Any 

17 

18from pydantic import BaseModel, Field 

19 

20from structured_tutorials.errors import ( 

21 ConfigurationException, 

22 RequiredExecutableNotFoundError, 

23 RunTutorialException, 

24) 

25from structured_tutorials.models.base import CommandType 

26from structured_tutorials.models.types import RelativePath 

27from structured_tutorials.runners.base import RunnerBase 

28 

29log = logging.getLogger(__name__) 

30command_logger = logging.getLogger("command") 

31 

32PATH_ENVIRONMENT_VARIABLES = ( 

33 "VAGRANT_CWD", 

34 "SSL_CERT_FILE", 

35 "VAGRANT_ALIAS_FILE", 

36 "VAGRANT_DOTFILE_PATH", 

37 "VAGRANT_HOME", 

38) 

39 

40 

41class PrepareBoxOptions(BaseModel): 

42 """Options for preparing a box.""" 

43 

44 name: str 

45 path: RelativePath 

46 

47 

48class VagrantOptions(BaseModel): 

49 """Options for running vagrant.""" 

50 

51 environment: dict[str, Any] = Field(default_factory=dict) 

52 cwds: dict[str, str] = Field(default_factory=dict) 

53 prepare_box: PrepareBoxOptions | None = None 

54 

55 

56class VagrantRunner(RunnerBase): 

57 """Runner to run Vagrant tutorials.""" 

58 

59 def __init__(self, *args: Any, **kwargs: Any) -> None: 

60 super().__init__(*args, **kwargs) 

61 

62 # Check if vagrant is installed 

63 if not shutil.which("vagrant", path=self.environment.get("PATH")): 

64 raise RequiredExecutableNotFoundError("vagrant: Executable not found.") 

65 

66 self.config = VagrantOptions.model_validate(self.tutorial.configuration.run.runner.options) 

67 self.config.environment.setdefault("VAGRANT_CWD", str(self.tutorial.root)) 

68 self.config.environment.setdefault("PATH", os.environ["PATH"]) 

69 self.cwds: dict[str, str] = self.config.cwds.copy() 

70 

71 # convert relative paths to absolute paths for known path variables: 

72 for key, value in self.config.environment.items(): 

73 if key in PATH_ENVIRONMENT_VARIABLES: 

74 value_path = Path(value) 

75 if not value_path.is_absolute(): 

76 value_path = self.tutorial.root / value_path 

77 self.config.environment[key] = str(value_path.resolve()) 

78 

79 def vagrant( 

80 self, args: Sequence[str], env: dict[str, str] | None = None, check: bool = True 

81 ) -> subprocess.CompletedProcess[bytes]: 

82 """Run a vagrant command. 

83 

84 NOTE: This method is intended to be run by boilerplate parts (e.g. preparation, cleanup) of running 

85 the tutorial, and *not* for commands from the tutorial. 

86 """ 

87 command = ["vagrant", *args] 

88 joined_command = shlex.join(command) 

89 command_logger.info(joined_command) 

90 env = {**os.environ, **self.config.environment, **(env or {})} 

91 proc = subprocess.run(command, env=env, capture_output=not self.show_command_output) 

92 if check and proc.returncode != 0: 

93 raise RunTutorialException(f"{joined_command} exited with status code {proc.returncode}.") 

94 return proc 

95 

96 def chdir(self, path: str, options: dict[str, Any]) -> None: 

97 machine = options.get("machine", "default") 

98 self.cwds[machine] = path 

99 

100 def run_shell_command( 

101 self, 

102 command: CommandType, 

103 show_output: bool, 

104 capture_output: bool = False, 

105 stdin: int | io.BufferedReader | None = None, 

106 input: bytes | None = None, 

107 environment: dict[str, Any] | None = None, 

108 clear_environment: bool = False, 

109 options: dict[str, Any] | None = None, 

110 ) -> CompletedProcess[bytes]: 

111 if options is None: # pragma: no cover # doesn't happen in real time 

112 options = {} 

113 

114 machine = options.get("machine", "default") 

115 command = self.render_command(command) 

116 if isinstance(command, tuple): 

117 command = shlex.join(command) 

118 

119 cmd = command 

120 cwd = self.cwds.get(machine) 

121 if cwd: 

122 cmd = f"{shlex.join(['cd', cwd])} && {command}" 

123 

124 vagrant_command = ["vagrant", "ssh", machine, "-c", cmd] 

125 

126 proc = super().run_shell_command( 

127 tuple(vagrant_command), 

128 show_output=show_output, 

129 capture_output=capture_output, 

130 stdin=stdin, 

131 input=input, 

132 environment=self.config.environment, 

133 clear_environment=True, 

134 ) 

135 return proc 

136 

137 def prepare_vagrantfile(self, path: Path) -> None: 

138 """Render a Vagrantfile in the given directory.""" 

139 if not path.exists(): 

140 raise ConfigurationException(f"{path}: Directory not found.") 

141 

142 vagrantfile_path = path / "Vagrantfile" 

143 template_path = vagrantfile_path.with_suffix(".jinja") 

144 if os.path.exists(template_path): 

145 with open(template_path) as template_stream: 

146 template = template_stream.read() 

147 

148 contents = self.render(template) 

149 with open(vagrantfile_path, "w") as vagrant_file_stream: 

150 vagrant_file_stream.write(contents) 

151 elif not os.path.exists(vagrantfile_path): 

152 raise ConfigurationException(f"Vagrantfile not found: {vagrantfile_path}") 

153 

154 def prepare_box(self, config: PrepareBoxOptions) -> None: 

155 """Prepare the box before running the tutorial, if requested.""" 

156 vagrant_cwd = self.tutorial.root / config.path 

157 self.context["box"] = config.name # update context with name of box 

158 

159 # Render the Vagrantfile to prepare the box 

160 self.prepare_vagrantfile(vagrant_cwd) 

161 

162 result = subprocess.run( 

163 ["vagrant", "box", "list", "--machine-readable"], capture_output=True, text=True, check=True 

164 ) 

165 if any( 

166 row.split(",")[2] == "box-name" and row.split(",")[3] == config.name 

167 for row in result.stdout.splitlines() 

168 ): 

169 log.info("Box '%s' already exists, skip creation.", config.name) 

170 return 

171 

172 log.info("Preparing Vagrant box in %s", vagrant_cwd) 

173 env = {"VAGRANT_CWD": str(vagrant_cwd)} 

174 try: 

175 self.vagrant(["up"], env=env) 

176 self.vagrant(["package", "--output", f"{config.name}.box"], env=env) 

177 finally: 

178 self.vagrant(["destroy", "-f"], env=env) 

179 

180 self.vagrant(["box", "add", config.name, f"{config.name}.box"], env=env) 

181 

182 def update_environment_variable(self, key: str, value: str, options: dict[str, str]) -> None: 

183 self.run_shell_command( 

184 f'echo export {key}="{value}" >> ~/.profile', show_output=False, options=options 

185 ) 

186 

187 def prepare_tutorial(self) -> None: 

188 """Prepare tutorial for running.""" 

189 if prepare_box_config := self.config.prepare_box: 

190 self.prepare_box(prepare_box_config) 

191 

192 # Render the main Vagrantfile for the tutorial 

193 self.prepare_vagrantfile(Path(self.config.environment["VAGRANT_CWD"])) 

194 self.vagrant(["up"]) 

195 

196 def cleanup_tutorial(self) -> None: 

197 """Cleanup tutorial after running.""" 

198 self.vagrant(["destroy", "-f"]) 

199 

200 def copy_file(self, source: Path, destination: Path, options: dict[str, Any]) -> None: 

201 machine = options.get("machine", "default") 

202 self.vagrant(["upload", str(source), str(destination), machine]) 

203 

204 def write_file_from_string(self, contents: str, destination: Path, options: dict[str, Any]) -> None: 

205 # Only way to write a file from string is to write it to a temporary file on the host system 

206 with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: 

207 temp_file.write(contents) 

208 temp_file.flush() 

209 self.copy_file(Path(temp_file.name), destination, options)