Coverage for tests/conftest.py: 96%

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()

104

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."""

114

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

116

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

118 sys.stdout.flush()

119 sys.stderr.flush()

120

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

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

123

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"]

127

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

129 original_env = os.environ.copy()

130

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)

134

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 )

140

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 )

155

156 apio_result = ApioResult(

157 process_result.returncode,

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

159 None,

160 )

161

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 )

171

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 )

178

179 # -- Dump to test log.

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

181 print("result.output:")

182 print(Result.output)

183

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

185 # -- such as PATH.

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

187

188 return apio_result

189

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

191 # -- lower case.

192 _DEFAULT_BAD_WORDS = ["error"]

193

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"""

203

204 assert isinstance(result, ApioResult)

205

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

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

208

209 # -- There should be no exceptions raised

210 assert not result.exception

211

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

216 assert bad_word not in lower_case_output, bad_word

217

218 def restore_system_env(

219 self, original_env: Dict[str, str], scope: str

220 ) -> None:

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

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

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

224 an apio sandbox."""

225

226 # -- Check that the sandbox not expired.

227 assert not self.expired, "Sandbox expired"

228

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

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

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

232

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

234

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

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

237 for name in all_var_names:

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

239 env_val = os.environ.get(name, None)

240 dict_val = original_env.get(name, None)

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

242 if env_val != dict_val:

243 print(f" set ${name}={dict_val} (was {env_val})")

244 if dict_val is None:

245 os.environ.pop(name)

246 else:

247 os.environ[name] = dict_val

248

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

250 assert os.environ == original_env

251

252 def write_file(

253 self,

254 file: Union[str, Path],

255 text: Union[str, List[str]],

256 exists_ok=False,

257 ) -> None:

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

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

260

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

262

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

264 if isinstance(text, list):

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

266

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

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

269

270 # -- Write.

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

272 f.write(text)

273

274 def read_file(

275 self, file: Union[str, Path], lines_mode=False

276 ) -> Union[str, List[str]]:

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

278 lines_mode is True, a list with the individual lines (excluding

279 the \n delimiters)

280 """

281

282 # -- Read the text.

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

284 text = f.read()

285

286 # -- Split to lines if requested.

287 if lines_mode:

288 text = text.split("\n")

289

290 # -- All done

291 return text

292

293 def write_apio_ini(

294 self,

295 sections: Dict[str, Dict[str, str]] = None,

296 ):

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

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

299

300 assert isinstance(sections, dict)

301

302 # -- Construct output file path.

303 path = Path("apio.ini")

304

305 # -- List with text of each section.

306 sections_texts: List[str] = []

307

308 # -- Add the apio section if specified.

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

310 lines = [section_header]

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

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

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

314

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

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

317

318 # # -- Write the file.

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

320

321 def write_default_apio_ini(self):

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

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

324

325 default_apio_ini = {

326 "[env:default]": {

327 "board": "alhambra-ii",

328 "top-module": "main",

329 }

330 }

331 self.write_apio_ini(default_apio_ini)

332

333

334class ApioRunner:

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

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

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

338

339 def test_my_cmd(apio_runner):

340 with apio_runner.in_sandbox() as sb:

341

342 <the test body>

343 """

344

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

346 print("*** creating ApioRunner")

347 assert isinstance(request, pytest.FixtureRequest)

348

349 print("\nOriginal env:")

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

351 print()

352

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

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

355 cache = request.config.cache

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

357

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

359 # -- and to invoke apio commands.

360 self._request = request

361

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

363 self._sandbox: ApioSandbox = None

364

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

366 # -- of the sandboxes.

367 self._shared_apio_home: Path = None

368

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

370 # -- apio_runner fixture scope.

371 request.addfinalizer(self._teardown)

372

373 def _teardown(self):

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

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

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

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

378 shutil.rmtree(self._shared_apio_home)

379 self._shared_apio_home = None

380

381 @staticmethod

382 def _get_local_config_url() -> str:

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

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

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

386 the published remote config.

387 """

388

389 # -- Read apio/resources/config.jsonc so we can extract the remote

390 # -- config file name template to construct the local URL.

391 this_file_path = Path(__file__).resolve()

392 config_file_path = (

393 this_file_path.parent.parent / "apio/resources/config.jsonc"

394 )

395

396 with config_file_path.open(encoding="utf8") as file:

397 # -- Read the json with comments file

398 jsonc_text: str = file.read()

399

400 # -- Convert the jsonc to json by removing '//' comments.

401 json_text: str = jsonc.to_json(jsonc_text)

402

403 # -- Parse the json format!

404 json_data: dict = json.loads(json_text)

405

406 # -- Get the original remote config url.

407 url_str: str = json_data["remote-config-url"]

408

409 # -- Extract the file name part. E.g. 'apio-{major}.{minor}.x.jsonc'.

410 # -- The {major} and {minor} are placeholders for apio's major and

411 # -- minor version number which we don't resolve here.

412 config_file_path = urlparse(url_str).path

413 config_file_name = PurePosixPath(config_file_path).name

414 print(f"Config-file-name = {config_file_name}")

415

416 # -- Construct the path of the config file in this repo. We compute it

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

418 local_config_file = os.path.normpath(

419 os.path.join(

420 os.path.abspath(__file__),

421 "..",

422 "..",

423 "remote-config",

424 config_file_name,

425 )

426 )

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

428 local_config_url = "file://" + str(local_config_file)

429

430 return local_config_url

431

432 @property

433 def sandbox(self) -> Optional[ApioSandbox]:

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

435 return self._sandbox

436

437 @contextlib.contextmanager

438 def in_sandbox(self):

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

440 and restore the system env upon exist.

441

442 Upon return, the current directory is proj_dir.

443 """

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

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

446

447 # -- Snapshot the system env.

448 original_env: Dict[str, str] = os.environ.copy()

449

450 # -- Snapshot the current directory.

451 original_cwd = os.getcwd()

452

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

454 # -- change to it.

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

456

457 # -- Make the sandbox's project directory. We intentionally use a

458 # -- directory name with a space and a non ascii character to test

459 # -- that apio can handle it.

460 proj_dir = sandbox_dir / "apio prój"

461 proj_dir.mkdir()

462 os.chdir(proj_dir)

463

464 # -- Determine the project home dir.

465 home_dir = sandbox_dir / "apio-home"

466

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

468 print()

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

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

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

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

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

474 print()

475

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

477 assert self._sandbox is None

478 self._sandbox = ApioSandbox(

479 self, sandbox_dir, proj_dir, home_dir, self._packages_dir

480 )

481

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

483 # -- home and packages dirs.

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

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

486

487 local_config_url = self._get_local_config_url()

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

489

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

491 assert os.environ.get("APIO_REMOTE_CONFIG_URL") is None

492

493 # Set the URL in the environment

494 os.environ["APIO_REMOTE_CONFIG_URL"] = local_config_url

495

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

497 # -- same process.

498 apio_console.configure(

499 terminal_mode=FORCE_TERMINAL, theme_name="light"

500 )

501

502 try:

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

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

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

506 yield cast(ApioSandbox, self._sandbox)

507

508 finally:

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

510 # -- exception.

511

512 # -- Restore the original system env.

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

514

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

516 self._sandbox = None

517

518 # -- Change back to the original directory.

519 os.chdir(original_cwd)

520

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

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

523 # -- home.

524 shutil.rmtree(sandbox_dir)

525

526 print("\nSandbox deleted. ")

527

528 # -- Flush the output so far.

529 sys.stdout.flush()

530 sys.stderr.flush()

531

532

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

534def apio_runner(request):

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

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

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

538 to reused previously installed packages.

539 """

540 assert isinstance(request, pytest.FixtureRequest)

541 return ApioRunner(request)

Coverage for tests / conftest.py: 96%

204 statements