77
88import builtins
99import datetime
10- import logging
1110import os
1211import sys
1312import tempfile
@@ -72,13 +71,13 @@ def setUp(self) -> None:
7271 def tearDown (self ) -> None :
7372 self .temp_dir .cleanup ()
7473
75- def _run_refresh (self , skip_bootstrap :bool = False ) -> Updater :
74+ def _run_refresh (self , skip_bootstrap : bool = False ) -> Updater :
7675 """Create a new Updater instance and refresh"""
7776 updater = self ._init_updater (skip_bootstrap )
7877 updater .refresh ()
7978 return updater
8079
81- def _init_updater (self , skip_bootstrap :bool = False ) -> Updater :
80+ def _init_updater (self , skip_bootstrap : bool = False ) -> Updater :
8281 """Create a new Updater instance"""
8382 if self .dump_dir is not None :
8483 self .sim .write ()
@@ -89,7 +88,7 @@ def _init_updater(self, skip_bootstrap:bool=False) -> Updater:
8988 self .targets_dir ,
9089 "https://example.com/targets/" ,
9190 self .sim ,
92- bootstrap = None if skip_bootstrap else self .sim .signed_roots [0 ]
91+ bootstrap = None if skip_bootstrap else self .sim .signed_roots [0 ],
9392 )
9493
9594 def _assert_files_exist (self , roles : Iterable [str ]) -> None :
@@ -706,9 +705,10 @@ def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None:
706705 updater .get_targetinfo ("non_existent_target" )
707706
708707 # Test that metadata is loaded from cache and not downloaded
708+ root_dir = os .path .join (self .metadata_dir , "root_history" )
709709 wrapped_open .assert_has_calls (
710710 [
711- call (os .path .join (self . metadata_dir , "root_history/ 2.root.json" ), "rb" ),
711+ call (os .path .join (root_dir , "2.root.json" ), "rb" ),
712712 call (os .path .join (self .metadata_dir , "timestamp.json" ), "rb" ),
713713 call (os .path .join (self .metadata_dir , "snapshot.json" ), "rb" ),
714714 call (os .path .join (self .metadata_dir , "targets.json" ), "rb" ),
@@ -719,7 +719,6 @@ def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None:
719719 expected_calls = [("root" , 2 ), ("timestamp" , None )]
720720 self .assertListEqual (self .sim .fetch_tracker .metadata , expected_calls )
721721
722-
723722 @patch .object (builtins , "open" , wraps = builtins .open )
724723 def test_intermediate_root_cache (self , wrapped_open : MagicMock ) -> None :
725724 """Test that refresh uses the intermediate roots from cache"""
@@ -733,17 +732,25 @@ def test_intermediate_root_cache(self, wrapped_open: MagicMock) -> None:
733732 self ._run_refresh ()
734733
735734 # assert that cache lookups happened but data was downloaded from remote
735+ root_dir = os .path .join (self .metadata_dir , "root_history" )
736736 wrapped_open .assert_has_calls (
737737 [
738- call (os .path .join (self . metadata_dir , "root_history/ 2.root.json" ), "rb" ),
739- call (os .path .join (self . metadata_dir , "root_history/ 3.root.json" ), "rb" ),
740- call (os .path .join (self . metadata_dir , "root_history/ 4.root.json" ), "rb" ),
738+ call (os .path .join (root_dir , "2.root.json" ), "rb" ),
739+ call (os .path .join (root_dir , "3.root.json" ), "rb" ),
740+ call (os .path .join (root_dir , "4.root.json" ), "rb" ),
741741 call (os .path .join (self .metadata_dir , "timestamp.json" ), "rb" ),
742742 call (os .path .join (self .metadata_dir , "snapshot.json" ), "rb" ),
743743 call (os .path .join (self .metadata_dir , "targets.json" ), "rb" ),
744744 ]
745745 )
746- expected_calls = [("root" , 2 ), ("root" , 3 ), ("root" , 4 ), ("timestamp" , None ), ("snapshot" , 1 ), ("targets" , 1 )]
746+ expected_calls = [
747+ ("root" , 2 ),
748+ ("root" , 3 ),
749+ ("root" , 4 ),
750+ ("timestamp" , None ),
751+ ("snapshot" , 1 ),
752+ ("targets" , 1 ),
753+ ]
747754 self .assertListEqual (self .sim .fetch_tracker .metadata , expected_calls )
748755
749756 # Clear statistics for open() calls and metadata requests
@@ -754,9 +761,9 @@ def test_intermediate_root_cache(self, wrapped_open: MagicMock) -> None:
754761 self ._run_refresh ()
755762 wrapped_open .assert_has_calls (
756763 [
757- call (os .path .join (self . metadata_dir , "root_history/ 2.root.json" ), "rb" ),
758- call (os .path .join (self . metadata_dir , "root_history/ 3.root.json" ), "rb" ),
759- call (os .path .join (self . metadata_dir , "root_history/ 4.root.json" ), "rb" ),
764+ call (os .path .join (root_dir , "2.root.json" ), "rb" ),
765+ call (os .path .join (root_dir , "3.root.json" ), "rb" ),
766+ call (os .path .join (root_dir , "4.root.json" ), "rb" ),
760767 call (os .path .join (self .metadata_dir , "timestamp.json" ), "rb" ),
761768 call (os .path .join (self .metadata_dir , "snapshot.json" ), "rb" ),
762769 call (os .path .join (self .metadata_dir , "targets.json" ), "rb" ),
@@ -777,7 +784,9 @@ def test_intermediate_root_cache_poisoning(self) -> None:
777784 self ._run_refresh ()
778785
779786 # Modify cached intermediate root v2 so that it's no longer signed correctly
780- root_path = os .path .join (self .metadata_dir , "root_history" , "2.root.json" )
787+ root_path = os .path .join (
788+ self .metadata_dir , "root_history" , "2.root.json"
789+ )
781790 md = Metadata .from_file (root_path )
782791 md .signatures .clear ()
783792 md .to_file (root_path )
0 commit comments