diff --git a/tests/test_updater_consistent_snapshot.py b/tests/test_updater_consistent_snapshot.py index abf6fb4a9b..cf272fcb42 100644 --- a/tests/test_updater_consistent_snapshot.py +++ b/tests/test_updater_consistent_snapshot.py @@ -126,23 +126,20 @@ def test_top_level_roles_update( ) -> None: # Test if the client fetches and stores metadata files with the # correct version prefix, depending on 'consistent_snapshot' config - try: - consistent_snapshot: bool = test_case_data["consistent_snapshot"] - exp_calls: list[Any] = test_case_data["calls"] + consistent_snapshot: bool = test_case_data["consistent_snapshot"] + exp_calls: list[Any] = test_case_data["calls"] - self.setup_subtest(consistent_snapshot) - updater = self._init_updater() + self.setup_subtest(consistent_snapshot) + updater = self._init_updater() - # cleanup fetch tracker metadata - self.sim.fetch_tracker.metadata.clear() - updater.refresh() + # cleanup fetch tracker metadata + self.sim.fetch_tracker.metadata.clear() + updater.refresh() - # metadata files are fetched with the expected version (or None) - self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) - # metadata files are always persisted without a version prefix - self._assert_metadata_files_exist(TOP_LEVEL_ROLE_NAMES) - finally: - self.teardown_subtest() + # metadata files are fetched with the expected version (or None) + self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) + # metadata files are always persisted without a version prefix + self._assert_metadata_files_exist(TOP_LEVEL_ROLE_NAMES) delegated_roles_data = { "consistent_snaphot disabled": { @@ -161,35 +158,30 @@ def test_delegated_roles_update( ) -> None: # Test if the client fetches and stores delegated metadata files with # the correct version prefix, depending on 'consistent_snapshot' config - try: - consistent_snapshot: bool = test_case_data["consistent_snapshot"] - exp_version: int | None = test_case_data["expected_version"] - rolenames = ["role1", "..", "."] - exp_calls = [(role, exp_version) for role in rolenames] - - self.setup_subtest(consistent_snapshot) - # Add new delegated targets - spec_version = ".".join(SPECIFICATION_VERSION) - for role in rolenames: - delegated_role = DelegatedRole(role, [], 1, False, ["*"], None) - targets = Targets( - 1, spec_version, self.sim.safe_expiry, {}, None - ) - self.sim.add_delegation("targets", delegated_role, targets) - self.sim.update_snapshot() - updater = self._init_updater() - updater.refresh() - - # cleanup fetch tracker metadata - self.sim.fetch_tracker.metadata.clear() - # trigger updater to fetch the delegated metadata - updater.get_targetinfo("anything") - # metadata files are fetched with the expected version (or None) - self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) - # metadata files are always persisted without a version prefix - self._assert_metadata_files_exist(rolenames) - finally: - self.teardown_subtest() + consistent_snapshot: bool = test_case_data["consistent_snapshot"] + exp_version: int | None = test_case_data["expected_version"] + rolenames = ["role1", "..", "."] + exp_calls = [(role, exp_version) for role in rolenames] + + self.setup_subtest(consistent_snapshot) + # Add new delegated targets + spec_version = ".".join(SPECIFICATION_VERSION) + for role in rolenames: + delegated_role = DelegatedRole(role, [], 1, False, ["*"], None) + targets = Targets(1, spec_version, self.sim.safe_expiry, {}, None) + self.sim.add_delegation("targets", delegated_role, targets) + self.sim.update_snapshot() + updater = self._init_updater() + updater.refresh() + + # cleanup fetch tracker metadata + self.sim.fetch_tracker.metadata.clear() + # trigger updater to fetch the delegated metadata + updater.get_targetinfo("anything") + # metadata files are fetched with the expected version (or None) + self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) + # metadata files are always persisted without a version prefix + self._assert_metadata_files_exist(rolenames) targets_download_data = { "consistent_snaphot disabled": { @@ -217,40 +209,37 @@ def test_download_targets(self, test_case_data: dict[str, Any]) -> None: # Test if the client fetches and stores target files with # the correct hash prefix, depending on 'consistent_snapshot' # and 'prefix_targets_with_hash' config - try: - consistent_snapshot: bool = test_case_data["consistent_snapshot"] - prefix_targets_with_hash: bool = test_case_data["prefix_targets"] - hash_algo: str | None = test_case_data["hash_algo"] - targetpaths: list[str] = test_case_data["targetpaths"] - - self.setup_subtest(consistent_snapshot, prefix_targets_with_hash) - # Add targets to repository - for targetpath in targetpaths: - self.sim.targets.version += 1 - self.sim.add_target("targets", b"content", targetpath) - self.sim.update_snapshot() - - updater = self._init_updater() - updater.config.prefix_targets_with_hash = prefix_targets_with_hash - updater.refresh() - - for path in targetpaths: - info = updater.get_targetinfo(path) - assert isinstance(info, TargetFile) - updater.download_target(info) - - # target files are always persisted without hash prefix - self._assert_targets_files_exist([info.path]) - - # files are fetched with the expected hash prefix (or None) - exp_calls = [ - (path, None if not hash_algo else info.hashes[hash_algo]) - ] - - self.assertListEqual(self.sim.fetch_tracker.targets, exp_calls) - self.sim.fetch_tracker.targets.clear() - finally: - self.teardown_subtest() + consistent_snapshot: bool = test_case_data["consistent_snapshot"] + prefix_targets_with_hash: bool = test_case_data["prefix_targets"] + hash_algo: str | None = test_case_data["hash_algo"] + targetpaths: list[str] = test_case_data["targetpaths"] + + self.setup_subtest(consistent_snapshot, prefix_targets_with_hash) + # Add targets to repository + for targetpath in targetpaths: + self.sim.targets.version += 1 + self.sim.add_target("targets", b"content", targetpath) + self.sim.update_snapshot() + + updater = self._init_updater() + updater.config.prefix_targets_with_hash = prefix_targets_with_hash + updater.refresh() + + for path in targetpaths: + info = updater.get_targetinfo(path) + assert isinstance(info, TargetFile) + updater.download_target(info) + + # target files are always persisted without hash prefix + self._assert_targets_files_exist([info.path]) + + # files are fetched with the expected hash prefix (or None) + exp_calls = [ + (path, None if not hash_algo else info.hashes[hash_algo]) + ] + + self.assertListEqual(self.sim.fetch_tracker.targets, exp_calls) + self.sim.fetch_tracker.targets.clear() if __name__ == "__main__": diff --git a/tests/test_updater_delegation_graphs.py b/tests/test_updater_delegation_graphs.py index 536bb13a2d..1fb3b3458b 100644 --- a/tests/test_updater_delegation_graphs.py +++ b/tests/test_updater_delegation_graphs.py @@ -264,32 +264,29 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None: """Test that delegated roles are traversed in the order of appearance in the delegator's metadata, using pre-order depth-first search""" - try: - exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order] - exp_calls = [(role, 1) for role in test_data.visited_order] - - self._init_repo(test_data) - self.setup_subtest() - - updater = self._init_updater() - # restrict the max number of delegations to simplify the test - updater.config.max_delegations = 4 - # Call explicitly refresh to simplify the expected_calls list - updater.refresh() - self.sim.fetch_tracker.metadata.clear() - # Check that metadata dir contains only top-level roles - self._assert_files_exist(TOP_LEVEL_ROLE_NAMES) - - # Looking for a non-existing targetpath forces updater - # to visit all possible delegated roles - targetfile = updater.get_targetinfo("missingpath") - self.assertIsNone(targetfile) - # Check that the delegated roles were visited in the expected - # order and the corresponding metadata files were persisted - self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) - self._assert_files_exist(exp_files) - finally: - self.teardown_subtest() + exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order] + exp_calls = [(role, 1) for role in test_data.visited_order] + + self._init_repo(test_data) + self.setup_subtest() + + updater = self._init_updater() + # restrict the max number of delegations to simplify the test + updater.config.max_delegations = 4 + # Call explicitly refresh to simplify the expected_calls list + updater.refresh() + self.sim.fetch_tracker.metadata.clear() + # Check that metadata dir contains only top-level roles + self._assert_files_exist(TOP_LEVEL_ROLE_NAMES) + + # Looking for a non-existing targetpath forces updater + # to visit all possible delegated roles + targetfile = updater.get_targetinfo("missingpath") + self.assertIsNone(targetfile) + # Check that the delegated roles were visited in the expected + # order and the corresponding metadata files were persisted + self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) + self._assert_files_exist(exp_files) invalid_metadata = { "unsigned delegated role": DelegationsTestCase( @@ -305,30 +302,27 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None: @utils.run_sub_tests_with_dataset(invalid_metadata) def test_invalid_metadata(self, test_data: DelegationsTestCase) -> None: - try: - self._init_repo(test_data) - # The invalid role is the last visited - invalid_role = test_data.visited_order[-1] - self.sim.signers[invalid_role].clear() - - self.setup_subtest() - # The invalid role metadata must not be persisted - exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order[:-1]] - exp_calls = [(role, 1) for role in test_data.visited_order] - - updater = self._init_updater() - # Call explicitly refresh to simplify the expected_calls list - updater.refresh() - self.sim.fetch_tracker.metadata.clear() - - with self.assertRaises(UnsignedMetadataError): - updater.get_targetinfo("missingpath") - # Check that there were no visited roles after the invalid one - # and only the valid metadata files were persisted - self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) - self._assert_files_exist(exp_files) - finally: - self.teardown_subtest() + self._init_repo(test_data) + # The invalid role is the last visited + invalid_role = test_data.visited_order[-1] + self.sim.signers[invalid_role].clear() + + self.setup_subtest() + # The invalid role metadata must not be persisted + exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order[:-1]] + exp_calls = [(role, 1) for role in test_data.visited_order] + + updater = self._init_updater() + # Call explicitly refresh to simplify the expected_calls list + updater.refresh() + self.sim.fetch_tracker.metadata.clear() + + with self.assertRaises(UnsignedMetadataError): + updater.get_targetinfo("missingpath") + # Check that there were no visited roles after the invalid one + # and only the valid metadata files were persisted + self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) + self._assert_files_exist(exp_files) def test_safely_encoded_rolenames(self) -> None: """Test that delegated roles names are safely encoded in the filenames @@ -398,34 +392,31 @@ def test_hash_bins_graph_traversal( in the delegator's metadata, using pre-order depth-first search and that they correctly refer to the corresponding hash bin prefixes""" - try: - exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order] - exp_calls = [(role, 1) for role in test_data.visited_order] - - self._init_repo(test_data) - self.setup_subtest() - - updater = self._init_updater() - # Call explicitly refresh to simplify the expected_calls list - updater.refresh() - self.sim.fetch_tracker.metadata.clear() - # Check that metadata dir contains only top-level roles - self._assert_files_exist(TOP_LEVEL_ROLE_NAMES) - - # Looking for a non-existing targetpath forces updater - # to visit a correspondning delegated role - targetfile = updater.get_targetinfo("missingpath") - self.assertIsNone(targetfile) - targetfile = updater.get_targetinfo("othermissingpath") - self.assertIsNone(targetfile) - targetfile = updater.get_targetinfo("thirdmissingpath") - self.assertIsNone(targetfile) - # Check that the delegated roles were visited in the expected - # order and the corresponding metadata files were persisted - self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) - self._assert_files_exist(exp_files) - finally: - self.teardown_subtest() + exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order] + exp_calls = [(role, 1) for role in test_data.visited_order] + + self._init_repo(test_data) + self.setup_subtest() + + updater = self._init_updater() + # Call explicitly refresh to simplify the expected_calls list + updater.refresh() + self.sim.fetch_tracker.metadata.clear() + # Check that metadata dir contains only top-level roles + self._assert_files_exist(TOP_LEVEL_ROLE_NAMES) + + # Looking for a non-existing targetpath forces updater + # to visit a correspondning delegated role + targetfile = updater.get_targetinfo("missingpath") + self.assertIsNone(targetfile) + targetfile = updater.get_targetinfo("othermissingpath") + self.assertIsNone(targetfile) + targetfile = updater.get_targetinfo("thirdmissingpath") + self.assertIsNone(targetfile) + # Check that the delegated roles were visited in the expected + # order and the corresponding metadata files were persisted + self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) + self._assert_files_exist(exp_files) @dataclass class SuccinctRolesTestCase: @@ -482,35 +473,31 @@ def test_succinct_roles_graph_traversal( # successful traversal all top level metadata files plus the expected # bin should exist locally and only one bin must be downloaded. - try: - exp_files = [*TOP_LEVEL_ROLE_NAMES, test_data.expected_target_bin] - exp_calls = [(test_data.expected_target_bin, 1)] - - self.sim = RepositorySimulator() - self.sim.add_succinct_roles("targets", test_data.bit_length, "bin") - self.sim.update_snapshot() + exp_files = [*TOP_LEVEL_ROLE_NAMES, test_data.expected_target_bin] + exp_calls = [(test_data.expected_target_bin, 1)] - self.setup_subtest() + self.sim = RepositorySimulator() + self.sim.add_succinct_roles("targets", test_data.bit_length, "bin") + self.sim.update_snapshot() - updater = self._init_updater() - # Call explicitly refresh to simplify the expected_calls list. - updater.refresh() - self.sim.fetch_tracker.metadata.clear() - # Check that metadata dir contains only top-level roles - self._assert_files_exist(TOP_LEVEL_ROLE_NAMES) + self.setup_subtest() - # Looking for a non-existing targetpath forces updater - # to visit a corresponding delegated role. - targetfile = updater.get_targetinfo(test_data.target_path) - self.assertIsNone(targetfile) + updater = self._init_updater() + # Call explicitly refresh to simplify the expected_calls list. + updater.refresh() + self.sim.fetch_tracker.metadata.clear() + # Check that metadata dir contains only top-level roles + self._assert_files_exist(TOP_LEVEL_ROLE_NAMES) - # Check that the delegated roles were visited in the expected - # order and the corresponding metadata files were persisted. - self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) - self._assert_files_exist(exp_files) + # Looking for a non-existing targetpath forces updater + # to visit a corresponding delegated role. + targetfile = updater.get_targetinfo(test_data.target_path) + self.assertIsNone(targetfile) - finally: - self.teardown_subtest() + # Check that the delegated roles were visited in the expected + # order and the corresponding metadata files were persisted. + self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) + self._assert_files_exist(exp_files) class TestTargetFileSearch(TestDelegations): @@ -564,29 +551,26 @@ def setUp(self) -> None: @utils.run_sub_tests_with_dataset(targets) def test_targetfile_search(self, test_data: TargetTestCase) -> None: - try: - self.setup_subtest() - exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order] - exp_calls = [(role, 1) for role in test_data.visited_order] - exp_target = self.sim.target_files[test_data.targetpath].target_file - - updater = self._init_updater() - # Call explicitly refresh to simplify the expected_calls list - updater.refresh() - self.sim.fetch_tracker.metadata.clear() - target = updater.get_targetinfo(test_data.targetpath) - if target is not None: - # Confirm that the expected TargetFile is found - self.assertTrue(test_data.found) - self.assertDictEqual(target.to_dict(), exp_target.to_dict()) - else: - self.assertFalse(test_data.found) - # Check that the delegated roles were visited in the expected - # order and the corresponding metadata files were persisted - self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) - self._assert_files_exist(exp_files) - finally: - self.teardown_subtest() + self.setup_subtest() + exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order] + exp_calls = [(role, 1) for role in test_data.visited_order] + exp_target = self.sim.target_files[test_data.targetpath].target_file + + updater = self._init_updater() + # Call explicitly refresh to simplify the expected_calls list + updater.refresh() + self.sim.fetch_tracker.metadata.clear() + target = updater.get_targetinfo(test_data.targetpath) + if target is not None: + # Confirm that the expected TargetFile is found + self.assertTrue(test_data.found) + self.assertDictEqual(target.to_dict(), exp_target.to_dict()) + else: + self.assertFalse(test_data.found) + # Check that the delegated roles were visited in the expected + # order and the corresponding metadata files were persisted + self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls) + self._assert_files_exist(exp_files) if __name__ == "__main__": diff --git a/tests/test_utils.py b/tests/test_utils.py index fcdc3c449b..a8c52ec8e4 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -43,6 +43,36 @@ def can_connect(port: int) -> bool: sock.close() +class TestRunSubTestsWithDataset(unittest.TestCase): + def test_teardown_subtest_runs_after_failure(self) -> None: + events = [] + + class SubTestCase(unittest.TestCase): + def teardown_subtest(self) -> None: + events.append(f"cleanup:{self.case_name}") + + @utils.run_sub_tests_with_dataset( + {"failing": False, "passing": True} + ) + def test_subtests(self, succeeds: bool) -> None: + events.append(f"run:{self.case_name}") + self.assertTrue(succeeds) + + result = unittest.TestResult() + SubTestCase("test_subtests").run(result) + + self.assertEqual( + [ + "run:failing", + "cleanup:failing", + "run:passing", + "cleanup:passing", + ], + events, + ) + self.assertEqual(len(result.failures), 1) + + class TestServerProcess(unittest.TestCase): """Test functionality provided in TestServerProcess from tests/utils.py.""" diff --git a/tests/utils.py b/tests/utils.py index cc35af0447..86a169fd25 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -64,7 +64,11 @@ def wrapper(test_cls: unittest.TestCase) -> None: with test_cls.subTest(case=case): # Save case name for future reference test_cls.case_name = case.replace(" ", "_") - function(test_cls, data) + try: + function(test_cls, data) + finally: + if hasattr(test_cls, "teardown_subtest"): + test_cls.teardown_subtest() return wrapper