(self)
| 38 | self.assertEqual(v, fv.get_flag_value(k, None)) |
| 39 | |
| 40 | def test_execute(self): |
| 41 | class FailNJob(Job): |
| 42 | """A dummy job that fails execute a fixed number of times.""" |
| 43 | |
| 44 | @config_class |
| 45 | class Config(Job.Config): |
| 46 | n: Required[int] = REQUIRED |
| 47 | |
| 48 | def __init__(self, cfg): |
| 49 | super().__init__(cfg) |
| 50 | self.i = 0 |
| 51 | self.delete_called = False |
| 52 | |
| 53 | def _delete(self): |
| 54 | self.delete_called = True |
| 55 | |
| 56 | def _execute(self): |
| 57 | if self.i < self.config.n: |
| 58 | self.i += 1 |
| 59 | raise ValueError(f"Deliberate failure {self.i}") |
| 60 | |
| 61 | fv = _mock_flag_values() |
| 62 | cfg = FailNJob.from_flags(fv) |
| 63 | cfg.set(max_tries=3, retry_interval=0.1) |
| 64 | |
| 65 | # Test max tries. |
| 66 | job = cfg.set(n=cfg.max_tries).instantiate() |
| 67 | with self.assertRaisesRegex(ValueError, "Failed to execute"): |
| 68 | job.execute() |
| 69 | self.assertEqual(cfg.max_tries, job.i) |
| 70 | self.assertTrue(job.delete_called) |
| 71 | |
| 72 | # Test success. |
| 73 | job = cfg.set(n=cfg.max_tries - 1).instantiate() |
| 74 | job.execute() |
| 75 | self.assertEqual(cfg.n, job.i) |
| 76 | self.assertFalse(job.delete_called) |
| 77 | |
| 78 | |
| 79 | class UtilTest(TestCase): |
nothing calls this directly
no test coverage detected