Coverage for bugscpp/processor/core/command.py : 84%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Core module of defining commands.
4Inherit from the classes to add a new command:
5- SimpleCommand
6- ShellCommand
7- DockerCommand
9Note that the module name of a newly defined command will be the command name.
10For instance, if MyNewCommand is defined at my_command.py,
11MyNewCommand can be invoked from "d++ my_command" at command-line.
12"""
13import argparse
14import stat
15from abc import ABC, ABCMeta, abstractmethod
16from dataclasses import dataclass
17from pathlib import Path
18from typing import Dict, Generator, List, Optional, Tuple
20import taxonomy
21from config import config
22from errors import DppCommandListInternalError
23from message import message
24from processor.core.data import Worktree
25from processor.core.docker import Docker
26from processor.core.shell import Shell
29class CommandRegistryMeta(type):
30 """
31 Metaclass which auto registers class.
33 The module name will be command to invoke associated class.
34 All registered commands can be retrieved via 'RegisteredCommands'.
36 Methods
37 -------
38 get_commands : Dict[str, Command]
39 Returns a dictionary of registered commands.
41 See Also
42 --------
43 RegisteredCommands : Get a list of registered commands.
44 """
46 _commands: Dict[str, "Command"] = {}
48 def __new__(mcs, name, bases, attrs):
49 new_class = type.__new__(mcs, name, bases, attrs)
50 m = attrs["__module__"]
51 if m != __name__ and not getattr(new_class, "_ignore_registry", False):
52 CommandRegistryMeta._commands[m.split(".")[-1]] = new_class()
53 return new_class
55 @staticmethod
56 def get_commands() -> Dict[str, "Command"]:
57 return CommandRegistryMeta._commands
60class AbstractCommandRegistryMeta(CommandRegistryMeta, ABCMeta):
61 """
62 Class that combines CommandRegistryMeta with ABCMeta.
63 """
65 pass
68class BaseCommandRegistry(ABC, metaclass=AbstractCommandRegistryMeta):
69 """
70 Base class of CommandRegistry.
71 """
73 pass
76class Command(BaseCommandRegistry):
77 """
78 Abstract class to implement a command.
79 Inherit from this class to add a new command.
80 The name of the command will be identical to the module name where the class is defined.
81 If there are multiple classes inherited from this inside the module,
82 the last one will be applied.
83 """
85 @property
86 def group(self) -> str:
87 """Represent the group of this command. It is not meaningful yet."""
88 raise NotImplementedError
90 @property
91 def help(self) -> str:
92 """Description of this command."""
93 raise NotImplementedError
95 @abstractmethod
96 def __call__(self, argv: List[str]):
97 """The actual behavior of the command."""
98 raise NotImplementedError
101class RegisteredCommands:
102 """
103 Descriptor to access registered commands.
104 Pass a module name to retrieve an instance of associated class.
105 """
107 def __get__(self, instance, owner) -> Dict[str, Command]:
108 return CommandRegistryMeta.get_commands()
110 def __set__(self, instance, value):
111 raise DppCommandListInternalError()
114class SimpleCommand(Command):
115 """
116 Command that does not use docker.
117 Not fully implemented yet.
118 """
120 @property
121 def group(self) -> str:
122 return "v1"
124 @abstractmethod
125 def run(self, argv: List[str]) -> bool:
126 raise NotImplementedError
128 def __call__(self, argv: List[str]) -> bool:
129 return self.run(argv)
132@dataclass
133class ShellCommandArguments:
134 commands: List[str]
137class ShellCommand(Command):
138 """
139 Command that does not use docker but shell instead.
140 Not fully implemented yet.
141 """
143 def __init__(self):
144 pass
146 @property
147 def group(self) -> str:
148 return "v1"
150 def __call__(self, argv: List[str]):
151 shell_args = self.run(argv)
152 with Shell() as shell:
153 shell.send(shell_args.commands)
155 @abstractmethod
156 def run(self, argv: List[str]) -> ShellCommandArguments:
157 raise NotImplementedError
160class DockerCommandScript(metaclass=ABCMeta):
161 """
162 A list of "DockerCommand"s to be serially executed
163 """
165 def __init__(self, command_type: taxonomy.CommandType, command: List[str]):
166 self.type = command_type
167 self.lines: Tuple[str] = tuple(command)
169 @abstractmethod
170 def before(self):
171 """
172 Invoked before script is executed.
173 """
174 raise NotImplementedError
176 @abstractmethod
177 def step(self, linenr: int, line: str):
178 """
179 Invoked before each line is executed.
181 Parameters
182 ----------
183 linenr : int
184 The current line number.
185 line: str
186 The current line to execute.
187 """
188 pass
190 @abstractmethod
191 def output(self, linenr: Optional[int], exit_code: Optional[int], output: str):
192 """
193 Invoked after each line is executed.
194 linenr is None if all commands are executed as if it is a script.
196 Parameters
197 ----------
198 linenr : Optional[int]
199 Index of the commands which has been executed.
200 None if the type is CommandType.Script.
201 exit_code: Optional[int]
202 Exit code of the executed command.
203 None when stream is set to False.
204 output : str
205 Captured stdout of the executed command.
206 """
207 pass
209 @abstractmethod
210 def after(self):
211 """
212 Invoked after script is executed.
213 """
214 raise NotImplementedError
216 def __iter__(self):
217 return iter(self.lines)
219 def should_be_run_at_once(self) -> bool:
220 """
221 Returns
222 -------
223 bool
224 Return true if it should be written to a file and executed at once,
225 otherwise if it is sent to a container line by line.
226 """
227 return self.type == taxonomy.CommandType.Script
230class DockerCommandScriptGenerator(metaclass=ABCMeta):
231 """
232 Factory class of DockerCommandScript.
233 """
235 def __init__(self, metadata: taxonomy.MetaData, worktree: Worktree, stream: bool):
236 self._metadata = metadata
237 self._worktree = worktree
238 self._stream = stream
240 @property
241 def metadata(self):
242 """
243 Metadata information of the current script.
244 """
245 return self._metadata
247 @property
248 def worktree(self):
249 """Worktree information of the current script."""
250 return self._worktree
252 @property
253 def stream(self):
254 """True if command should be streamed, otherwise False."""
255 return self._stream
257 @abstractmethod
258 def create(self) -> Generator[DockerCommandScript, None, None]:
259 """Yield DockerCommandScript."""
260 raise NotImplementedError
263class DockerCommand(Command):
264 """
265 Executes each command of DockerCommandLine one by one inside a docker container.
266 Inherit from this class to implement a new command that should be run inside a docker.
268 Methods
269 -------
270 __call__ : None
271 Execute commands.
272 """
274 SCRIPT_NAME = "DPP_COMMAND_SCRIPT"
276 def __init__(self, parser: argparse.ArgumentParser):
277 self.parser = parser
278 self.environ: Dict[str, str] = {}
280 @property
281 def group(self) -> str:
282 return "v1"
284 def __call__(self, argv: List[str]):
285 """
286 Execute commands inside a container.
288 Parameters
289 ----------
290 argv : List[str]
291 Command line argument vector.
293 Returns
294 -------
295 None
296 """
298 def parse_exec_result(ec, output, line_number: Optional[int] = None) -> None:
299 # Depending on 'stream' value, return value is a bit different.
300 # 'exit_code' is None when 'stream' is True.
301 # https://docker-py.readthedocs.io/en/stable/containers.html
302 if ec is None:
303 for stream_line in output:
304 message.stdout_stream(stream_line.decode("utf-8", errors="ignore"))
305 else:
306 script.output(line_number, ec, output.decode("utf-8", errors="ignore"))
308 args = self.parser.parse_args(argv)
310 if args.jobs <= 0:
311 raise ValueError("jobs must be greater than 0")
312 else:
313 config.DPP_PARALLEL_BUILD = str(args.jobs)
315 self.environ = args.env
317 script_generator = self.create_script_generator(args)
318 worktree = script_generator.worktree
319 stream = script_generator.stream
320 rebuild_image = True if args.rebuild_image else False
321 user = "" # root
322 uid = args.uid if hasattr(args, "uid") and args.uid is not None else None
324 self.setup(script_generator)
325 with Docker(
326 script_generator.metadata.dockerfile,
327 script_generator.worktree,
328 self.environ,
329 rebuild_image,
330 user,
331 uid,
332 ) as docker:
333 for script in script_generator.create():
334 script.before()
335 if script.should_be_run_at_once():
336 file = Path(f"{worktree.host}/{DockerCommand.SCRIPT_NAME}")
337 with open(file, "w+") as fp:
338 fp.write("\n".join(script))
339 file.chmod(
340 file.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
341 )
342 exit_code, output_stream = docker.send(
343 f"{worktree.container}/{DockerCommand.SCRIPT_NAME}", stream
344 )
345 parse_exec_result(exit_code, output_stream)
346 else:
347 for linenr, line in enumerate(script, start=1):
348 script.step(linenr, line)
349 exit_code, output_stream = docker.send(line, stream)
350 parse_exec_result(exit_code, output_stream, linenr)
351 script.after()
352 self.teardown(script_generator)
354 @abstractmethod
355 def create_script_generator(
356 self, args: argparse.Namespace
357 ) -> DockerCommandScriptGenerator:
358 """
359 Return DockerExecInfo which has information of a command list to run inside docker container.
361 Parameters
362 ----------
363 args : argparse.Namespace
364 argparse.Namespace instance.
366 Returns
367 -------
368 DockerCommandScriptGenerator
369 An instance of generator class which is used to create DockerCommandScript.
370 """
371 raise NotImplementedError
373 @abstractmethod
374 def setup(self, generator: DockerCommandScriptGenerator):
375 """
376 Invoked before container is created.
378 Parameters
379 ----------
380 generator : DockerCommandScriptGenerator
381 The current instance being used.
382 """
383 raise NotImplementedError
385 @abstractmethod
386 def teardown(self, generator: DockerCommandScriptGenerator):
387 """
388 Invoked after container is destroyed.
390 Parameters
391 ----------
392 generator : DockerCommandScriptGenerator
393 The current instance being used.
394 """
395 raise NotImplementedError