Coverage for src/jobs/runner.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from argparse import Namespace 

5from os import chmod, environ, stat, system 

6from pathlib import Path, PurePosixPath 

7from re import sub as regex_sub 

8from signal import getsignal, SIGINT, signal, SIGTERM 

9from stat import S_IRGRP, S_IROTH, S_IXGRP, S_IXOTH, S_IXUSR 

10from subprocess import run 

11from sys import stdout 

12from time import sleep 

13from types import FrameType 

14from typing import Dict, List, Optional 

15 

16# Components 

17from ..engines.engine import Engine, privileged_default 

18from ..models.flags import Flags 

19from ..models.images import Entrypoint 

20from ..models.jobs import Job 

21from ..package.bundle import Bundle 

22from ..prints.histories import PipelineHistory 

23from ..system.git import Git 

24from ..system.platform import Platform 

25from ..system.xauth import Xauth 

26from ..types.environment import Environment 

27from ..types.paths import Paths 

28from ..types.strings import Strings 

29from ..types.volumes import Volumes 

30from .outputs import Outputs 

31from .scripts import ScriptsFile 

32 

33# Runner class, pylint: disable=too-few-public-methods 

34class Runner: 

35 

36 # Constants 

37 __MARKER_DEBUG: str = '__GITLAB_CI_LOCAL_DEBUG__' 

38 __MARKER_ENDING: str = '__GITLAB_CI_LOCAL_ENDING__' 

39 __MARKER_RESULT: str = '__GITLAB_CI_LOCAL_RESULT__' 

40 

41 # Members 

42 __engines: Dict[str, Optional[Engine]] = {} 

43 __interrupted: bool 

44 __options: Namespace 

45 

46 # Constructor 

47 def __init__( 

48 self, 

49 options: Namespace, 

50 ) -> None: 

51 

52 # Prepare flags 

53 self.__interrupted = False 

54 

55 # Prepare options 

56 self.__options = options 

57 

58 # Run container, pylint: disable=too-many-arguments,too-many-branches,too-many-locals,too-many-positional-arguments,too-many-statements 

59 def __run_container( 

60 self, 

61 engine: Optional[Engine], 

62 variables: Dict[str, str], 

63 path_parent: str, 

64 target_parent: str, 

65 image: str, 

66 job: Job, 

67 script_file: ScriptsFile, 

68 entrypoint: Entrypoint, 

69 network: str, 

70 privileged: bool, 

71 target_workdir: str, 

72 last_result: bool, 

73 result: bool, 

74 ) -> bool: 

75 

76 # Validate engine 

77 assert engine is not None 

78 

79 # Configure engine variables 

80 variables[Bundle.ENV_ENGINE_NAME] = engine.name 

81 

82 # Prepare volumes mounts 

83 volumes = Volumes() 

84 

85 # Mount repository folder 

86 volumes.add( 

87 path_parent, 

88 target_parent, 

89 'rw', 

90 True, 

91 ) 

92 

93 # Extend mounts 

94 if self.__options.volume: 

95 for volume in self.__options.volume: 

96 

97 # Handle .local volumes 

98 cwd = Path('.') 

99 volume_local = False 

100 if volume.startswith(Volumes.LOCAL_FLAG): 

101 cwd = self.__options.path 

102 volume_local = True 

103 volume = volume[len(Volumes.LOCAL_FLAG):] 

104 

105 # Parse volume fields 

106 volume_nodes = Volumes.parse(volume) 

107 

108 # Parse HOST:TARGET:MODE, pylint: disable=line-too-long 

109 if len(volume_nodes) == 3: 

110 volume_host = Paths.resolve(cwd / Paths.expand(volume_nodes[0])) 

111 volume_target = Paths.expand(volume_nodes[1], home=False) 

112 volume_mode = volume_nodes[2] 

113 

114 # Parse HOST:TARGET, pylint: disable=line-too-long 

115 elif len(volume_nodes) == 2: 

116 volume_host = Paths.resolve(cwd / Paths.expand(volume_nodes[0])) 

117 volume_target = Paths.expand(volume_nodes[1], home=False) 

118 volume_mode = 'rw' 

119 

120 # Parse VOLUME, pylint: disable=line-too-long 

121 else: 

122 volume_host = Paths.resolve(cwd / Paths.expand(volume_nodes[0])) 

123 volume_target = Paths.resolve(cwd / Paths.expand(volume_nodes[0])) 

124 volume_mode = 'rw' 

125 

126 # Append volume mounts 

127 volumes.add( 

128 volume_host, 

129 volume_target, 

130 volume_mode, 

131 not volume_local, 

132 ) 

133 

134 # Image validation 

135 if not image: # pragma: no cover 

136 raise ValueError(f"Missing image for \"{job.stage} / {job.name}\"") 

137 engine.get(image) 

138 

139 # Display option 

140 if Flags.check_bool( 

141 option=self.__options.display, 

142 flag=Bundle.FLAG_DISPLAY, 

143 variables=job.variables, 

144 default=False, 

145 ): 

146 

147 # Bind display environment 

148 if Platform.ENV_DISPLAY in environ: # pragma: no cover 

149 variables[Platform.ENV_DISPLAY] = environ[Platform.ENV_DISPLAY] 

150 if Platform.ENV_XAUTHORITY in environ: # pragma: no cover 

151 variables[Platform.ENV_XAUTHORITY] = environ[Platform.ENV_XAUTHORITY] 

152 

153 # Bind display socket 

154 for display_socket in Platform.display(): # pragma: no cover 

155 volumes.add( 

156 display_socket, 

157 display_socket, 

158 'rw', 

159 False, 

160 ) 

161 

162 # Bind display Xauthority 

163 display_xauth_magic: str = Xauth.magic() 

164 if display_xauth_magic: # pragma: no cover 

165 variables[Bundle.ENV_DISPLAY_XAUTH_MAGIC_ADD] = display_xauth_magic 

166 

167 # SSH option 

168 ssh: str = Flags.check_strbool( 

169 option=self.__options.ssh, 

170 flag=Bundle.FLAG_SSH, 

171 variables=job.variables, 

172 default='', 

173 const=Bundle.ARGUMENT_SSH_USER_DEFAULT, 

174 ) 

175 if ssh: 

176 

177 # Bind SSH credentials 

178 volumes.add( 

179 Paths.expand('~/.ssh'), 

180 Paths.home(ssh) + '/.ssh', 

181 'ro', 

182 False, 

183 ) 

184 

185 # Bind SSH agent 

186 if Platform.ENV_SSH_AUTH_SOCK in environ: 

187 if environ[Platform.ENV_SSH_AUTH_SOCK]: 

188 volumes.add( 

189 Paths.expand(environ[Platform.ENV_SSH_AUTH_SOCK]), 

190 Paths.expand(environ[Platform.ENV_SSH_AUTH_SOCK]), 

191 'ro', 

192 True, 

193 ) 

194 variables[Platform.ENV_SSH_AUTH_SOCK] = environ[ 

195 Platform.ENV_SSH_AUTH_SOCK] 

196 

197 # Launch container 

198 engine.run( 

199 image=image, 

200 commands=[script_file.target()], 

201 entrypoint=entrypoint, 

202 variables=variables, 

203 network=network, 

204 option_privileged=privileged, 

205 option_sockets=job.options.sockets, 

206 services=bool(job.services), 

207 volumes=volumes, 

208 directory=target_workdir, 

209 temp_folder=script_file.folder, 

210 ) 

211 

212 # Create interruption handler 

213 def interrupt_handler( 

214 __signalnum: Optional[int] = None, 

215 __frame: Optional[FrameType] = None, 

216 ) -> None: 

217 self.__interrupted = True 

218 Outputs.interruption() 

219 assert engine is not None 

220 engine.stop(0) 

221 

222 # Register interruption handler 

223 handler_int_original = getsignal(SIGINT) 

224 handler_term_original = getsignal(SIGTERM) 

225 signal(SIGINT, interrupt_handler) 

226 signal(SIGTERM, interrupt_handler) 

227 

228 # Execution wrapper 

229 marker_result: Optional[bool] = None 

230 success = False 

231 

232 # Show container logs 

233 for line in engine.logs(): 

234 if isinstance(line, bytes): 

235 line_decoded = line.decode() 

236 if self.__MARKER_DEBUG in line_decoded: 

237 break # pragma: no cover 

238 if self.__MARKER_RESULT in line_decoded: 

239 marker_result = int( 

240 line_decoded.split(f'{self.__MARKER_RESULT}:', 1)[1]) == 0 

241 continue 

242 if self.__MARKER_ENDING in line_decoded: 

243 break 

244 stdout.buffer.write(line) 

245 Platform.flush() 

246 

247 # Runner bash or debug mode 

248 if not self.__interrupted and (Flags.check_bool( 

249 option=self.__options.bash, 

250 flag=Bundle.FLAG_BASH, 

251 variables=job.variables, 

252 default=False, 

253 ) or Flags.check_bool( 

254 option=self.__options.debug, 

255 flag=Bundle.FLAG_DEBUG, 

256 variables=job.variables, 

257 default=False, 

258 )): 

259 

260 # Select shell 

261 shell: str = '' 

262 if Flags.has( 

263 option=self.__options.shell, 

264 flag=Bundle.FLAG_SHELL, 

265 variables=job.variables, 

266 ): 

267 shell = Flags.check_str( 

268 option=self.__options.shell, 

269 flag=Bundle.FLAG_SHELL, 

270 variables=job.variables, 

271 default='sh', 

272 ) 

273 elif engine.supports('bash'): 

274 shell = 'bash' 

275 else: 

276 shell = 'sh' 

277 

278 # Select console 

279 console: bool = Platform.IS_TTY_STDIN and not Flags.check_bool( 

280 option=self.__options.no_console, 

281 flag=Bundle.FLAG_NO_CONSOLE, 

282 variables=job.variables, 

283 default=False, 

284 ) 

285 

286 # Acquire container informations 

287 container_exec = engine.cmd_exec() 

288 container_name = engine.container 

289 

290 # Debugging informations 

291 Outputs.debugging( 

292 container_exec, 

293 container_name, 

294 shell, 

295 console=console, 

296 ) 

297 

298 # Launch container console 

299 if console: 

300 run( 

301 f'{container_exec} {container_name} {shell}', 

302 check=False, 

303 shell=True, 

304 ) 

305 self.__interrupted = True 

306 engine.stop(0) 

307 

308 # Check container status 

309 success = engine.wait() 

310 

311 # Stop container 

312 engine.stop(0) 

313 sleep(0.1) 

314 

315 # Remove container 

316 engine.remove() 

317 

318 # Unregister interruption handler 

319 signal(SIGINT, handler_int_original) 

320 signal(SIGTERM, handler_term_original) 

321 

322 # Result evaluation 

323 if job.when in [ 

324 'on_failure', 

325 'always', 

326 ]: 

327 result = last_result 

328 elif success: 

329 result = True 

330 elif self.__interrupted and marker_result is not None: 

331 result = marker_result 

332 return result 

333 

334 # Run native, pylint: disable=too-many-arguments,too-many-positional-arguments 

335 @staticmethod 

336 def __run_native( 

337 variables: Dict[str, str], 

338 entrypoint: Entrypoint, 

339 script_file: ScriptsFile, 

340 job: Job, 

341 last_result: bool, 

342 result: bool, 

343 ) -> bool: 

344 

345 # Configure host variables 

346 variables[Bundle.ENV_HOST] = 'true' 

347 

348 # Prepare environment 

349 _environ = dict(environ) 

350 environ.update(variables) 

351 

352 # Detect nested execution 

353 if ScriptsFile.SHEBANG_MARKER_BASH in environ: 

354 del environ[ScriptsFile.SHEBANG_MARKER_BASH] 

355 

356 # Native execution 

357 scripts: List[str] = [] 

358 if entrypoint: 

359 scripts += entrypoint 

360 if not scripts: 

361 scripts = ['sh'] 

362 scripts += [f'"{script_file.name}"'] 

363 success = system(' '.join(scripts)) == 0 

364 

365 # Result evaluation 

366 if job.when in [ 

367 'on_failure', 

368 'always', 

369 ]: 

370 result = last_result 

371 elif success: 

372 result = True 

373 

374 # Restore environment 

375 environ.clear() 

376 environ.update(_environ) 

377 

378 # Result 

379 return result 

380 

381 # Run, pylint: disable=too-many-arguments,too-many-branches,too-many-locals,too-many-statements 

382 def run( 

383 self, 

384 job: Job, 

385 last_result: bool, 

386 pipeline_history: PipelineHistory, 

387 ) -> bool: 

388 

389 # Variables 

390 engine: Optional[Engine] = None 

391 engines_names: str 

392 engine_type: str = '' 

393 host: bool = False 

394 network: str 

395 privileged: bool 

396 quiet: bool 

397 random_paths: bool = False 

398 real_paths: bool = False 

399 result: bool = False 

400 script_file: ScriptsFile 

401 

402 # Prepare history 

403 job_history = pipeline_history.add( 

404 job.stage, 

405 job.name, 

406 ) 

407 

408 # Validate paths 

409 if Flags.check_bool( 

410 option=self.__options.random_paths, 

411 flag=Bundle.FLAG_RANDOM_PATHS, 

412 variables=job.variables, 

413 default=False, 

414 ) and Flags.check_bool( 

415 option=self.__options.real_paths, 

416 flag=Bundle.FLAG_REAL_PATHS, 

417 variables=job.variables, 

418 default=False, 

419 ): 

420 Outputs.warning( 

421 'The real paths feature is in conflict with the random paths feature...') 

422 random_paths = True 

423 

424 # Prepare random paths 

425 elif Flags.check_bool( 

426 option=self.__options.random_paths, 

427 flag=Bundle.FLAG_RANDOM_PATHS, 

428 variables=job.variables, 

429 default=False, 

430 ): 

431 random_paths = True 

432 

433 # Prepare real paths 

434 elif Flags.check_bool( 

435 option=self.__options.real_paths, 

436 flag=Bundle.FLAG_REAL_PATHS, 

437 variables=job.variables, 

438 default=False, 

439 ): 

440 if Platform.IS_LINUX or Platform.IS_MAC_OS: 

441 real_paths = True 

442 

443 # Unavailable feature 

444 else: # pragma: no cover 

445 Outputs.warning('The real paths feature is not available...') 

446 

447 # Initial job details 

448 job_details_list: List[str] = [] 

449 job_details_string: str = '' 

450 

451 # Prepare when details 

452 if job.when not in ['on_success']: 

453 job_details_list += [f"when: {job.when}"] 

454 

455 # Prepare allow_failure details 

456 if job.allow_failure: 

457 job_details_list += ['failure allowed'] 

458 job_history.failure_allowed = True 

459 

460 # Prepare job details 

461 if job_details_list: 

462 job_details_string = f" ({', '.join(job_details_list)})" 

463 

464 # Update job details 

465 job_history.details = job_details_string 

466 

467 # Filter when 

468 if last_result and job.when not in [ 

469 'on_success', 

470 'manual', 

471 'always', 

472 ]: 

473 return last_result 

474 if not last_result and job.when not in [ 

475 'on_failure', 

476 'always', 

477 ]: 

478 return last_result 

479 

480 # Prepare image 

481 image: str = job.image 

482 

483 # Prepare local runner 

484 if self.__options.host or job.options.host: 

485 image = 'local' 

486 host = True 

487 

488 # Prepare quiet flag 

489 quiet = Flags.check_bool( 

490 option=self.__options.quiet, 

491 flag=Bundle.FLAG_QUIET, 

492 variables=job.variables, 

493 default=False, 

494 ) 

495 

496 # Drop quiet flag 

497 if not quiet and pipeline_history.jobs_quiet: 

498 pipeline_history.jobs_quiet = False 

499 

500 # Prepare engine execution 

501 if not host: 

502 engines_names = Flags.check_str( 

503 option=None, 

504 flag=Bundle.FLAG_ENGINE, 

505 variables=job.variables, 

506 default=self.__options.engine, 

507 ) 

508 if engines_names not in self.__engines or self.__engines[ 

509 engines_names] is None: 

510 self.__engines[engines_names] = Engine(engines_names) 

511 engine = self.__engines[engines_names] 

512 assert engine is not None 

513 engine_type = engine.name 

514 

515 # Prepare native execution 

516 else: 

517 engine_type = 'native' 

518 

519 # Header 

520 if not quiet: 

521 job_history.header( 

522 pipeline_history.jobs_count, 

523 image, 

524 engine_type, 

525 ) 

526 

527 # Acquire project paths 

528 path_project = Paths.resolve(self.__options.path) 

529 path_parent = Paths.resolve(Path(self.__options.path).parent) 

530 

531 # Acquire project targets 

532 if host or real_paths: 

533 target_project = path_project 

534 target_parent = path_parent 

535 elif job.options.git_clone_path: 

536 git_clone_path = PurePosixPath(job.options.git_clone_path) 

537 target_project = Paths.get(git_clone_path / Path(path_project).name) 

538 target_parent = Paths.get(git_clone_path) 

539 elif random_paths: 

540 random_path: str = Strings.random(8) 

541 target_project = Paths.get( 

542 Platform.BUILDS_DIR / random_path / Path(path_project).name) 

543 target_parent = Paths.get(Platform.BUILDS_DIR / random_path) 

544 else: 

545 target_project = Paths.get(Platform.BUILDS_DIR / Path(path_project).name) 

546 target_parent = Paths.get(Platform.BUILDS_DIR) 

547 

548 # Prepare specific working directory 

549 if Flags.has( 

550 option=self.__options.workdir, 

551 flag=Bundle.FLAG_WORKDIR, 

552 variables=job.variables, 

553 ): 

554 relativedir: Path = Path('.') 

555 workdir: str = Flags.check_str( 

556 option=self.__options.workdir, 

557 flag=Bundle.FLAG_WORKDIR, 

558 variables=job.variables, 

559 default='.', 

560 ) 

561 

562 # Expand real working directory 

563 if Platform.IS_LINUX or Platform.IS_MAC_OS: 

564 workdir = Paths.expand(workdir) 

565 if host or real_paths: 

566 target_workdir = Paths.get((relativedir / workdir).resolve()) 

567 else: 

568 target_workdir = Paths.get(PurePosixPath(target_project) / workdir) 

569 

570 # Expand remote working directory 

571 else: # pragma: no cover 

572 if workdir[0:1] == '~': 

573 target_workdir = Paths.get(workdir) 

574 else: 

575 workdir = Paths.expand(workdir, home=False) 

576 if host or real_paths: 

577 target_workdir = Paths.get((relativedir / workdir).resolve()) 

578 else: 

579 target_workdir = Paths.get( 

580 PurePosixPath(target_project) / workdir) 

581 

582 # Prepare real working directory 

583 elif host or real_paths: 

584 target_workdir = Paths.get(self.__options.path) 

585 

586 # Prepare target working directory 

587 else: 

588 target_workdir = target_project 

589 

590 # Prepare entrypoint and scripts 

591 scripts_after: List[str] = [] 

592 scripts_before: List[str] = [] 

593 scripts_commands: List[str] = [] 

594 scripts_debug: List[str] = [] 

595 

596 # Prepare before_scripts 

597 if Flags.check_bool( 

598 option=self.__options.before, 

599 flag=Bundle.FLAG_BEFORE, 

600 variables=job.variables, 

601 default=True, 

602 ): 

603 scripts_before += job.before_script 

604 

605 # Prepare scripts 

606 scripts_commands += job.script 

607 if not host: 

608 if Flags.check_bool( 

609 option=self.__options.bash, 

610 flag=Bundle.FLAG_BASH, 

611 variables=job.variables, 

612 default=False, 

613 ): 

614 scripts_commands = [] 

615 if Flags.check_bool( 

616 option=self.__options.bash, 

617 flag=Bundle.FLAG_BASH, 

618 variables=job.variables, 

619 default=False, 

620 ) or Flags.check_bool( 

621 option=self.__options.debug, 

622 flag=Bundle.FLAG_DEBUG, 

623 variables=job.variables, 

624 default=False, 

625 ): 

626 scripts_debug += [ 

627 f"echo \"{self.__MARKER_DEBUG}\"", 'tail -f /dev/null || true' 

628 ] 

629 

630 # Prepare after_scripts 

631 if Flags.check_bool( 

632 option=self.__options.after, 

633 flag=Bundle.FLAG_AFTER, 

634 variables=job.variables, 

635 default=True, 

636 ): 

637 scripts_after += job.after_script 

638 

639 # Prepare script file 

640 script_file = ScriptsFile( 

641 paths={ 

642 path_parent: target_parent, 

643 path_project: target_project 

644 }, 

645 prefix='.tmp.entrypoint.', 

646 ) 

647 

648 # Prepare execution context 

649 script_file.shebang() 

650 script_file.writelines([ 

651 '# Variables', 

652 'result=1', 

653 '', 

654 ]) 

655 

656 # Prepare host working directory 

657 if host: 

658 script_file.writelines([ 

659 '# Work directory', 

660 f'cd "{target_workdir}"', 

661 '', 

662 ]) 

663 

664 # Configure Git 

665 if not host and not self.__options.no_git_safeties: 

666 script_file.writelines([ 

667 '# Configure Git safeties', 

668 'if type git >/dev/null 2>&1; then', 

669 f' git config --global --add safe.directory "{target_workdir}"', 

670 'else', 

671 ' echo "[safe]" >>~/.gitconfig', 

672 f' echo -e "\tdirectory = {target_workdir}" >>~/.gitconfig', 

673 'fi', 

674 '', 

675 ]) 

676 

677 # Configure Xauth, pylint: disable=line-too-long 

678 if not host: 

679 script_file.writelines([ 

680 '# Configure Xauthority', 

681 f'if [ ! -z "${{{Bundle.ENV_DISPLAY_XAUTH_MAGIC_ADD}}}" ] && type xauth >/dev/null 2>&1; then', 

682 f' if [ ! -z "${{{Platform.ENV_XAUTHORITY}}}" ] && [ ! -w "${{{Platform.ENV_XAUTHORITY}}}" ]; then', 

683 ' true', 

684 ' else', 

685 f' xauth -q add ${{{Bundle.ENV_DISPLAY_XAUTH_MAGIC_ADD}}} 2>/dev/null', 

686 ' fi', 

687 'fi', 

688 '', 

689 ]) 

690 

691 # Prepare before_script/script context 

692 script_file.subshell_start('before_script/script') 

693 script_file.configure( 

694 errors=True, 

695 verbose=job.options.verbose, 

696 ) 

697 

698 # Prepare before_script commands 

699 if len(scripts_before) > 0: 

700 script_file.write('# Run before_script') 

701 script_file.subgroup_start() 

702 script_file.writelines(scripts_before) 

703 script_file.subgroup_stop() 

704 script_file.write('') 

705 

706 # Prepare script commands 

707 if len(scripts_commands) > 0: 

708 script_file.write('# Run script') 

709 script_file.subgroup_start() 

710 script_file.writelines(scripts_commands) 

711 script_file.subgroup_stop() 

712 script_file.write('') 

713 else: 

714 script_file.write('# Missing script') 

715 script_file.write('false') 

716 script_file.write('') 

717 

718 # Finish before_script/script context 

719 script_file.subshell_stop('before_script/script') 

720 script_file.write('') 

721 

722 # Get before_script/script result 

723 script_file.write('# Result') 

724 script_file.write('result=${?}') 

725 script_file.write('') 

726 

727 # Prepare container result 

728 if not host: 

729 script_file.writelines([ 

730 '# Result marker', 

731 f'echo "{self.__MARKER_RESULT}:${{result}}"', 

732 '', 

733 ]) 

734 

735 # Prepare debug script commands 

736 if len(scripts_debug) > 0: 

737 script_file.subshell_start('debug') 

738 if job.options.verbose: 

739 script_file.configure( 

740 errors=False, 

741 verbose=True, 

742 ) 

743 script_file.writelines(scripts_debug) 

744 script_file.subshell_stop('debug') 

745 script_file.write('') 

746 

747 # Prepare after_script commands 

748 if len(scripts_after) > 0: 

749 script_file.subshell_start('after_script') 

750 script_file.configure( 

751 errors=True, 

752 verbose=job.options.verbose, 

753 ) 

754 script_file.write('# Run after_script') 

755 script_file.subgroup_start() 

756 script_file.writelines(scripts_after) 

757 script_file.subgroup_stop() 

758 script_file.subshell_stop('after_script') 

759 script_file.write('') 

760 

761 # Prepare container ending 

762 if not host: 

763 script_file.writelines([ 

764 '# Ending marker', 

765 f'echo "{self.__MARKER_ENDING}"', 

766 '', 

767 ]) 

768 

769 # Prepare execution result 

770 script_file.writelines([ 

771 '# Exit', 

772 'exit "${result}"', 

773 '', 

774 ]) 

775 

776 # Prepare script execution 

777 script_stat = stat(script_file.name) 

778 chmod(script_file.name, script_stat.st_mode | S_IXUSR | S_IXGRP 

779 | S_IRGRP | S_IROTH | S_IXOTH) 

780 script_file.close() 

781 

782 # Print script contents 

783 if self.__options.scripts: 

784 script_file.print() 

785 return True 

786 

787 # Store environment 

788 _environ = dict(environ) 

789 

790 # Acquire job data 

791 env_builds_path: str = job.options.env_builds_path 

792 env_job_name: str = job.options.env_job_name 

793 env_job_name_slug: str = job.options.env_job_name_slug 

794 env_job_path: str = job.options.env_job_path 

795 variables = job.variables.copy() 

796 

797 # Configure job settings 

798 environ[env_builds_path] = target_parent 

799 environ[env_job_name] = job.name 

800 environ[env_job_name_slug] = regex_sub( 

801 r'[^a-z0-9]', 

802 '-', 

803 job.name.lower()[0:63], 

804 ).strip('-') 

805 environ[env_job_path] = target_project 

806 environ[Bundle.ENV_LOCAL] = 'true' 

807 environ[Bundle.ENV_PROJECT_NAME] = Paths.basename(target_project) 

808 environ[Bundle.ENV_PROJECT_NAMESPACE] = Paths.basename(target_parent) 

809 

810 # Prepare Git environment 

811 git: Git = Git() 

812 

813 # Configure Git variables 

814 if Bundle.ENV_COMMIT_REF_NAME in variables: 

815 environ[Bundle.ENV_COMMIT_REF_NAME] = variables[Bundle.ENV_COMMIT_REF_NAME] 

816 else: 

817 environ[Bundle.ENV_COMMIT_REF_NAME] = git.head_reference_name(path_project) 

818 if Bundle.ENV_COMMIT_REF_SLUG in variables: 

819 environ[Bundle.ENV_COMMIT_REF_SLUG] = variables[Bundle.ENV_COMMIT_REF_SLUG] 

820 else: 

821 environ[Bundle.ENV_COMMIT_REF_SLUG] = git.head_reference_slug( 

822 path_project, name=environ[Bundle.ENV_COMMIT_REF_NAME]) 

823 if Bundle.ENV_COMMIT_SHA in variables: 

824 environ[Bundle.ENV_COMMIT_SHA] = variables[Bundle.ENV_COMMIT_SHA] 

825 else: 

826 environ[Bundle.ENV_COMMIT_SHA] = git.head_revision_hash(path_project) 

827 if Bundle.ENV_COMMIT_SHORT_SHA1 in variables: 

828 environ[Bundle.ENV_COMMIT_SHORT_SHA1] = variables[ 

829 Bundle.ENV_COMMIT_SHORT_SHA1] 

830 else: 

831 environ[Bundle.ENV_COMMIT_SHORT_SHA1] = git.head_revision_short_hash( 

832 path_project) 

833 if Bundle.ENV_COMMIT_TIMESTAMP in variables: 

834 environ[Bundle.ENV_COMMIT_TIMESTAMP] = variables[Bundle.ENV_COMMIT_TIMESTAMP] 

835 else: 

836 environ[Bundle.ENV_COMMIT_TIMESTAMP] = git.head_revision_timestamp( 

837 path_project) 

838 if Bundle.ENV_DEFAULT_BRANCH in variables: 

839 environ[Bundle.ENV_DEFAULT_BRANCH] = variables[Bundle.ENV_DEFAULT_BRANCH] 

840 else: 

841 environ[Bundle.ENV_DEFAULT_BRANCH] = git.branch_default(path_project) 

842 

843 # Prepare network 

844 network = '' 

845 if self.__options.network: 

846 network = self.__options.network 

847 variables[Bundle.FLAG_NETWORK] = network 

848 elif Bundle.FLAG_NETWORK in variables: 

849 network = variables[Bundle.FLAG_NETWORK] 

850 variables[Bundle.FLAG_NETWORK] = network 

851 

852 # Prepare privileged 

853 privileged = Flags.check_bool( 

854 option=self.__options.privileged,