Coverage for tests/conftest.py: 95%
254 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 03:53 +0000
1"""Pytest TEST configuration file"""
3import sys
4import json
5import subprocess
6from subprocess import CompletedProcess
7from dataclasses import dataclass
8from io import StringIO, TextIOBase
9from collections.abc import Iterator
10import shutil
11import tempfile
12import contextlib
13from pathlib import Path, PurePosixPath
14from typing import cast, Any
15import os
16from urllib.parse import urlparse
17from pprint import pprint
18import pytest
19from click.testing import CliRunner, Result
20from rich.ansi import AnsiDecoder
21import json5
22from apio import __main__
23from apio.common import apio_console
24from apio.common.proto.apio_scons_pb2 import FORCE_PIPE, FORCE_TERMINAL
26# -- Debug mode on/off
27DEBUG = True
29# -- This is the marker we use to identify the sandbox directories.
30SANDBOX_MARKER = "apio-sandbox"
33# -- Decoder for removing ansi styles and colors.
34ANSI_DECODER = AnsiDecoder()
37class _LogTee(TextIOBase):
38 """Output stream that writes both to a buffer and to the real
39 std/err stream. Used by the 'with_log' operation."""
41 def __init__(self, log_buffer: StringIO, live_stream: TextIOBase):
42 self._buf = log_buffer
43 self._live = live_stream
45 # @overrides
46 def write(self, s: str) -> int:
47 # -- Write to the log buffer
48 self._buf.write(s)
49 # -- Write to live output.
50 self._live.write(s)
51 self._live.flush()
52 # -- All done.
53 return len(s)
55 # @overrides
56 def flush(self) -> None:
57 self._buf.flush()
58 self._live.flush()
61class CapturedLog:
62 """Holds the captured stdout + stderr."""
64 def __init__(self):
65 self._buf = StringIO()
66 self._output_stream = _LogTee(self._buf, sys.__stdout__)
68 @property
69 def output_stream(self) -> TextIOBase:
70 """Getter to the output stream that sends the output to the logger
71 buffer and to live output."""
72 return self._output_stream
74 @property
75 def buf(self) -> StringIO:
76 """Getter to the capture buffer."""
77 return self._buf
79 @property
80 def out(self) -> str:
81 """Return the full captured text with ansi styles and colors
82 removed."""
83 text_obj = ANSI_DECODER.decode_line(self._buf.getvalue())
84 return text_obj.plain
86 @property
87 def out_styled(self) -> str:
88 """The original output, without the style and color stripping."""
89 return self.out
92@dataclass(frozen=True)
93class ApioResult:
94 """Represent the outcome of an apio invocation."""
96 exit_code: int
97 output: str # stdout only
98 exception: Any
101class ApioSandbox:
102 """Accessor for sandbox values. Available to the user inside an
103 ApioRunner sandbox scope."""
105 def __init__(
106 self,
107 apio_runner_: "ApioRunner",
108 sandbox_dir: Path,
109 proj_dir: Path,
110 home_dir: Path,
111 packages_dir_: Path,
112 ):
114 # pylint: disable=too-many-arguments
115 # pylint: disable=too-many-positional-arguments
117 assert isinstance(sandbox_dir, Path)
118 assert isinstance(proj_dir, Path)
119 assert isinstance(home_dir, Path)
120 assert isinstance(packages_dir_, Path)
122 self._apio_runner = apio_runner_
123 self._sandbox_dir = sandbox_dir
124 self._proj_dir = proj_dir
125 self._home_dir = home_dir
126 self._packages_dir = packages_dir_
127 self._click_runner = CliRunner()
129 @property
130 def expired(self) -> bool:
131 """Returns true if this sandbox was expired."""
132 # -- This tests if this sandbox is still the active sandbox at the
133 # -- apio runner that creates it.
134 return self is not self._apio_runner.sandbox
136 @property
137 def sandbox_dir(self) -> Path:
138 """Returns the sandbox's dir."""
139 assert not self.expired, "Sandbox expired"
140 return self._sandbox_dir
142 @property
143 def proj_dir(self) -> Path:
144 """Returns the sandbox's apio project dir."""
145 assert not self.expired, "Sandbox expired"
146 return self._proj_dir
148 @property
149 def home_dir(self) -> Path:
150 """Returns the sandbox's apio home dir."""
151 assert not self.expired, "Sandbox expired"
152 return self._home_dir
154 @property
155 def packages_dir(self) -> Path:
156 """Returns the sandbox's apio packages dir."""
157 return self._packages_dir
159 def clear_packages(self):
160 """Clear the packages cache, in case a test needs a clean start."""
161 # -- Sanity check the path and delete.
162 assert "packages" in str(self.packages_dir).lower()
163 shutil.rmtree(self.packages_dir)
164 assert not self.packages_dir.exists()
166 def invoke_apio_cmd(
167 self,
168 cli,
169 args: list[str],
170 terminal_mode: bool = True,
171 in_subprocess: bool = False,
172 ) -> ApioResult:
173 """Invoke an apio command. in_subprocess run apios in a subprocess,
174 currently this suppresses colors because of the piping."""
176 print(f"\nInvoking apio command [{cli.name}], args={args}.")
178 # -- It's a good opportunity to flush the output so far.
179 sys.stdout.flush()
180 sys.stderr.flush()
182 # -- Check that this sandbox is still alive.
183 assert not self.expired, "Sandbox expired."
185 # -- Since we restore the env after invoking the apio command, we
186 # -- don't expect path changes by the command to survive here.
187 assert SANDBOX_MARKER not in os.environ["PATH"]
189 # -- Take a snapshot of the system env.
190 original_env = os.environ.copy()
192 # -- These two env vars are set when creating the context. Let's
193 # -- check that the test didn't corrupt them.
194 assert os.environ["APIO_HOME"] == str(self.home_dir)
196 # -- If True, force terminal mode, if False, forces pipe mode,
197 # -- otherwise auto which is pipe mode under pytest.
198 apio_console.configure(
199 terminal_mode=FORCE_TERMINAL if terminal_mode else FORCE_PIPE,
200 )
202 if in_subprocess:
203 # -- Invoke apio in a sub process.
204 print("Invoking apio in a sub process.")
205 process_result: CompletedProcess = subprocess.run(
206 [
207 sys.executable,
208 __main__.__file__,
209 ]
210 + args,
211 capture_output=True,
212 encoding="utf-8",
213 text=True,
214 check=False,
215 )
217 apio_result = ApioResult(
218 process_result.returncode,
219 (process_result.stdout or "") + (process_result.stderr or ""),
220 None,
221 )
223 else:
224 # -- Invoke the command in the same process using click.
225 print("Invoking apio in-process using click")
226 click_result: Result = self._click_runner.invoke(
227 prog_name="apio",
228 cli=cli,
229 args=args,
230 color=terminal_mode,
231 )
233 # -- Convert click result to apio result.
234 apio_result = ApioResult(
235 click_result.exit_code,
236 click_result.output,
237 click_result.exception,
238 )
240 # -- Dump to test log.
241 print(f"result.exit_code:{apio_result.exit_code}")
242 print("result.output:")
243 print(Result.output)
245 # -- Restore system env. Since apio commands tend to change vars
246 # -- such as PATH.
247 self.restore_system_env(original_env, "apio-command")
249 return apio_result
251 # -- List of default bad words for assert_ok(). All words should be
252 # -- lower case.
253 _DEFAULT_BAD_WORDS = ["error"]
255 def assert_result_ok(
256 self,
257 result: ApioResult,
258 bad_words: list[str] | tuple[str, ...] = tuple(_DEFAULT_BAD_WORDS),
259 ):
260 """Check if apio command results where ok. Bad words is an optional
261 list of lower case strings strings if found in the lower case version
262 of the output text, trigger an error. The default is a tuple and not
263 a list to avoid pylint warning about unsafe default value"""
265 assert isinstance(result, ApioResult)
267 # -- It should return an exit code of 0: success
268 assert result.exit_code == 0, result.output
270 # -- There should be no exceptions raised
271 assert not result.exception
273 # -- Check for bad words.
274 lower_case_output = result.output.lower()
275 for bad_word in bad_words:
276 assert bad_word.islower(), bad_word
278 # -- Special case. For Xilinx arch it may be the case that
279 # -- the message contains the string "0 errors". It has the
280 # -- bad word 'error', but it is NOT an error
281 # -- We exclude that case
282 if "0 errors" not in lower_case_output:
284 # -- Check for no errors
285 assert bad_word not in lower_case_output, bad_word
287 def restore_system_env(
288 self, original_env: dict[str, str], scope: str
289 ) -> None:
290 """Overwrites the existing sys.environ with the given dict. Vars
291 that are not in the dict are deleted and vars that have a different
292 value in the dict is updated. Can be called only within a
293 an apio sandbox."""
295 # -- Check that the sandbox not expired.
296 assert not self.expired, "Sandbox expired"
298 # -- NOTE: naively assigning the dict to os.environ will break
299 # -- os.environ since a simple dict doesn't update the underlying
300 # -- system env when it's mutated.
302 print(f"\nRestoring os.environ ({scope} scope):")
304 # -- Construct the union of the env and the dict var names.
305 all_var_names = set(os.environ.keys()).union(original_env.keys())
306 for name in all_var_names:
307 # Get the env and dict values. None if doesn't exist.
308 current_val = os.environ.get(name, None)
309 original_val = original_env.get(name, None)
310 # -- If values are not the same, update the env.
311 if current_val != original_val:
312 print(f" set ${name}={original_val} (was {current_val})")
313 if original_val is None:
314 os.environ.pop(name)
315 else:
316 os.environ[name] = original_val
318 # -- Sanity check. System env and the dict should be the same.
319 assert os.environ == original_env
321 def write_file(
322 self,
323 file: str | Path,
324 text: str | list[str],
325 exists_ok=False,
326 ) -> None:
327 """Write text to given file. If text is a list, items are joined with
328 "\n". 'file' can be a string or a Path."""
330 assert exists_ok or not Path(file).exists(), f"File exists: {file}"
332 # -- If a list is given, join with '\n"
333 if isinstance(text, list):
334 text = "\n".join(text)
336 # -- Make dir(s) if needed.
337 Path(file).parent.mkdir(parents=True, exist_ok=True)
339 # -- Write.
340 with open(file, "w", encoding="utf-8") as f:
341 f.write(text)
343 def read_file_text(self, file: str | Path) -> str:
344 """Read a text file. Returns a string with the text or if"""
345 with open(file, "r", encoding="utf8") as f:
346 text = f.read()
347 return text
349 def read_file_lines(self, file: str | Path) -> list[str]:
350 """Read a text file. Returns a string split into lines."""
351 text = self.read_file_text(file)
352 text_lines = text.split("\n")
353 return text_lines
355 def write_json_file(
356 self,
357 file: str | Path,
358 json_data: dict[str, dict],
359 exists_ok=False,
360 ):
361 """Write a dict to given json file. 'file' can be a string or a
362 Path."""
363 self.write_file(
364 file, json.dumps(json_data, indent=2), exists_ok=exists_ok
365 )
367 def read_json_file(self, file: str | Path) -> dict[str, Any]:
368 """Read a json file. 'file' can be a string or a Path."""
369 json_text = self.read_file_text(file)
370 json_data = json.loads(json_text)
371 return json_data
373 def write_apio_ini(
374 self,
375 sections: dict[str, dict[str, str]] | None = None,
376 ):
377 """Write in the current directory an apio.ini file with given
378 section. If an apio.ini file already exists, overwrite it."""
380 assert isinstance(sections, dict)
382 # -- Construct output file path.
383 path = Path("apio.ini")
385 # -- List with text of each section.
386 sections_texts: list[str] = []
388 # -- Add the apio section if specified.
389 for section_header, section_options in sections.items():
390 lines = [section_header]
391 for name, value in section_options.items():
392 lines.append(f"{name} = {value}")
393 sections_texts.append("\n".join(lines))
395 # -- Join the sections with a blank line.
396 file_text = "\n\n".join(sections_texts)
398 # # -- Write the file.
399 self.write_file(path, file_text, exists_ok=True)
401 def write_default_apio_ini(self):
402 """Write in the local directory an apio.ini file with default values
403 for testing. If the file exists, it's overwritten."""
405 default_apio_ini = {
406 "[env:default]": {
407 "board": "alhambra-ii",
408 "top-module": "main",
409 }
410 }
411 self.write_apio_ini(default_apio_ini)
414class ApioRunner:
415 """Apio commands test helper. Provides an apio sandbox functionality
416 (disposable temp dirs and sys.environ restoration) as well as a few
417 utility functions. A typical test with the ApiRunner looks like this:
419 def test_my_cmd(apio_runner):
420 with apio_runner.in_sandbox() as sb:
422 <the test body>
423 """
425 def __init__(self, request: pytest.FixtureRequest):
426 print("*** creating ApioRunner")
427 assert isinstance(request, pytest.FixtureRequest)
429 print("\nOriginal env:")
430 pprint(dict(os.environ), width=80, sort_dicts=True)
431 print()
433 # -- Get a pytest directory for the apio packages cache. This will
434 # -- avoid reloading packages by each apio invocation.
435 cache = request.config.cache
436 self._packages_dir = cache.mkdir("apio-cached-packages")
438 # -- A CliRunner instance that is used for creating temp directories
439 # -- and to invoke apio commands.
440 self._request = request
442 # -- Indicate that we are not in a sandbox
443 self._sandbox: ApioSandbox | None = None
445 # -- A placeholder for a shared apio home that we may use in some
446 # -- of the sandboxes.
447 self._shared_apio_home: Path | None = None
449 # -- Register a cleanup method. It's called at the end of the
450 # -- apio_runner fixture scope.
451 request.addfinalizer(self._teardown)
453 def _teardown(self):
454 """Teardown at the end of the apio_runner fixture scope."""
455 if self._shared_apio_home: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 print(f"Deleting apio shared home {str(self._shared_apio_home)}")
457 assert "apio" in str(self._shared_apio_home)
458 shutil.rmtree(self._shared_apio_home)
459 self._shared_apio_home = None
461 @staticmethod
462 def _get_local_config_url() -> str:
463 """Returns a file:/ URL to the remote config file in the local depot.
464 This is used to set APIO_REMOTE_CONFIG_URL for testing to make sure
465 we test with the latest remote config in this change rather than with
466 the published remote config.
467 """
469 # -- Read apio/resources/config.jsonc so we can extract the remote
470 # -- config file name template to construct the local URL.
471 this_file_path = Path(__file__).resolve()
472 config_file_path = (
473 this_file_path.parent.parent / "apio/resources/config.jsonc"
474 )
476 jsonc_text: str = config_file_path.read_text(encoding="utf-8")
477 json_data: dict = json5.loads(jsonc_text)
479 # -- Get the original remote config url.
480 url_str: str = json_data["remote-config-url"]
482 # -- Extract the file name part. E.g. 'apio-{major}.{minor}.x.jsonc'.
483 # -- The {major} and {minor} are placeholders for apio's major and
484 # -- minor version number which we don't resolve here.
485 remote_config_file_path: str = urlparse(url_str).path
486 remote_config_file_name: str = PurePosixPath(
487 remote_config_file_path
488 ).name
489 print(f"remote-config-file-name = {remote_config_file_name}")
491 # -- Construct the path of the config file in this repo. We compute it
492 # -- based on the path of this conftest.py python file.
493 local_config_file = os.path.normpath(
494 os.path.join(
495 os.path.abspath(__file__),
496 "..",
497 "..",
498 "remote-config",
499 remote_config_file_name,
500 )
501 )
502 # -- Convert the file path to a URL with a 'file://' form.
503 local_config_url = "file://" + str(local_config_file)
505 return local_config_url
507 @contextlib.contextmanager
508 def with_logger(self):
509 """Capture stdout + stderr and yield a CapturedLog object."""
510 print("----- Begin log")
511 log = CapturedLog()
512 with (
513 contextlib.redirect_stdout(log.output_stream),
514 contextlib.redirect_stderr(log.output_stream),
515 ):
516 yield log
517 print("----- End log")
519 @property
520 def sandbox(self) -> ApioSandbox | None:
521 """Returns the sandbox object or None if not in a sandbox."""
522 return self._sandbox
524 @contextlib.contextmanager
525 def in_sandbox(self) -> Iterator[ApioSandbox]:
526 """Create an apio sandbox context manager that delete the temp dir
527 and restore the system env upon exist.
529 Upon return, the current directory is proj_dir.
530 """
531 # -- Make sure we don't try to nest sandboxes.
532 assert self._sandbox is None, "Already in a sandbox."
534 # -- Snapshot the system env.
535 original_env: dict[str, str] = os.environ.copy()
537 # -- Snapshot the current directory.
538 original_cwd = os.getcwd()
540 # -- Create a temp sandbox dir that will be deleted on exit and
541 # -- change to it.
542 sandbox_dir = Path(tempfile.mkdtemp(prefix=SANDBOX_MARKER + "-"))
544 # -- Make the sandbox's project directory.
545 # -- Initially, we intentionally used a
546 # -- directory name with non ascii character to test
547 # -- that apio can handle it.
548 # proj_dir = sandbox_dir / "apio prój"
549 # -- I works ok for the architectures: ice40, ecp5, gowin
550 # -- BUT not for Xilinx. If the path contains a non-ascii character
551 # -- it complains
552 proj_dir = sandbox_dir / "apio proj"
553 proj_dir.mkdir()
554 os.chdir(proj_dir)
556 # -- Determine the project home dir.
557 home_dir = sandbox_dir / "apio-home"
559 if DEBUG: 559 ↛ 569line 559 didn't jump to line 569 because the condition on line 559 was always true
560 print()
561 print("--> apio sandbox:")
562 print(f" sandbox dir : {str(sandbox_dir)}")
563 print(f" apio proj dir : {str(proj_dir)}")
564 print(f" apio home dir : {str(home_dir)}")
565 print(f" apio packages dir : {str(self._packages_dir)}")
566 print()
568 # -- Register a sandbox objet to indicate that we are in a sandbox.
569 assert self._sandbox is None
570 self._sandbox = ApioSandbox(
571 self, sandbox_dir, proj_dir, home_dir, self._packages_dir
572 )
574 # -- Set the system env vars to inform ApioContext what are the
575 # -- home and packages dirs.
576 os.environ["APIO_HOME"] = str(home_dir)
577 os.environ["APIO_PACKAGES"] = str(self._packages_dir)
579 local_config_url = self._get_local_config_url()
580 print(f"Local config url: {local_config_url}")
582 # Sanity check to detect conflicts from prior URL settings.
583 assert (
584 os.environ.get("APIO_REMOTE_CONFIG_URL") is None
585 or os.environ.get("APIO_REMOTE_CONFIG_URL") == local_config_url
586 ), (
587 "A predefined env var APIO_REMOTE_CONFIG_URL conflicts with "
588 "test settings. Update or unset it."
589 )
591 # Set the URL in the environment
592 os.environ["APIO_REMOTE_CONFIG_URL"] = local_config_url
594 # -- Reset the apio console, since we run multiple sandboxes in the
595 # -- same process.
596 apio_console.configure(
597 terminal_mode=FORCE_TERMINAL, theme_name="light"
598 )
600 try:
601 # -- This is the end of the context manager _entry part. The
602 # -- call to _exit will continue execution after the yield.
603 # -- Value is the sandbox object we pass to the user.
604 yield cast(ApioSandbox, self._sandbox)
606 finally:
607 # -- Here when the context manager exit, normally or through an
608 # -- exception.
610 # -- Restore the original system env.
611 self._sandbox.restore_system_env(original_env, "sandbox")
613 # -- Mark that we exited the sandbox. This expires the sandbox.
614 self._sandbox = None
616 # -- Change back to the original directory.
617 os.chdir(original_cwd)
619 # -- Delete the temp directory. This also deletes the apio home
620 # -- if it's not shared but doesn't touch it if we use a shared
621 # -- home.
622 shutil.rmtree(sandbox_dir)
624 print("\nSandbox deleted. ")
626 # -- Flush the output so far.
627 sys.stdout.flush()
628 sys.stderr.flush()
630 def is_on_github_workflow(self) -> bool:
631 """Returns True if running on a github workflow."""
632 # -- NOTE: The env var GITHUB_ACTIONS needs to be whitelisted
633 # -- in tox.ini for it to reach the test. It's a standard env var
634 # -- defined by the github workflows.
635 val = os.environ.get("GITHUB_ACTIONS")
636 return val == "true"
639@pytest.fixture(scope="module")
640def apio_runner(request):
641 """A pytest fixture that provides tests with a ApioRunner test
642 helper object. We use a 'module' scope so sandboxes can share the apio
643 home with other tests in the same file, if we choose to do so,
644 to reused previously installed packages.
645 """
646 assert isinstance(request, pytest.FixtureRequest)
647 return ApioRunner(request)