Coverage for tests/conftest.py: 96%
205 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-24 03:51 +0000
1"""Pytest TEST configuration file"""
3import sys
4import json
5import subprocess
6from subprocess import CompletedProcess
7from dataclasses import dataclass
8import shutil
9import tempfile
10import contextlib
11from pathlib import Path, PurePosixPath
12from typing import List, Union, cast, Optional, Dict, Any, Tuple
13import os
14from urllib.parse import urlparse
15from pprint import pprint
16import pytest
17from click.testing import CliRunner, Result
18from apio import __main__
19from apio.common import apio_console
20from apio.common.proto.apio_pb2 import FORCE_PIPE, FORCE_TERMINAL
21from apio.utils import jsonc
24# -- Debug mode on/off
25DEBUG = True
27# -- This is the marker we use to identify the sandbox directories.
28SANDBOX_MARKER = "apio-sandbox"
31@dataclass(frozen=True)
32class ApioResult:
33 """Represent the outcome of an apio invocation."""
35 exit_code: int
36 output: str # stdout only
37 exception: Any
40class ApioSandbox:
41 """Accessor for sandbox values. Available to the user inside an
42 ApioRunner sandbox scope."""
44 def __init__(
45 self,
46 apio_runner_: "ApioRunner",
47 sandbox_dir: Path,
48 proj_dir: Path,
49 home_dir: Path,
50 packages_dir_: Path,
51 ):
53 # pylint: disable=too-many-arguments
54 # pylint: disable=too-many-positional-arguments
56 assert isinstance(sandbox_dir, Path)
57 assert isinstance(proj_dir, Path)
58 assert isinstance(home_dir, Path)
59 assert isinstance(packages_dir_, Path)
61 self._apio_runner = apio_runner_
62 self._sandbox_dir = sandbox_dir
63 self._proj_dir = proj_dir
64 self._home_dir = home_dir
65 self._packages_dir = packages_dir_
66 self._click_runner = CliRunner()
68 @property
69 def expired(self) -> bool:
70 """Returns true if this sandbox was expired."""
71 # -- This tests if this sandbox is still the active sandbox at the
72 # -- apio runner that creates it.
73 return self is not self._apio_runner.sandbox
75 @property
76 def sandbox_dir(self) -> Path:
77 """Returns the sandbox's dir."""
78 assert not self.expired, "Sandbox expired"
79 return self._sandbox_dir
81 @property
82 def proj_dir(self) -> Path:
83 """Returns the sandbox's apio project dir."""
84 assert not self.expired, "Sandbox expired"
85 return self._proj_dir
87 @property
88 def home_dir(self) -> Path:
89 """Returns the sandbox's apio home dir."""
90 assert not self.expired, "Sandbox expired"
91 return self._home_dir
93 @property
94 def packages_dir(self) -> Path:
95 """Returns the sandbox's apio packages dir."""
96 return self._packages_dir
98 def clear_packages(self):
99 """Clear the packages cache, in case a test needs a clean start."""
100 # -- Sanity check the path and delete.
101 assert "packages" in str(self.packages_dir).lower()
102 shutil.rmtree(self.packages_dir)
103 assert not self.packages_dir.exists()
105 def invoke_apio_cmd(
106 self,
107 cli,
108 args: List[str],
109 terminal_mode: bool = True,
110 in_subprocess: bool = False,
111 ) -> ApioResult:
112 """Invoke an apio command. in_subprocess run apios in a subprocess,
113 currently this suppresses colors because of the piping."""
115 print(f"\nInvoking apio command [{cli.name}], args={args}.")
117 # -- It's a good opportunity to flush the output so far.
118 sys.stdout.flush()
119 sys.stderr.flush()
121 # -- Check that this sandbox is still alive.
122 assert not self.expired, "Sandbox expired."
124 # -- Since we restore the env after invoking the apio command, we
125 # -- don't expect path changes by the command to survive here.
126 assert SANDBOX_MARKER not in os.environ["PATH"]
128 # -- Take a snapshot of the system env.
129 original_env = os.environ.copy()
131 # -- These two env vars are set when creating the context. Let's
132 # -- check that the test didn't corrupt them.
133 assert os.environ["APIO_HOME"] == str(self.home_dir)
135 # -- If True, force terminal mode, if False, forces pipe mode,
136 # -- otherwise auto which is pipe mode under pytest.
137 apio_console.configure(
138 terminal_mode=FORCE_TERMINAL if terminal_mode else FORCE_PIPE,
139 )
141 if in_subprocess:
142 # -- Invoke apio in a sub process.
143 print("Invoking apio in a sub process.")
144 process_result: CompletedProcess = subprocess.run(
145 [
146 sys.executable,
147 __main__.__file__,
148 ]
149 + args,
150 capture_output=True,
151 encoding="utf-8",
152 text=True,
153 check=False,
154 )
156 apio_result = ApioResult(
157 process_result.returncode,
158 (process_result.stdout or "") + (process_result.stderr or ""),
159 None,
160 )
162 else:
163 # -- Invoke the command in the same process using click.
164 print("Invoking apio in-process using click")
165 click_result: Result = self._click_runner.invoke(
166 prog_name="apio",
167 cli=cli,
168 args=args,
169 color=terminal_mode,
170 )
172 # -- Convert click result to apio result.
173 apio_result = ApioResult(
174 click_result.exit_code,
175 click_result.output,
176 click_result.exception,
177 )
179 # -- Dump to test log.
180 print(f"result.exit_code:{apio_result.exit_code}")
181 print("result.output:")
182 print(Result.output)
184 # -- Restore system env. Since apio commands tend to change vars
185 # -- such as PATH.
186 self.restore_system_env(original_env, "apio-command")
188 return apio_result
190 # -- List of default bad words for assert_ok(). All words should be
191 # -- lower case.
192 _DEFAULT_BAD_WORDS = ["error"]
194 def assert_result_ok(
195 self,
196 result: ApioResult,
197 bad_words: List[str] | Tuple[str, ...] = tuple(_DEFAULT_BAD_WORDS),
198 ):
199 """Check if apio command results where ok. Bad words is an optional
200 list of lower case strings strings if found in the lower case version
201 of the output text, trigger an error. The default is a tuple and not
202 a list to avoid pylint warning about unsafe default value"""
204 assert isinstance(result, ApioResult)
206 # -- It should return an exit code of 0: success
207 assert result.exit_code == 0, result.output
209 # -- There should be no exceptions raised
210 assert not result.exception
212 # -- Check for bad words.
213 lower_case_output = result.output.lower()
214 for bad_word in bad_words:
215 assert bad_word.islower(), bad_word
217 # -- Special case. For Xilinx arch it may be the case that
218 # -- the message contains the string "0 errors". It has the
219 # -- bad word 'error', but it is NOT an error
220 # -- We exclude that case
221 if "0 errors" not in lower_case_output:
223 # -- Check for no errors
224 assert bad_word not in lower_case_output, bad_word
226 def restore_system_env(
227 self, original_env: Dict[str, str], scope: str
228 ) -> None:
229 """Overwrites the existing sys.environ with the given dict. Vars
230 that are not in the dict are deleted and vars that have a different
231 value in the dict is updated. Can be called only within a
232 an apio sandbox."""
234 # -- Check that the sandbox not expired.
235 assert not self.expired, "Sandbox expired"
237 # -- NOTE: naively assigning the dict to os.environ will break
238 # -- os.environ since a simple dict doesn't update the underlying
239 # -- system env when it's mutated.
241 print(f"\nRestoring os.environ ({scope} scope):")
243 # -- Construct the union of the env and the dict var names.
244 all_var_names = set(os.environ.keys()).union(original_env.keys())
245 for name in all_var_names:
246 # Get the env and dict values. None if doesn't exist.
247 current_val = os.environ.get(name, None)
248 original_val = original_env.get(name, None)
249 # -- If values are not the same, update the env.
250 if current_val != original_val:
251 print(f" set ${name}={original_val} (was {current_val})")
252 if original_val is None:
253 os.environ.pop(name)
254 else:
255 os.environ[name] = original_val
257 # -- Sanity check. System env and the dict should be the same.
258 assert os.environ == original_env
260 def write_file(
261 self,
262 file: Union[str, Path],
263 text: Union[str, List[str]],
264 exists_ok=False,
265 ) -> None:
266 """Write text to given file. If text is a list, items are joined with
267 "\n". 'file' can be a string or a Path."""
269 assert exists_ok or not Path(file).exists(), f"File exists: {file}"
271 # -- If a list is given, join with '\n"
272 if isinstance(text, list):
273 text = "\n".join(text)
275 # -- Make dir(s) if needed.
276 Path(file).parent.mkdir(parents=True, exist_ok=True)
278 # -- Write.
279 with open(file, "w", encoding="utf-8") as f:
280 f.write(text)
282 def read_file(
283 self, file: Union[str, Path], lines_mode=False
284 ) -> Union[str, List[str]]:
285 """Read a text file. Returns a string with the text or if
286 lines_mode is True, a list with the individual lines (excluding
287 the \n delimiters)
288 """
290 # -- Read the text.
291 with open(file, "r", encoding="utf8") as f:
292 text = f.read()
294 # -- Split to lines if requested.
295 if lines_mode:
296 text = text.split("\n")
298 # -- All done
299 return text
301 def write_apio_ini(
302 self,
303 sections: Dict[str, Dict[str, str]] | None = None,
304 ):
305 """Write in the current directory an apio.ini file with given
306 section. If an apio.ini file already exists, overwrite it."""
308 assert isinstance(sections, dict)
310 # -- Construct output file path.
311 path = Path("apio.ini")
313 # -- List with text of each section.
314 sections_texts: List[str] = []
316 # -- Add the apio section if specified.
317 for section_header, section_options in sections.items():
318 lines = [section_header]
319 for name, value in section_options.items():
320 lines.append(f"{name} = {value}")
321 sections_texts.append("\n".join(lines))
323 # -- Join the sections with a blank line.
324 file_text = "\n\n".join(sections_texts)
326 # # -- Write the file.
327 self.write_file(path, file_text, exists_ok=True)
329 def write_default_apio_ini(self):
330 """Write in the local directory an apio.ini file with default values
331 for testing. If the file exists, it's overwritten."""
333 default_apio_ini = {
334 "[env:default]": {
335 "board": "alhambra-ii",
336 "top-module": "main",
337 }
338 }
339 self.write_apio_ini(default_apio_ini)
342class ApioRunner:
343 """Apio commands test helper. Provides an apio sandbox functionality
344 (disposable temp dirs and sys.environ restoration) as well as a few
345 utility functions. A typical test with the ApiRunner looks like this:
347 def test_my_cmd(apio_runner):
348 with apio_runner.in_sandbox() as sb:
350 <the test body>
351 """
353 def __init__(self, request: pytest.FixtureRequest):
354 print("*** creating ApioRunner")
355 assert isinstance(request, pytest.FixtureRequest)
357 print("\nOriginal env:")
358 pprint(dict(os.environ), width=80, sort_dicts=True)
359 print()
361 # -- Get a pytest directory for the apio packages cache. This will
362 # -- avoid reloading packages by each apio invocation.
363 cache = request.config.cache
364 self._packages_dir = cache.mkdir("apio-cached-packages")
366 # -- A CliRunner instance that is used for creating temp directories
367 # -- and to invoke apio commands.
368 self._request = request
370 # -- Indicate that we are not in a sandbox
371 self._sandbox: ApioSandbox | None = None
373 # -- A placeholder for a shared apio home that we may use in some
374 # -- of the sandboxes.
375 self._shared_apio_home: Path | None = None
377 # -- Register a cleanup method. It's called at the end of the
378 # -- apio_runner fixture scope.
379 request.addfinalizer(self._teardown)
381 def _teardown(self):
382 """Teardown at the end of the apio_runner fixture scope."""
383 if self._shared_apio_home: 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true
384 print(f"Deleting apio shared home {str(self._shared_apio_home)}")
385 assert "apio" in str(self._shared_apio_home)
386 shutil.rmtree(self._shared_apio_home)
387 self._shared_apio_home = None
389 @staticmethod
390 def _get_local_config_url() -> str:
391 """Returns a file:/ URL to the remote config file in the local depot.
392 This is used to set APIO_REMOTE_CONFIG_URL for testing to make sure
393 we test with the latest remote config in this change rather than with
394 the published remote config.
395 """
397 # -- Read apio/resources/config.jsonc so we can extract the remote
398 # -- config file name template to construct the local URL.
399 this_file_path = Path(__file__).resolve()
400 config_file_path = (
401 this_file_path.parent.parent / "apio/resources/config.jsonc"
402 )
404 with config_file_path.open(encoding="utf8") as file:
405 # -- Read the json with comments file
406 jsonc_text: str = file.read()
408 # -- Convert the jsonc to json by removing '//' comments.
409 json_text: str = jsonc.to_json(jsonc_text)
411 # -- Parse the json format!
412 json_data: dict = json.loads(json_text)
414 # -- Get the original remote config url.
415 url_str: str = json_data["remote-config-url"]
417 # -- Extract the file name part. E.g. 'apio-{major}.{minor}.x.jsonc'.
418 # -- The {major} and {minor} are placeholders for apio's major and
419 # -- minor version number which we don't resolve here.
420 config_file_path = urlparse(url_str).path
421 config_file_name = PurePosixPath(config_file_path).name
422 print(f"Config-file-name = {config_file_name}")
424 # -- Construct the path of the config file in this repo. We compute it
425 # -- based on the path of this conftest.py python file.
426 local_config_file = os.path.normpath(
427 os.path.join(
428 os.path.abspath(__file__),
429 "..",
430 "..",
431 "remote-config",
432 config_file_name,
433 )
434 )
435 # -- Convert the file path to a URL with a 'file://' form.
436 local_config_url = "file://" + str(local_config_file)
438 return local_config_url
440 @property
441 def sandbox(self) -> Optional[ApioSandbox]:
442 """Returns the sandbox object or None if not in a sandbox."""
443 return self._sandbox
445 @contextlib.contextmanager
446 def in_sandbox(self):
447 """Create an apio sandbox context manager that delete the temp dir
448 and restore the system env upon exist.
450 Upon return, the current directory is proj_dir.
451 """
452 # -- Make sure we don't try to nest sandboxes.
453 assert self._sandbox is None, "Already in a sandbox."
455 # -- Snapshot the system env.
456 original_env: Dict[str, str] = os.environ.copy()
458 # -- Snapshot the current directory.
459 original_cwd = os.getcwd()
461 # -- Create a temp sandbox dir that will be deleted on exit and
462 # -- change to it.
463 sandbox_dir = Path(tempfile.mkdtemp(prefix=SANDBOX_MARKER + "-"))
465 # -- Make the sandbox's project directory.
466 # -- Initially, we intentionally used a
467 # -- directory name with non ascii character to test
468 # -- that apio can handle it.
469 # proj_dir = sandbox_dir / "apio prój"
470 # -- I works ok for the architectures: ice40, ecp5, gowin
471 # -- BUT not for Xilinx. If the path contains a non-ascii character
472 # -- it complains
473 proj_dir = sandbox_dir / "apio proj"
474 proj_dir.mkdir()
475 os.chdir(proj_dir)
477 # -- Determine the project home dir.
478 home_dir = sandbox_dir / "apio-home"
480 if DEBUG: 480 ↛ 490line 480 didn't jump to line 490 because the condition on line 480 was always true
481 print()
482 print("--> apio sandbox:")
483 print(f" sandbox dir : {str(sandbox_dir)}")
484 print(f" apio proj dir : {str(proj_dir)}")
485 print(f" apio home dir : {str(home_dir)}")
486 print(f" apio packages dir : {str(self._packages_dir)}")
487 print()
489 # -- Register a sandbox objet to indicate that we are in a sandbox.
490 assert self._sandbox is None
491 self._sandbox = ApioSandbox(
492 self, sandbox_dir, proj_dir, home_dir, self._packages_dir
493 )
495 # -- Set the system env vars to inform ApioContext what are the
496 # -- home and packages dirs.
497 os.environ["APIO_HOME"] = str(home_dir)
498 os.environ["APIO_PACKAGES"] = str(self._packages_dir)
500 local_config_url = self._get_local_config_url()
501 print(f"Local config url: {local_config_url}")
503 # Sanity check to detect conflicts from prior URL settings.
504 assert os.environ.get("APIO_REMOTE_CONFIG_URL") is None
506 # Set the URL in the environment
507 os.environ["APIO_REMOTE_CONFIG_URL"] = local_config_url
509 # -- Reset the apio console, since we run multiple sandboxes in the
510 # -- same process.
511 apio_console.configure(
512 terminal_mode=FORCE_TERMINAL, theme_name="light"
513 )
515 try:
516 # -- This is the end of the context manager _entry part. The
517 # -- call to _exit will continue execution after the yield.
518 # -- Value is the sandbox object we pass to the user.
519 yield cast(ApioSandbox, self._sandbox)
521 finally:
522 # -- Here when the context manager exit, normally or through an
523 # -- exception.
525 # -- Restore the original system env.
526 self._sandbox.restore_system_env(original_env, "sandbox")
528 # -- Mark that we exited the sandbox. This expires the sandbox.
529 self._sandbox = None
531 # -- Change back to the original directory.
532 os.chdir(original_cwd)
534 # -- Delete the temp directory. This also deletes the apio home
535 # -- if it's not shared but doesn't touch it if we use a shared
536 # -- home.
537 shutil.rmtree(sandbox_dir)
539 print("\nSandbox deleted. ")
541 # -- Flush the output so far.
542 sys.stdout.flush()
543 sys.stderr.flush()
546@pytest.fixture(scope="module")
547def apio_runner(request):
548 """A pytest fixture that provides tests with a ApioRunner test
549 helper object. We use a 'module' scope so sandboxes can share the apio
550 home with other tests in the same file, if we choose to do so,
551 to reused previously installed packages.
552 """
553 assert isinstance(request, pytest.FixtureRequest)
554 return ApioRunner(request)