diff --git a/repository_service_tuf_worker/repository.py b/repository_service_tuf_worker/repository.py index c4212739..21846c4b 100644 --- a/repository_service_tuf_worker/repository.py +++ b/repository_service_tuf_worker/repository.py @@ -17,10 +17,7 @@ from celery.exceptions import ChordError from celery.result import AsyncResult, states from dynaconf.loaders import redis_loader -from securesystemslib.exceptions import ( # type: ignore - StorageError, - UnverifiedSignatureError, -) +from securesystemslib.exceptions import UnverifiedSignatureError from securesystemslib.signer import Signature, SSlibKey from tuf.api.exceptions import ( BadVersionNumberError, @@ -960,7 +957,7 @@ def remove_targets( }, ) - def _run_online_roles_bump(self, force: Optional[bool] = False) -> bool: + def _run_online_roles_bump(self, force: Optional[bool] = False): """ Bumps version and expiration date of all online roles (`Targets`, `Succinct Delegated` targets roles, `Timestamp` and `Snapshot`). @@ -975,12 +972,7 @@ def _run_online_roles_bump(self, force: Optional[bool] = False) -> bool: force: force all target roles bump even if they have more than `self._hours_before_expire` hours to expire. """ - try: - targets: Metadata = self._storage_backend.get(Targets.type) - except StorageError: - logging.error(f"{Targets.type} not found, not bumping.") - return False - + targets: Metadata = self._storage_backend.get(Targets.type) timestamp: Metadata snapshot_bump = False if self._settings.get_fresh("TARGETS_ONLINE_KEY") is None: @@ -1040,9 +1032,7 @@ def _run_online_roles_bump(self, force: Optional[bool] = False) -> bool: f"[scheduled bump] Timestamp version bumped: {timestamp_v}" ) - return True - - def bump_snapshot(self, force: Optional[bool] = False) -> bool: + def bump_snapshot(self, force: Optional[bool] = False): """ Bumps version and expiration date of TUF 'snapshot' role metadata. @@ -1057,12 +1047,7 @@ def bump_snapshot(self, force: Optional[bool] = False) -> bool: expire (`self._hours_before_expire`) """ - try: - snapshot = self._storage_backend.get(Snapshot.type) - except StorageError: - logging.error(f"{Snapshot.type} not found, not bumping.") - return False - + snapshot = self._storage_backend.get(Snapshot.type) if (snapshot.signed.expires - datetime.now()) < timedelta( hours=self._hours_before_expire ) or force: @@ -1084,8 +1069,6 @@ def bump_snapshot(self, force: Optional[bool] = False) -> bool: f"{self._hours_before_expire} hour, skipping" ) - return True - def bump_online_roles(self, force: Optional[bool] = False) -> bool: """ Bump online roles (Snapshot, Timestamp, Targets and BINS). diff --git a/tests/unit/tuf_repository_service_worker/test_repository.py b/tests/unit/tuf_repository_service_worker/test_repository.py index 468aa1d4..c8d49af6 100644 --- a/tests/unit/tuf_repository_service_worker/test_repository.py +++ b/tests/unit/tuf_repository_service_worker/test_repository.py @@ -10,6 +10,7 @@ import pytest from celery.exceptions import ChordError from celery.result import states +from securesystemslib.exceptions import StorageError from tuf.api.metadata import ( Metadata, MetaFile, @@ -2143,8 +2144,7 @@ def test__run_online_roles_bump_only_expired( ) ) ) - result = test_repo._run_online_roles_bump() - assert result is True + test_repo._run_online_roles_bump() assert test_repo._storage_backend.get.calls == [ pretend.call(Targets.type), pretend.call("bin-a"), @@ -2198,8 +2198,7 @@ def test__run_online_roles_bump_force( ) ) ) - result = test_repo._run_online_roles_bump(force=True) - assert result is True + test_repo._run_online_roles_bump(force=True) assert test_repo._storage_backend.get.calls == [ pretend.call(Targets.type), ] @@ -2262,10 +2261,9 @@ def test__run_online_roles_bump_target_targets_online_key_config_false( ) ) ) - result = test_repo._run_online_roles_bump() + test_repo._run_online_roles_bump() msg = "targets don't use online key, skipping 'Targets' role" assert msg == caplog.messages[0] - assert result is True assert test_repo._storage_backend.get.calls == [ pretend.call(Targets.type), pretend.call("bin-a"), @@ -2315,10 +2313,9 @@ def test__run_online_roles_bump_warning_missing_config( ) ) ) - result = test_repo._run_online_roles_bump() + test_repo._run_online_roles_bump() msg = "No configuration found for TARGETS_ONLINE_KEY" assert msg == caplog.messages[0] - assert result is True assert test_repo._storage_backend.get.calls == [ pretend.call(Targets.type), pretend.call("bin-a"), @@ -2355,8 +2352,7 @@ def test__run_online_roles_bump_no_changes(self, test_repo, caplog): else fake_bins ) - result = test_repo._run_online_roles_bump() - assert result is True + test_repo._run_online_roles_bump() assert test_repo._storage_backend.get.calls == [ pretend.call(Targets.type), pretend.call("bin-a"), @@ -2370,11 +2366,11 @@ def test__run_online_roles_bump_no_changes(self, test_repo, caplog): def test__run_online_roles_bump_StorageError(self, test_repo): test_repo._storage_backend.get = pretend.raiser( - repository.StorageError("Overwrite it") + StorageError("Overwrite it") ) - result = test_repo._run_online_roles_bump() - assert result is False + with pytest.raises(StorageError): + test_repo._run_online_roles_bump() def test_bump_snapshot(self, test_repo, mocked_datetime): fake_snapshot = pretend.stub( @@ -2400,8 +2396,7 @@ def test_bump_snapshot(self, test_repo, mocked_datetime): ) ) - result = test_repo.bump_snapshot() - assert result is True + test_repo.bump_snapshot() assert test_repo._storage_backend.get.calls == [ pretend.call("snapshot") ] @@ -2422,8 +2417,7 @@ def test_bump_snapshot_unexpired(self, test_repo): lambda *a: fake_snapshot ) - result = test_repo.bump_snapshot() - assert result is True + test_repo.bump_snapshot() assert test_repo._storage_backend.get.calls == [ pretend.call("snapshot") ] @@ -2466,8 +2460,7 @@ def test_bump_snapshot_check_force_is_acknowledged( ) ) - result = test_repo.bump_snapshot(force=True) - assert result is True + test_repo.bump_snapshot(force=True) assert test_repo._storage_backend.get.calls == [ pretend.call(Snapshot.type) ] @@ -2477,12 +2470,9 @@ def test_bump_snapshot_check_force_is_acknowledged( ] def test_bump_snapshot_not_found(self, test_repo): - test_repo._storage_backend.get = pretend.raiser( - repository.StorageError - ) - - result = test_repo.bump_snapshot() - assert result is False + test_repo._storage_backend.get = pretend.raiser(StorageError) + with pytest.raises(StorageError): + test_repo.bump_snapshot() def test_bump_online_roles(self, monkeypatch, test_repo): @contextmanager