Coverage for tests/conftest.py: 95%
209 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-06 10:20 +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# -- This function is called by pytest. It adds the pytest --fast-only flag
32# -- which is is passed to tests such that slow tests can skip.
33# --
34# -- More info: https://docs.pytest.org/en/7.1.x/example/simple.html
35def pytest_addoption(parser: pytest.Parser):
36 """Register the --fast-only command line option when invoking pytest"""
38 # -- Option: --fast-only
39 # -- It causes slow tests to skip. Note that even in fast mode, the
40 # -- first test may need to update the packages cache which may take
41 # -- a minute or two.
42 parser.addoption(
43 "--fast-only", action="store_true", help="Run only the fast tests."
44 )
47@dataclass(frozen=True)
48class ApioResult:
49 """Represent the outcome of an apio invocation."""
51 exit_code: int
52 output: str # stdout only
53 exception: Any
56class ApioSandbox:
57 """Accessor for sandbox values. Available to the user inside an
58 ApioRunner sandbox scope."""
60 def __init__(
61 self,
62 apio_runner_: "ApioRunner",
63 sandbox_dir: Path,
64 proj_dir: Path,
65 home_dir: Path,
66 packages_dir_: Path,
67 ):
69 # pylint: disable=too-many-arguments
70 # pylint: disable=too-many-positional-arguments
72 assert isinstance(sandbox_dir, Path)
73 assert isinstance(proj_dir, Path)
74 assert isinstance(home_dir, Path)
75 assert isinstance(packages_dir_, Path)
77 self._apio_runner = apio_runner_
78 self._sandbox_dir = sandbox_dir
79 self._proj_dir = proj_dir
80 self._home_dir = home_dir
81 self._packages_dir = packages_dir_
82 self._click_runner = CliRunner()
84 @property
85 def expired(self) -> bool:
86 """Returns true if this sandbox was expired."""
87 # -- This tests if this sandbox is still the active sandbox at the
88 # -- apio runner that creates it.
89 return self is not self._apio_runner.sandbox
91 @property
92 def sandbox_dir(self) -> Path:
93 """Returns the sandbox's dir."""
94 assert not self.expired, "Sandbox expired"
95 return self._sandbox_dir
97 @property
98 def proj_dir(self) -> Path:
99 """Returns the sandbox's apio project dir."""
100 assert not self.expired, "Sandbox expired"
101 return self._proj_dir
103 @property
104 def home_dir(self) -> Path:
105 """Returns the sandbox's apio home dir."""
106 assert not self.expired, "Sandbox expired"
107 return self._home_dir
109 @property
110 def packages_dir(self) -> Path:
111 """Returns the sandbox's apio packages dir."""
112 return self._packages_dir
114 def clear_packages(self):
115 """Clear the packages cache, in case a test needs a clean start."""
116 # -- Sanity check the path and delete.
117 assert "packages" in str(self.packages_dir).lower()
118 shutil.rmtree(self.packages_dir)
119 assert not self.packages_dir.exists()
121 def invoke_apio_cmd(
122 self,
123 cli,
124 args: List[str],
125 terminal_mode: bool = True,
126 in_subprocess: bool = False,
127 ) -> ApioResult:
128 """Invoke an apio command. in_subprocess run apios in a subprocess,
129 currently this suppresses colors because of the piping."""
131 print(f"\nInvoking apio command [{cli.name}], args={args}.")
133 # -- It's a good opportunity to flush the output so far.
134 sys.stdout.flush()
135 sys.stderr.flush()
137 # -- Check that this sandbox is still alive.
138 assert not self.expired, "Sandbox expired."
140 # -- Since we restore the env after invoking the apio command, we
141 # -- don't expect path changes by the command to survive here.
142 assert SANDBOX_MARKER not in os.environ["PATH"]
144 # -- Take a snapshot of the system env.
145 original_env = os.environ.copy()
147 # -- These two env vars are set when creating the context. Let's
148 # -- check that the test didn't corrupt them.
149 assert os.environ["APIO_HOME"] == str(self.home_dir)
151 # -- If True, force terminal mode, if False, forces pipe mode,
152 # -- otherwise auto which is pipe mode under pytest.
153 apio_console.configure(
154 terminal_mode=FORCE_TERMINAL if terminal_mode else FORCE_PIPE,
155 )
157 if in_subprocess:
158 # -- Invoke apio in a sub process.
159 print("Invoking apio in a sub process.")
160 process_result: CompletedProcess = subprocess.run(
161 [
162 sys.executable,
163 __main__.__file__,
164 ]
165 + args,
166 capture_output=True,
167 encoding="utf-8",
168 text=True,
169 check=False,
170 )
172 apio_result = ApioResult(
173 process_result.returncode,
174 (process_result.stdout or "") + (process_result.stderr or ""),
175 None,
176 )
178 else:
179 # -- Invoke the command in the same process using click.
180 print("Invoking apio in-process using click")
181 click_result: Result = self._click_runner.invoke(
182 prog_name="apio",
183 cli=cli,
184 args=args,
185 color=terminal_mode,
186 )
188 # -- Convert click result to apio result.
189 apio_result = ApioResult(
190 click_result.exit_code,
191 click_result.output,
192 click_result.exception,
193 )
195 # -- Dump to test log.
196 print(f"result.exit_code:{apio_result.exit_code}")
197 print("result.output:")
198 print(Result.output)
200 # -- Restore system env. Since apio commands tend to change vars
201 # -- such as PATH.
202 self.restore_system_env(original_env, "apio-command")
204 return apio_result
206 # -- List of default bad words for assert_ok(). All words should be
207 # -- lower case.
208 _DEFAULT_BAD_WORDS = ["error"]
210 def assert_result_ok(
211 self,
212 result: ApioResult,
213 bad_words: List[str] | Tuple[str] = tuple(_DEFAULT_BAD_WORDS),
214 ):
215 """Check if apio command results where ok. Bad words is an optional
216 list of lower case strings strings if found in the lower case version
217 of the output text, trigger an error. The default is a tuple and not
218 a list to avoid pylint warning about unsafe default value"""
220 assert isinstance(result, ApioResult)
222 # -- It should return an exit code of 0: success
223 assert result.exit_code == 0, result.output
225 # -- There should be no exceptions raised
226 assert not result.exception
228 # -- Check for bad words.
229 lower_case_output = result.output.lower()
230 for bad_word in bad_words:
231 assert bad_word.islower(), bad_word
232 assert bad_word not in lower_case_output, bad_word
234 def restore_system_env(
235 self, original_env: Dict[str, str], scope: str
236 ) -> None:
237 """Overwrites the existing sys.environ with the given dict. Vars
238 that are not in the dict are deleted and vars that have a different
239 value in the dict is updated. Can be called only within a
240 an apio sandbox."""
242 # -- Check that the sandbox not expired.
243 assert not self.expired, "Sandbox expired"
245 # -- NOTE: naively assigning the dict to os.environ will break
246 # -- os.environ since a simple dict doesn't update the underlying
247 # -- system env when it's mutated.
249 print(f"\nRestoring os.environ ({scope} scope):")
251 # -- Construct the union of the env and the dict var names.
252 all_var_names = set(os.environ.keys()).union(original_env.keys())
253 for name in all_var_names:
254 # Get the env and dict values. None if doesn't exist.
255 env_val = os.environ.get(name, None)
256 dict_val = original_env.get(name, None)
257 # -- If values are not the same, update the env.
258 if env_val != dict_val:
259 print(f" set ${name}={dict_val} (was {env_val})")
260 if dict_val is None:
261 os.environ.pop(name)
262 else:
263 os.environ[name] = dict_val
265 # -- Sanity check. System env and the dict should be the same.
266 assert os.environ == original_env
268 def write_file(
269 self,
270 file: Union[str, Path],
271 text: Union[str, List[str]],
272 exists_ok=False,
273 ) -> None:
274 """Write text to given file. If text is a list, items are joined with
275 "\n". 'file' can be a string or a Path."""
277 assert exists_ok or not Path(file).exists(), f"File exists: {file}"
279 # -- If a list is given, join with '\n"
280 if isinstance(text, list):
281 text = "\n".join(text)
283 # -- Make dir(s) if needed.
284 Path(file).parent.mkdir(parents=True, exist_ok=True)
286 # -- Write.
287 with open(file, "w", encoding="utf-8") as f:
288 f.write(text)
290 def read_file(
291 self, file: Union[str, Path], lines_mode=False
292 ) -> Union[str, List[str]]:
293 """Read a text file. Returns a string with the text or if
294 lines_mode is True, a list with the individual lines (excluding
295 the \n delimiters)
296 """
298 # -- Read the text.
299 with open(file, "r", encoding="utf8") as f:
300 text = f.read()
302 # -- Split to lines if requested.
303 if lines_mode:
304 text = text.split("\n")
306 # -- All done
307 return text
309 def write_apio_ini(
310 self,
311 sections: Dict[str, Dict[str, str]] = None,
312 ):
313 """Write in the current directory an apio.ini file with given
314 section. If an apio.ini file already exists, overwrite it."""
316 assert isinstance(sections, dict)
318 # -- Construct output file path.
319 path = Path("apio.ini")
321 # -- List with text of each section.
322 sections_texts: List[str] = []
324 # -- Add the apio section if specified.
325 for section_header, section_options in sections.items():
326 lines = [section_header]
327 for name, value in section_options.items():
328 lines.append(f"{name} = {value}")
329 sections_texts.append("\n".join(lines))
331 # -- Join the sections with a blank line.
332 file_text = "\n\n".join(sections_texts)
334 # # -- Write the file.
335 self.write_file(path, file_text, exists_ok=True)
337 def write_default_apio_ini(self):
338 """Write in the local directory an apio.ini file with default values
339 for testing. If the file exists, it's overwritten."""
341 default_apio_ini = {
342 "[env:default]": {
343 "board": "alhambra-ii",
344 "top-module": "main",
345 }
346 }
347 self.write_apio_ini(default_apio_ini)
350class ApioRunner:
351 """Apio commands test helper. Provides an apio sandbox functionality
352 (disposable temp dirs and sys.environ restoration) as well as a few
353 utility functions. A typical test with the ApiRunner looks like this:
355 def test_my_cmd(apio_runner):
356 with apio_runner.in_sandbox() as sb:
358 <the test body>
359 """
361 def __init__(self, request: pytest.FixtureRequest):
362 print("*** creating ApioRunner")
363 assert isinstance(request, pytest.FixtureRequest)
365 print("\nOriginal env:")
366 pprint(dict(os.environ), width=80, sort_dicts=True)
367 print()
369 # -- Get a pytest directory for the apio packages cache. This will
370 # -- avoid reloading packages by each apio invocation.
371 cache = request.config.cache
372 self._packages_dir = cache.mkdir("apio-cached-packages")
374 # -- A CliRunner instance that is used for creating temp directories
375 # -- and to invoke apio commands.
376 self._request = request
378 # -- Indicate that we are not in a sandbox
379 self._sandbox: ApioSandbox = None
381 # -- A placeholder for a shared apio home that we may use in some
382 # -- of the sandboxes.
383 self._shared_apio_home: Path = None
385 # -- Register a cleanup method. It's called at the end of the
386 # -- apio_runner fixture scope.
387 request.addfinalizer(self._teardown)
389 def _teardown(self):
390 """Teardown at the end of the apio_runner fixture scope."""
391 if self._shared_apio_home: 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 print(f"Deleting apio shared home {str(self._shared_apio_home)}")
393 assert "apio" in str(self._shared_apio_home)
394 shutil.rmtree(self._shared_apio_home)
395 self._shared_apio_home = None
397 @staticmethod
398 def _get_local_config_url() -> str:
399 """Returns a file:/ URL to the remote config file in the local depot.
400 This is used to set APIO_REMOTE_CONFIG_URL for testing to make sure
401 we test with the latest remote config in this change rather than with
402 the published remote config.
403 """
405 # -- Read apio/resources/config.jsonc so we can extract the remote
406 # -- config file name template to construct the local URL.
407 this_file_path = Path(__file__).resolve()
408 config_file_path = (
409 this_file_path.parent.parent / "apio/resources/config.jsonc"
410 )
412 with config_file_path.open(encoding="utf8") as file:
413 # -- Read the json with comments file
414 jsonc_text: str = file.read()
416 # -- Convert the jsonc to json by removing '//' comments.
417 json_text: str = jsonc.to_json(jsonc_text)
419 # -- Parse the json format!
420 json_data: dict = json.loads(json_text)
422 # -- Get the original remote config url.
423 url_str: str = json_data["remote-config-url"]
425 # -- Extract the file name part. E.g. 'apio-{V}.jsonc. The '{V}' marker
426 # -- is a place holder for the apio version which we don't resolve
427 # -- here.
428 config_file_path = urlparse(url_str).path
429 config_file_name = PurePosixPath(config_file_path).name
430 print(f"Config-file-name = {config_file_name}")
432 # -- Construct the path of the config file in this repo. We compute it
433 # -- based on the path of this conftest.py python file.
434 local_config_file = os.path.normpath(
435 os.path.join(
436 os.path.abspath(__file__),
437 "..",
438 "..",
439 "remote-config",
440 config_file_name,
441 )
442 )
443 # -- Convert the file path to a URL with a 'file://' form.
444 local_config_url = "file://" + str(local_config_file)
446 return local_config_url
448 @property
449 def sandbox(self) -> Optional[ApioSandbox]:
450 """Returns the sandbox object or None if not in a sandbox."""
451 return self._sandbox
453 @contextlib.contextmanager
454 def in_sandbox(self):
455 """Create an apio sandbox context manager that delete the temp dir
456 and restore the system env upon exist.
458 Upon return, the current directory is proj_dir.
459 """
460 # -- Make sure we don't try to nest sandboxes.
461 assert self._sandbox is None, "Already in a sandbox."
463 # -- Snapshot the system env.
464 original_env: Dict[str, str] = os.environ.copy()
466 # -- Snapshot the current directory.
467 original_cwd = os.getcwd()
469 # -- Create a temp sandbox dir that will be deleted on exit and
470 # -- change to it.
471 sandbox_dir = Path(tempfile.mkdtemp(prefix=SANDBOX_MARKER + "-"))
473 # -- Make the sandbox's project directory. We intentionally use a
474 # -- directory name with a space and a non ascii character to test
475 # -- that apio can handle it.
476 proj_dir = sandbox_dir / "apio prój"
477 proj_dir.mkdir()
478 os.chdir(proj_dir)
480 # -- Determine the project home dir.
481 home_dir = sandbox_dir / "apio-home"
483 if DEBUG: 483 ↛ 493line 483 didn't jump to line 493 because the condition on line 483 was always true
484 print()
485 print("--> apio sandbox:")
486 print(f" sandbox dir : {str(sandbox_dir)}")
487 print(f" apio proj dir : {str(proj_dir)}")
488 print(f" apio home dir : {str(home_dir)}")
489 print(f" apio packages dir : {str(self._packages_dir)}")
490 print()
492 # -- Register a sandbox objet to indicate that we are in a sandbox.
493 assert self._sandbox is None
494 self._sandbox = ApioSandbox(
495 self, sandbox_dir, proj_dir, home_dir, self._packages_dir
496 )
498 # -- Set the system env vars to inform ApioContext what are the
499 # -- home and packages dirs.
500 os.environ["APIO_HOME"] = str(home_dir)
501 os.environ["APIO_PACKAGES"] = str(self._packages_dir)
503 local_config_url = self._get_local_config_url()
504 print(f"Local config url: {local_config_url}")
506 # Sanity check to detect conflicts from prior URL settings.
507 assert os.environ.get("APIO_REMOTE_CONFIG_URL") is None
509 # Set the URL in the environment
510 os.environ["APIO_REMOTE_CONFIG_URL"] = local_config_url
512 # -- Reset the apio console, since we run multiple sandboxes in the
513 # -- same process.
514 apio_console.configure(
515 terminal_mode=FORCE_TERMINAL, theme_name="light"
516 )
518 try:
519 # -- This is the end of the context manager _entry part. The
520 # -- call to _exit will continue execution after the yield.
521 # -- Value is the sandbox object we pass to the user.
522 yield cast(ApioSandbox, self._sandbox)
524 finally:
525 # -- Here when the context manager exit, normally or through an
526 # -- exception.
528 # -- Restore the original system env.
529 self._sandbox.restore_system_env(original_env, "sandbox")
531 # -- Mark that we exited the sandbox. This expires the sandbox.
532 self._sandbox = None
534 # -- Change back to the original directory.
535 os.chdir(original_cwd)
537 # -- Delete the temp directory. This also deletes the apio home
538 # -- if it's not shared but doesn't touch it if we use a shared
539 # -- home.
540 shutil.rmtree(sandbox_dir)
542 print("\nSandbox deleted. ")
544 # -- Flush the output so far.
545 sys.stdout.flush()
546 sys.stderr.flush()
548 def skip_test_if_fast_only(self):
549 """The calling test is skipped if running in --fast-only mode.
550 Should be called from slow tests. The fast/slow classification of tests
551 should be done after the packages cache was filled (automatically).
552 """
553 if self._request.config.getoption("--fast-only"): 553 ↛ 554line 553 didn't jump to line 554 because the condition on line 553 was never true
554 pytest.skip("slow test")
557@pytest.fixture(scope="module")
558def apio_runner(request):
559 """A pytest fixture that provides tests with a ApioRunner test
560 helper object. We use a 'module' scope so sandboxes can share the apio
561 home with other tests in the same file, if we choose to do so,
562 to reused previously installed packages.
563 """
564 assert isinstance(request, pytest.FixtureRequest)
565 return ApioRunner(request)