Coverage for tests/conftest.py: 95%

1# This file is part of the Apio project 

2# (C) 2016-present FPGAwars 

3# Project authors: 

4# * Jesús Arroyo (Jesus89) (2016-2019) 

5# * Juan Gonzalez (obijuan) (2019-2024) 

6# * The Apio Team (2024-present) 

7# License: GPLv2 

8 

9"""Pytest TEST configuration file""" 

10 

11import sys 

12import json 

13import subprocess 

14from subprocess import CompletedProcess 

15from dataclasses import dataclass 

16from io import StringIO, TextIOBase 

17from collections.abc import Iterator 

18import shutil 

19import tempfile 

20import contextlib 

21from pathlib import Path 

22from typing import cast, Any 

23import os 

24from pprint import pprint 

25import pytest 

26from typing_extensions import override 

27from click.testing import CliRunner, Result 

28from rich.ansi import AnsiDecoder 

29from apio import __main__ 

30from apio.common import apio_console 

31from apio.common.proto.stubs.apio_scons_pb2 import FORCE_PIPE, FORCE_TERMINAL 

32 

33# -- Debug mode on/off 

34DEBUG = True 

35 

36# -- This is the marker we use to identify the sandbox directories. 

37SANDBOX_MARKER = "apio-sandbox" 

38 

39 

40# -- Decoder for removing ansi styles and colors. 

41ANSI_DECODER = AnsiDecoder() 

42 

43 

44class _LogTee(TextIOBase): 

45 """Output stream that writes both to a buffer and to the real 

46 std/err stream. Used by the 'with_log' operation.""" 

47 

48 def __init__(self, log_buffer: StringIO, live_stream: TextIOBase): 

49 self._buf = log_buffer 

50 self._live = live_stream 

51 

52 @override 

53 def write(self, s: str) -> int: 

54 # -- Write to the log buffer 

55 self._buf.write(s) 

56 # -- Write to live output. 

57 self._live.write(s) 

58 self._live.flush() 

59 # -- All done. 

60 return len(s) 

61 

62 @override 

63 def flush(self) -> None: 

64 self._buf.flush() 

65 self._live.flush() 

66 

67 

68class CapturedLog: 

69 """Holds the captured stdout + stderr.""" 

70 

71 def __init__(self): 

72 self._buf = StringIO() 

73 self._output_stream = _LogTee(self._buf, sys.__stdout__) 

74 

75 @property 

76 def output_stream(self) -> TextIOBase: 

77 """Getter to the output stream that sends the output to the logger 

78 buffer and to live output.""" 

79 return self._output_stream 

80 

81 @property 

82 def buf(self) -> StringIO: 

83 """Getter to the capture buffer.""" 

84 return self._buf 

85 

86 @property 

87 def out(self) -> str: 

88 """Return the full captured text with ansi styles and colors 

89 removed.""" 

90 text_obj = ANSI_DECODER.decode_line(self._buf.getvalue()) 

91 return text_obj.plain 

92 

93 @property 

94 def out_styled(self) -> str: 

95 """The original output, without the style and color stripping.""" 

96 return self.out 

97 

98 

99@dataclass(frozen=True) 

100class ApioResult: 

101 """Represent the outcome of an apio invocation.""" 

102 

103 exit_code: int 

104 output: str # stdout only 

105 exception: Any 

106 

107 

108class ApioSandbox: 

109 """Accessor for sandbox values. Available to the user inside an 

110 ApioRunner sandbox scope.""" 

111 

112 def __init__( 

113 self, 

114 apio_runner_: "ApioRunner", 

115 sandbox_dir: Path, 

116 proj_dir: Path, 

117 home_dir: Path, 

118 packages_dir_: Path, 

119 ): 

120 

121 # pylint: disable=too-many-arguments 

122 # pylint: disable=too-many-positional-arguments 

123 

124 assert isinstance(sandbox_dir, Path) 

125 assert isinstance(proj_dir, Path) 

126 assert isinstance(home_dir, Path) 

127 assert isinstance(packages_dir_, Path) 

128 

129 self._apio_runner = apio_runner_ 

130 self._sandbox_dir = sandbox_dir 

131 self._proj_dir = proj_dir 

132 self._home_dir = home_dir 

133 self._packages_dir = packages_dir_ 

134 self._click_runner = CliRunner() 

135 

136 @property 

137 def expired(self) -> bool: 

138 """Returns true if this sandbox was expired.""" 

139 # -- This tests if this sandbox is still the active sandbox at the 

140 # -- apio runner that creates it. 

141 return self is not self._apio_runner.sandbox 

142 

143 @property 

144 def sandbox_dir(self) -> Path: 

145 """Returns the sandbox's dir.""" 

146 assert not self.expired, "Sandbox expired" 

147 return self._sandbox_dir 

148 

149 @property 

150 def proj_dir(self) -> Path: 

151 """Returns the sandbox's apio project dir.""" 

152 assert not self.expired, "Sandbox expired" 

153 return self._proj_dir 

154 

155 @property 

156 def home_dir(self) -> Path: 

157 """Returns the sandbox's apio home dir.""" 

158 assert not self.expired, "Sandbox expired" 

159 return self._home_dir 

160 

161 @property 

162 def packages_dir(self) -> Path: 

163 """Returns the sandbox's apio packages dir.""" 

164 return self._packages_dir 

165 

166 def clear_packages(self): 

167 """Clear the packages cache, in case a test needs a clean start.""" 

168 # -- Sanity check the path and delete. 

169 assert "packages" in str(self.packages_dir).lower() 

170 shutil.rmtree(self.packages_dir) 

171 assert not self.packages_dir.exists() 

172 

173 def invoke_apio_cmd( 

174 self, 

175 cli, 

176 args: list[str], 

177 terminal_mode: bool = True, 

178 in_subprocess: bool = False, 

179 ) -> ApioResult: 

180 """Invoke an apio command. in_subprocess run apios in a subprocess, 

181 currently this suppresses colors because of the piping.""" 

182 

183 print(f"\nInvoking apio command [{cli.name}], args={args}.") 

184 

185 # -- It's a good opportunity to flush the output so far. 

186 sys.stdout.flush() 

187 sys.stderr.flush() 

188 

189 # -- Check that this sandbox is still alive. 

190 assert not self.expired, "Sandbox expired." 

191 

192 # -- Since we restore the env after invoking the apio command, we 

193 # -- don't expect path changes by the command to survive here. 

194 assert SANDBOX_MARKER not in os.environ["PATH"] 

195 

196 # -- Take a snapshot of the system env. 

197 original_env = os.environ.copy() 

198 

199 # -- These two env vars are set when creating the context. Let's 

200 # -- check that the test didn't corrupt them. 

201 assert os.environ["APIO_HOME"] == str(self.home_dir) 

202 

203 # -- If True, force terminal mode, if False, forces pipe mode, 

204 # -- otherwise auto which is pipe mode under pytest. 

205 apio_console.configure( 

206 terminal_mode=FORCE_TERMINAL if terminal_mode else FORCE_PIPE, 

207 ) 

208 

209 if in_subprocess: 

210 # -- Invoke apio in a sub process. 

211 print("Invoking apio in a sub process.") 

212 process_result: CompletedProcess = subprocess.run( 

213 [ 

214 sys.executable, 

215 __main__.__file__, 

216 ] 

217 + args, 

218 capture_output=True, 

219 encoding="utf-8", 

220 text=True, 

221 check=False, 

222 ) 

223 

224 apio_result = ApioResult( 

225 process_result.returncode, 

226 (process_result.stdout or "") + (process_result.stderr or ""), 

227 None, 

228 ) 

229 

230 else: 

231 # -- Invoke the command in the same process using click. 

232 print("Invoking apio in-process using click") 

233 click_result: Result = self._click_runner.invoke( 

234 prog_name="apio", 

235 cli=cli, 

236 args=args, 

237 color=terminal_mode, 

238 ) 

239 

240 # -- Convert click result to apio result. 

241 apio_result = ApioResult( 

242 click_result.exit_code, 

243 click_result.output, 

244 click_result.exception, 

245 ) 

246 

247 # -- Dump to test log. 

248 print(f"result.exit_code:{apio_result.exit_code}") 

249 print("result.output:") 

250 print(Result.output) 

251 

252 # -- Restore system env. Since apio commands tend to change vars 

253 # -- such as PATH. 

254 self.restore_system_env(original_env, "apio-command") 

255 

256 return apio_result 

257 

258 # -- List of default bad words for assert_ok(). All words should be 

259 # -- lower case. 

260 _DEFAULT_BAD_WORDS = ["error"] 

261 

262 def assert_result_ok( 

263 self, 

264 result: ApioResult, 

265 bad_words: list[str] | tuple[str, ...] = tuple(_DEFAULT_BAD_WORDS), 

266 ): 

267 """Check if apio command results where ok. Bad words is an optional 

268 list of lower case strings strings if found in the lower case version 

269 of the output text, trigger an error. The default is a tuple and not 

270 a list to avoid pylint warning about unsafe default value""" 

271 

272 assert isinstance(result, ApioResult) 

273 

274 # -- It should return an exit code of 0: success 

275 assert result.exit_code == 0, result.output 

276 

277 # -- There should be no exceptions raised 

278 assert not result.exception 

279 

280 # -- Check for bad words. 

281 lower_case_output = result.output.lower() 

282 for bad_word in bad_words: 

283 assert bad_word.islower(), bad_word 

284 

285 # -- Special case. For Xilinx arch it may be the case that 

286 # -- the message contains the string "0 errors". It has the 

287 # -- bad word 'error', but it is NOT an error 

288 # -- We exclude that case 

289 if "0 errors" not in lower_case_output: 289 ↛ 282line 289 didn't jump to line 282 because the condition on line 289 was always true

290 

291 # -- Check for no errors 

292 assert bad_word not in lower_case_output, bad_word 

293 

294 def restore_system_env( 

295 self, original_env: dict[str, str], scope: str 

296 ) -> None: 

297 """Overwrites the existing sys.environ with the given dict. Vars 

298 that are not in the dict are deleted and vars that have a different 

299 value in the dict is updated. Can be called only within a 

300 an apio sandbox.""" 

301 

302 # -- Check that the sandbox not expired. 

303 assert not self.expired, "Sandbox expired" 

304 

305 # -- NOTE: naively assigning the dict to os.environ will break 

306 # -- os.environ since a simple dict doesn't update the underlying 

307 # -- system env when it's mutated. 

308 

309 print(f"\nRestoring os.environ ({scope} scope):") 

310 

311 # -- Construct the union of the env and the dict var names. 

312 all_var_names = set(os.environ.keys()).union(original_env.keys()) 

313 for name in all_var_names: 

314 # Get the env and dict values. None if doesn't exist. 

315 current_val = os.environ.get(name, None) 

316 original_val = original_env.get(name, None) 

317 # -- If values are not the same, update the env. 

318 if current_val != original_val: 

319 print(f" set ${name}={original_val} (was {current_val})") 

320 if original_val is None: 320 ↛ 323line 320 didn't jump to line 323 because the condition on line 320 was always true

321 os.environ.pop(name) 

322 else: 

323 os.environ[name] = original_val 

324 

325 # -- Sanity check. System env and the dict should be the same. 

326 assert os.environ == original_env 

327 

328 def write_file( 

329 self, 

330 file: str | Path, 

331 text: str | list[str], 

332 exists_ok=False, 

333 ) -> None: 

334 """Write text to given file. If text is a list, items are joined with 

335 "\n". 'file' can be a string or a Path.""" 

336 

337 assert exists_ok or not Path(file).exists(), f"File exists: {file}" 

338 

339 # -- If a list is given, join with '\n" 

340 if isinstance(text, list): 

341 text = "\n".join(text) 

342 

343 # -- Make dir(s) if needed. 

344 Path(file).parent.mkdir(parents=True, exist_ok=True) 

345 

346 # -- Write. 

347 with open(file, "w", encoding="utf-8") as f: 

348 f.write(text) 

349 

350 def read_file_text(self, file: str | Path) -> str: 

351 """Read a text file. Returns a string with the text or if""" 

352 with open(file, "r", encoding="utf8") as f: 

353 text = f.read() 

354 return text 

355 

356 def read_file_lines(self, file: str | Path) -> list[str]: 

357 """Read a text file. Returns a string split into lines.""" 

358 text = self.read_file_text(file) 

359 text_lines = text.split("\n") 

360 return text_lines 

361 

362 def write_json_file( 

363 self, 

364 file: str | Path, 

365 json_data: dict[str, Any], 

366 exists_ok=False, 

367 ): 

368 """Write a dict to given json file. 'file' can be a string or a 

369 Path.""" 

370 self.write_file( 

371 file, json.dumps(json_data, indent=2), exists_ok=exists_ok 

372 ) 

373 

374 def read_json_file(self, file: str | Path) -> dict[str, Any]: 

375 """Read a json file. 'file' can be a string or a Path.""" 

376 json_text = self.read_file_text(file) 

377 json_data = json.loads(json_text) 

378 return json_data 

379 

380 def write_apio_ini( 

381 self, 

382 sections: dict[str, dict[str, str]] | None = None, 

383 ): 

384 """Write in the current directory an apio.ini file with given 

385 section. If an apio.ini file already exists, overwrite it.""" 

386 

387 assert isinstance(sections, dict) 

388 

389 # -- Construct output file path. 

390 path = Path("apio.ini") 

391 

392 # -- List with text of each section. 

393 sections_texts: list[str] = [] 

394 

395 # -- Add the apio section if specified. 

396 for section_header, section_options in sections.items(): 

397 lines = [section_header] 

398 for name, value in section_options.items(): 

399 lines.append(f"{name} = {value}") 

400 sections_texts.append("\n".join(lines)) 

401 

402 # -- Join the sections with a blank line. 

403 file_text = "\n\n".join(sections_texts) 

404 

405 # # -- Write the file. 

406 self.write_file(path, file_text, exists_ok=True) 

407 

408 def write_default_apio_ini(self): 

409 """Write in the local directory an apio.ini file with default values 

410 for testing. If the file exists, it's overwritten.""" 

411 

412 default_apio_ini = { 

413 "[env:default]": { 

414 "board": "alhambra-ii", 

415 "top-module": "main", 

416 } 

417 } 

418 self.write_apio_ini(default_apio_ini) 

419 

420 

421class ApioRunner: 

422 """Apio commands test helper. Provides an apio sandbox functionality 

423 (disposable temp dirs and sys.environ restoration) as well as a few 

424 utility functions. A typical test with the ApiRunner looks like this: 

425 

426 def test_my_cmd(apio_runner): 

427 with apio_runner.in_sandbox() as sb: 

428 

429 <the test body> 

430 """ 

431 

432 def __init__(self, request: pytest.FixtureRequest): 

433 print("*** creating ApioRunner") 

434 assert isinstance(request, pytest.FixtureRequest) 

435 

436 print("\nOriginal env:") 

437 pprint(dict(os.environ), width=80, sort_dicts=True) 

438 print() 

439 

440 # -- Get a pytest directory for the apio packages cache. This will 

441 # -- avoid reloading packages by each apio invocation. 

442 cache = request.config.cache 

443 self._packages_dir = cache.mkdir("apio-cached-packages") 

444 

445 # -- A CliRunner instance that is used for creating temp directories 

446 # -- and to invoke apio commands. 

447 self._request = request 

448 

449 # -- Indicate that we are not in a sandbox 

450 self._sandbox: ApioSandbox | None = None 

451 

452 # -- A placeholder for a shared apio home that we may use in some 

453 # -- of the sandboxes. 

454 self._shared_apio_home: Path | None = None 

455 

456 # -- Register a cleanup method. It's called at the end of the 

457 # -- apio_runner fixture scope. 

458 request.addfinalizer(self._teardown) 

459 

460 def _teardown(self): 

461 """Teardown at the end of the apio_runner fixture scope.""" 

462 if self._shared_apio_home: 462 ↛ 463line 462 didn't jump to line 463 because the condition on line 462 was never true

463 print(f"Deleting apio shared home {str(self._shared_apio_home)}") 

464 assert "apio" in str(self._shared_apio_home) 

465 shutil.rmtree(self._shared_apio_home) 

466 self._shared_apio_home = None 

467 

468 @staticmethod 

469 def _get_local_config_url() -> str: 

470 """Returns a file:/ URL to the remote config file in the local depot. 

471 This is used to set APIO_REMOTE_CONFIG_URL for testing to make sure 

472 we test with the latest remote config in this change rather than with 

473 the published remote config. 

474 """ 

475 

476 # -- Construct the path of the config dir in this repo. We compute it 

477 # -- based on the path of this conftest.py python file. 

478 local_config_dir = os.path.normpath( 

479 os.path.join( 

480 os.path.abspath(__file__), 

481 "..", 

482 "..", 

483 "remote-config", 

484 ) 

485 ) 

486 # -- Convert the file path to a URL with a 'file://' form. 

487 local_config_url = "file://" + str(local_config_dir) 

488 

489 return local_config_url 

490 

491 @contextlib.contextmanager 

492 def with_logger(self): 

493 """Capture stdout + stderr and yield a CapturedLog object.""" 

494 print("----- Begin log") 

495 log = CapturedLog() 

496 with ( 

497 contextlib.redirect_stdout(log.output_stream), 

498 contextlib.redirect_stderr(log.output_stream), 

499 ): 

500 yield log 

501 print("----- End log") 

502 

503 @property 

504 def sandbox(self) -> ApioSandbox | None: 

505 """Returns the sandbox object or None if not in a sandbox.""" 

506 return self._sandbox 

507 

508 @contextlib.contextmanager 

509 def in_sandbox(self) -> Iterator[ApioSandbox]: 

510 """Create an apio sandbox context manager that delete the temp dir 

511 and restore the system env upon exist. 

512 

513 Upon return, the current directory is proj_dir. 

514 """ 

515 # -- Make sure we don't try to nest sandboxes. 

516 assert self._sandbox is None, "Already in a sandbox." 

517 

518 # -- Snapshot the system env. 

519 original_env: dict[str, str] = os.environ.copy() 

520 

521 # -- Snapshot the current directory. 

522 original_cwd = os.getcwd() 

523 

524 # -- Create a temp sandbox dir that will be deleted on exit and 

525 # -- change to it. 

526 sandbox_dir = Path(tempfile.mkdtemp(prefix=SANDBOX_MARKER + "-")) 

527 

528 # -- Make the sandbox's project directory. 

529 # -- Initially, we intentionally used a 

530 # -- directory name with non ascii character to test 

531 # -- that apio can handle it. 

532 # proj_dir = sandbox_dir / "apio prój" 

533 # -- I works ok for the architectures: ice40, ecp5, gowin 

534 # -- BUT not for Xilinx. If the path contains a non-ascii character 

535 # -- it complains 

536 proj_dir = sandbox_dir / "apio proj" 

537 proj_dir.mkdir() 

538 os.chdir(proj_dir) 

539 

540 # -- Determine the project home dir. 

541 home_dir = sandbox_dir / "apio-home" 

542 

543 if DEBUG: 543 ↛ 553line 543 didn't jump to line 553 because the condition on line 543 was always true

544 print() 

545 print("--> apio sandbox:") 

546 print(f" sandbox dir : {str(sandbox_dir)}") 

547 print(f" apio proj dir : {str(proj_dir)}") 

548 print(f" apio home dir : {str(home_dir)}") 

549 print(f" apio packages dir : {str(self._packages_dir)}") 

550 print() 

551 

552 # -- Register a sandbox objet to indicate that we are in a sandbox. 

553 assert self._sandbox is None 

554 self._sandbox = ApioSandbox( 

555 self, sandbox_dir, proj_dir, home_dir, self._packages_dir 

556 ) 

557 

558 # -- Set the system env vars to inform ApioContext what are the 

559 # -- home and packages dirs. 

560 os.environ["APIO_HOME"] = str(home_dir) 

561 os.environ["APIO_PACKAGES"] = str(self._packages_dir) 

562 

563 local_config_url = self._get_local_config_url() 

564 print(f"Local config url: {local_config_url}") 

565 

566 # Sanity check to detect conflicts from prior URL settings. 

567 assert ( 

568 os.environ.get("APIO_REMOTE_CONFIG_URL") is None 

569 or os.environ.get("APIO_REMOTE_CONFIG_URL") == local_config_url 

570 ), ( 

571 "A predefined env var APIO_REMOTE_CONFIG_URL conflicts with " 

572 "test settings. Update or unset it." 

573 ) 

574 

575 # Set the URL in the environment 

576 os.environ["APIO_REMOTE_CONFIG_URL"] = local_config_url 

577 

578 # -- Reset the apio console, since we run multiple sandboxes in the 

579 # -- same process. 

580 apio_console.configure( 

581 terminal_mode=FORCE_TERMINAL, theme_name="light" 

582 ) 

583 

584 try: 

585 # -- This is the end of the context manager _entry part. The 

586 # -- call to _exit will continue execution after the yield. 

587 # -- Value is the sandbox object we pass to the user. 

588 yield cast(ApioSandbox, self._sandbox) 

589 

590 finally: 

591 # -- Here when the context manager exit, normally or through an 

592 # -- exception. 

593 

594 # -- Restore the original system env. 

595 self._sandbox.restore_system_env(original_env, "sandbox") 

596 

597 # -- Mark that we exited the sandbox. This expires the sandbox. 

598 self._sandbox = None 

599 

600 # -- Change back to the original directory. 

601 os.chdir(original_cwd) 

602 

603 # -- Delete the temp directory. This also deletes the apio home 

604 # -- if it's not shared but doesn't touch it if we use a shared 

605 # -- home. 

606 shutil.rmtree(sandbox_dir) 

607 

608 print("\nSandbox deleted. ") 

609 

610 # -- Flush the output so far. 

611 sys.stdout.flush() 

612 sys.stderr.flush() 

613 

614 def is_on_github_workflow(self) -> bool: 

615 """Returns True if running on a github workflow.""" 

616 # -- NOTE: The env var GITHUB_ACTIONS needs to be whitelisted 

617 # -- in tox.ini for it to reach the test. It's a standard env var 

618 # -- defined by the github workflows. 

619 val = os.environ.get("GITHUB_ACTIONS") 

620 return val == "true" 

621 

622 

623@pytest.fixture(scope="module") 

624def apio_runner(request): 

625 """A pytest fixture that provides tests with a ApioRunner test 

626 helper object. We use a 'module' scope so sandboxes can share the apio 

627 home with other tests in the same file, if we choose to do so, 

628 to reused previously installed packages. 

629 """ 

630 assert isinstance(request, pytest.FixtureRequest) 

631 return ApioRunner(request)