|
12 | 12 | import unittest
|
13 | 13 | from datetime import datetime, timedelta
|
14 | 14 | from typing import Iterable, Optional
|
15 |
| -from unittest.mock import MagicMock, patch |
| 15 | +from unittest.mock import MagicMock, call, patch |
16 | 16 |
|
17 | 17 | from tests import utils
|
18 | 18 | from tests.repository_simulator import RepositorySimulator
|
@@ -636,6 +636,50 @@ def test_snapshot_rollback_with_local_snapshot_hash_mismatch(self) -> None:
|
636 | 636 | with self.assertRaises(BadVersionNumberError):
|
637 | 637 | self._run_refresh()
|
638 | 638 |
|
| 639 | + @patch.object(builtins, "open", wraps=builtins.open) |
| 640 | + def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None: |
| 641 | + |
| 642 | + # Add new delegated targets |
| 643 | + spec_version = ".".join(SPECIFICATION_VERSION) |
| 644 | + targets = Targets(1, spec_version, self.sim.safe_expiry, {}, None) |
| 645 | + role = DelegatedRole("role1", [], 1, False, ["*"], None) |
| 646 | + self.sim.add_delegation("targets", role, targets) |
| 647 | + self.sim.update_snapshot() |
| 648 | + |
| 649 | + # Make a successful update of valid metadata which stores it in cache |
| 650 | + updater = self._run_refresh() |
| 651 | + updater.get_targetinfo("non_existent_target") |
| 652 | + |
| 653 | + # Clean up calls to open during refresh() |
| 654 | + wrapped_open.reset_mock() |
| 655 | + # Clean up fetch tracker metadata |
| 656 | + self.sim.fetch_tracker.metadata.clear() |
| 657 | + |
| 658 | + # Create a new updater and perform a second update while |
| 659 | + # the metadata is already stored in cache (metadata dir) |
| 660 | + updater = Updater( |
| 661 | + self.metadata_dir, |
| 662 | + "https://example.com/metadata/", |
| 663 | + self.targets_dir, |
| 664 | + "https://example.com/targets/", |
| 665 | + self.sim, |
| 666 | + ) |
| 667 | + updater.get_targetinfo("non_existent_target") |
| 668 | + |
| 669 | + # Test that metadata is loaded from cache and not downloaded |
| 670 | + wrapped_open.assert_has_calls( |
| 671 | + [ |
| 672 | + call(os.path.join(self.metadata_dir, "root.json"), "rb"), |
| 673 | + call(os.path.join(self.metadata_dir, "timestamp.json"), "rb"), |
| 674 | + call(os.path.join(self.metadata_dir, "snapshot.json"), "rb"), |
| 675 | + call(os.path.join(self.metadata_dir, "targets.json"), "rb"), |
| 676 | + call(os.path.join(self.metadata_dir, "role1.json"), "rb"), |
| 677 | + ] |
| 678 | + ) |
| 679 | + |
| 680 | + expected_calls = [("root", 2), ("timestamp", None)] |
| 681 | + self.assertListEqual(self.sim.fetch_tracker.metadata, expected_calls) |
| 682 | + |
639 | 683 |
|
640 | 684 | if __name__ == "__main__":
|
641 | 685 | if "--dump" in sys.argv:
|
|
0 commit comments