|
12 | 12 | import tempfile
|
13 | 13 | import unittest
|
14 | 14 | from typing import Optional, Tuple
|
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
|
@@ -254,6 +254,49 @@ def test_not_loading_targets_twice(self, wrapped_open: MagicMock) -> None:
|
254 | 254 | updater.get_targetinfo("somepath")
|
255 | 255 | wrapped_open.assert_not_called()
|
256 | 256 |
|
| 257 | + @patch.object(builtins, "open", wraps=builtins.open) |
| 258 | + def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None: |
| 259 | + |
| 260 | + # Add new delegated targets |
| 261 | + spec_version = ".".join(SPECIFICATION_VERSION) |
| 262 | + targets = Targets(1, spec_version, self.sim.safe_expiry, {}, None) |
| 263 | + self.sim.add_delegation("targets", "role1", targets, False, ["*"], None) |
| 264 | + self.sim.update_snapshot() |
| 265 | + |
| 266 | + # Make a successful update of valid metadata which stores it in cache |
| 267 | + updater = self._run_refresh() |
| 268 | + updater.get_targetinfo("non_existent_target") |
| 269 | + |
| 270 | + # Clean up calls to open during refresh() |
| 271 | + wrapped_open.reset_mock() |
| 272 | + # Clean up fetch tracker metadata |
| 273 | + self.sim.fetch_tracker.metadata.clear() |
| 274 | + |
| 275 | + # Create a new updater and perform a second update while |
| 276 | + # the metadata is already stored in cache (metadata dir) |
| 277 | + updater = Updater( |
| 278 | + self.metadata_dir, |
| 279 | + "https://example.com/metadata/", |
| 280 | + self.targets_dir, |
| 281 | + "https://example.com/targets/", |
| 282 | + self.sim, |
| 283 | + ) |
| 284 | + updater.get_targetinfo("non_existent_target") |
| 285 | + |
| 286 | + # Test that metadata is loaded from cache and not downloaded |
| 287 | + wrapped_open.assert_has_calls( |
| 288 | + [ |
| 289 | + call(os.path.join(self.metadata_dir, "root.json"), "rb"), |
| 290 | + call(os.path.join(self.metadata_dir, "timestamp.json"), "rb"), |
| 291 | + call(os.path.join(self.metadata_dir, "snapshot.json"), "rb"), |
| 292 | + call(os.path.join(self.metadata_dir, "targets.json"), "rb"), |
| 293 | + call(os.path.join(self.metadata_dir, "role1.json"), "rb"), |
| 294 | + ] |
| 295 | + ) |
| 296 | + |
| 297 | + expected_calls = [("root", 2), ("timestamp", None)] |
| 298 | + self.assertListEqual(self.sim.fetch_tracker.metadata, expected_calls) |
| 299 | + |
257 | 300 |
|
258 | 301 | if __name__ == "__main__":
|
259 | 302 | if "--dump" in sys.argv:
|
|
0 commit comments