diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9baaa9c19..298c38681 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -3,7 +3,8 @@ name: CI on: pull_request: types: [opened, synchronize, reopened] - branches: ['*'] + branches: ['**'] + workflow_dispatch: {} jobs: test: diff --git a/.gitignore b/.gitignore index 273ca0eb2..dec2cc930 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ __pycache__/ *.egg-info/ # Virtual environments +.venv/ .test_venv/ .cvs_venv/ .ruff_venv/ diff --git a/cvs/runners/_base_runner.py b/cvs/runners/_base_runner.py index 714c2bed0..80a9a2e1f 100644 --- a/cvs/runners/_base_runner.py +++ b/cvs/runners/_base_runner.py @@ -184,11 +184,17 @@ def execute(self, **kwargs) -> RunResult: try: # Setup phase log.info(f"Setting up {self.__class__.__name__}...") - if not self.setup(): + try: + setup_ok = self.setup() + finally: + # Set even if setup() raises or partially succeeds, so the + # finally block below still tears down resources a + # partially-successful setup already created. + self._setup_complete = True + if not setup_ok: return RunResult( status=RunStatus.FAILED, start_time=start_time, end_time=time.time(), error_message="Setup failed" ) - self._setup_complete = True # Run phase log.info(f"Running {self.__class__.__name__}...") diff --git a/cvs/runners/unittests/__init__.py b/cvs/runners/unittests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/cvs/runners/unittests/test_base_runner.py b/cvs/runners/unittests/test_base_runner.py new file mode 100644 index 000000000..890e0d05a --- /dev/null +++ b/cvs/runners/unittests/test_base_runner.py @@ -0,0 +1,91 @@ +""" +Unit tests for BaseRunner.execute()'s setup -> run -> teardown lifecycle. + +Copyright 2025 Advanced Micro Devices, Inc. +All rights reserved. +""" + +import unittest + +from cvs.runners._base_runner import BaseRunner, RunConfig, RunResult, RunStatus + + +class _FakeRunner(BaseRunner): + """Minimal concrete BaseRunner for exercising execute().""" + + def __init__(self, config, setup_return=True, setup_raises=None, run_return=None, run_raises=None): + super().__init__(config) + self._setup_return = setup_return + self._setup_raises = setup_raises + self._run_return = run_return + self._run_raises = run_raises + self.teardown_calls = 0 + + def setup(self) -> bool: + if self._setup_raises is not None: + raise self._setup_raises + return self._setup_return + + def run(self, **kwargs) -> RunResult: + if self._run_raises is not None: + raise self._run_raises + return self._run_return + + def teardown(self) -> bool: + self.teardown_calls += 1 + return True + + +def _config() -> RunConfig: + return RunConfig(nodes=["10.0.0.1"], username="testuser") + + +class TestExecuteTeardownLifecycle(unittest.TestCase): + def test_teardown_runs_after_successful_setup_and_run(self): + run_result = RunResult(status=RunStatus.COMPLETED, start_time=0, end_time=1) + runner = _FakeRunner(_config(), setup_return=True, run_return=run_result) + + result = runner.execute() + + self.assertEqual(result.status, RunStatus.COMPLETED) + self.assertEqual(runner.teardown_calls, 1) + + def test_teardown_runs_when_setup_fails(self): + runner = _FakeRunner(_config(), setup_return=False) + + result = runner.execute() + + self.assertEqual(result.status, RunStatus.FAILED) + self.assertEqual(result.error_message, "Setup failed") + self.assertEqual(runner.teardown_calls, 1) + + def test_teardown_runs_when_setup_raises(self): + runner = _FakeRunner(_config(), setup_raises=RuntimeError("setup exploded")) + + result = runner.execute() + + self.assertEqual(result.status, RunStatus.FAILED) + self.assertIn("setup exploded", result.error_message) + self.assertEqual(runner.teardown_calls, 1) + + def test_teardown_runs_when_run_raises(self): + runner = _FakeRunner(_config(), setup_return=True, run_raises=RuntimeError("boom")) + + result = runner.execute() + + self.assertEqual(result.status, RunStatus.FAILED) + self.assertIn("boom", result.error_message) + self.assertEqual(runner.teardown_calls, 1) + + def test_teardown_not_run_before_setup_attempted(self): + runner = _FakeRunner( + _config(), setup_return=True, run_return=RunResult(status=RunStatus.COMPLETED, start_time=0, end_time=1) + ) + + self.assertFalse(runner._setup_complete) + runner.execute() + self.assertTrue(runner._setup_complete) + + +if __name__ == "__main__": + unittest.main()