5
5
6
6
import builtins
7
7
import datetime
8
- import logging
9
8
import os
10
9
import sys
11
10
import tempfile
@@ -68,13 +67,13 @@ def setUp(self) -> None:
68
67
def tearDown (self ) -> None :
69
68
self .temp_dir .cleanup ()
70
69
71
- def _run_refresh (self , skip_bootstrap :bool = False ) -> Updater :
70
+ def _run_refresh (self , skip_bootstrap : bool = False ) -> Updater :
72
71
"""Create a new Updater instance and refresh"""
73
72
updater = self ._init_updater (skip_bootstrap )
74
73
updater .refresh ()
75
74
return updater
76
75
77
- def _init_updater (self , skip_bootstrap :bool = False ) -> Updater :
76
+ def _init_updater (self , skip_bootstrap : bool = False ) -> Updater :
78
77
"""Create a new Updater instance"""
79
78
if self .dump_dir is not None :
80
79
self .sim .write ()
@@ -85,7 +84,7 @@ def _init_updater(self, skip_bootstrap:bool=False) -> Updater:
85
84
self .targets_dir ,
86
85
"https://example.com/targets/" ,
87
86
self .sim ,
88
- bootstrap = None if skip_bootstrap else self .sim .signed_roots [0 ]
87
+ bootstrap = None if skip_bootstrap else self .sim .signed_roots [0 ],
89
88
)
90
89
91
90
def _assert_files_exist (self , roles : Iterable [str ]) -> None :
@@ -702,9 +701,10 @@ def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None:
702
701
updater .get_targetinfo ("non_existent_target" )
703
702
704
703
# Test that metadata is loaded from cache and not downloaded
704
+ root_dir = os .path .join (self .metadata_dir , "root_history" )
705
705
wrapped_open .assert_has_calls (
706
706
[
707
- call (os .path .join (self . metadata_dir , "root_history/ 2.root.json" ), "rb" ),
707
+ call (os .path .join (root_dir , "2.root.json" ), "rb" ),
708
708
call (os .path .join (self .metadata_dir , "timestamp.json" ), "rb" ),
709
709
call (os .path .join (self .metadata_dir , "snapshot.json" ), "rb" ),
710
710
call (os .path .join (self .metadata_dir , "targets.json" ), "rb" ),
@@ -715,7 +715,6 @@ def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None:
715
715
expected_calls = [("root" , 2 ), ("timestamp" , None )]
716
716
self .assertListEqual (self .sim .fetch_tracker .metadata , expected_calls )
717
717
718
-
719
718
@patch .object (builtins , "open" , wraps = builtins .open )
720
719
def test_intermediate_root_cache (self , wrapped_open : MagicMock ) -> None :
721
720
"""Test that refresh uses the intermediate roots from cache"""
@@ -729,17 +728,25 @@ def test_intermediate_root_cache(self, wrapped_open: MagicMock) -> None:
729
728
self ._run_refresh ()
730
729
731
730
# assert that cache lookups happened but data was downloaded from remote
731
+ root_dir = os .path .join (self .metadata_dir , "root_history" )
732
732
wrapped_open .assert_has_calls (
733
733
[
734
- call (os .path .join (self . metadata_dir , "root_history/ 2.root.json" ), "rb" ),
735
- call (os .path .join (self . metadata_dir , "root_history/ 3.root.json" ), "rb" ),
736
- call (os .path .join (self . metadata_dir , "root_history/ 4.root.json" ), "rb" ),
734
+ call (os .path .join (root_dir , "2.root.json" ), "rb" ),
735
+ call (os .path .join (root_dir , "3.root.json" ), "rb" ),
736
+ call (os .path .join (root_dir , "4.root.json" ), "rb" ),
737
737
call (os .path .join (self .metadata_dir , "timestamp.json" ), "rb" ),
738
738
call (os .path .join (self .metadata_dir , "snapshot.json" ), "rb" ),
739
739
call (os .path .join (self .metadata_dir , "targets.json" ), "rb" ),
740
740
]
741
741
)
742
- expected_calls = [("root" , 2 ), ("root" , 3 ), ("root" , 4 ), ("timestamp" , None ), ("snapshot" , 1 ), ("targets" , 1 )]
742
+ expected_calls = [
743
+ ("root" , 2 ),
744
+ ("root" , 3 ),
745
+ ("root" , 4 ),
746
+ ("timestamp" , None ),
747
+ ("snapshot" , 1 ),
748
+ ("targets" , 1 ),
749
+ ]
743
750
self .assertListEqual (self .sim .fetch_tracker .metadata , expected_calls )
744
751
745
752
# Clear statistics for open() calls and metadata requests
@@ -750,9 +757,9 @@ def test_intermediate_root_cache(self, wrapped_open: MagicMock) -> None:
750
757
self ._run_refresh ()
751
758
wrapped_open .assert_has_calls (
752
759
[
753
- call (os .path .join (self . metadata_dir , "root_history/ 2.root.json" ), "rb" ),
754
- call (os .path .join (self . metadata_dir , "root_history/ 3.root.json" ), "rb" ),
755
- call (os .path .join (self . metadata_dir , "root_history/ 4.root.json" ), "rb" ),
760
+ call (os .path .join (root_dir , "2.root.json" ), "rb" ),
761
+ call (os .path .join (root_dir , "3.root.json" ), "rb" ),
762
+ call (os .path .join (root_dir , "4.root.json" ), "rb" ),
756
763
call (os .path .join (self .metadata_dir , "timestamp.json" ), "rb" ),
757
764
call (os .path .join (self .metadata_dir , "snapshot.json" ), "rb" ),
758
765
call (os .path .join (self .metadata_dir , "targets.json" ), "rb" ),
@@ -773,7 +780,9 @@ def test_intermediate_root_cache_poisoning(self) -> None:
773
780
self ._run_refresh ()
774
781
775
782
# Modify cached intermediate root v2 so that it's no longer signed correctly
776
- root_path = os .path .join (self .metadata_dir , "root_history" , "2.root.json" )
783
+ root_path = os .path .join (
784
+ self .metadata_dir , "root_history" , "2.root.json"
785
+ )
777
786
md = Metadata .from_file (root_path )
778
787
md .signatures .clear ()
779
788
md .to_file (root_path )
0 commit comments