5
5
6
6
import builtins
7
7
import datetime
8
+ import logging
8
9
import os
9
10
import sys
10
11
import tempfile
@@ -58,10 +59,6 @@ def setUp(self) -> None:
58
59
59
60
self .sim = RepositorySimulator ()
60
61
61
- # boostrap client with initial root metadata
62
- with open (os .path .join (self .metadata_dir , "root.json" ), "bw" ) as f :
63
- f .write (self .sim .signed_roots [0 ])
64
-
65
62
if self .dump_dir is not None :
66
63
# create test specific dump directory
67
64
name = self .id ().split ("." )[- 1 ]
@@ -71,22 +68,13 @@ def setUp(self) -> None:
71
68
def tearDown (self ) -> None :
72
69
self .temp_dir .cleanup ()
73
70
74
- def _run_refresh (self ) -> Updater :
71
+ def _run_refresh (self , skip_bootstrap : bool = False ) -> Updater :
75
72
"""Create a new Updater instance and refresh"""
76
- if self .dump_dir is not None :
77
- self .sim .write ()
78
-
79
- updater = Updater (
80
- self .metadata_dir ,
81
- "https://example.com/metadata/" ,
82
- self .targets_dir ,
83
- "https://example.com/targets/" ,
84
- self .sim ,
85
- )
73
+ updater = self ._init_updater (skip_bootstrap )
86
74
updater .refresh ()
87
75
return updater
88
76
89
- def _init_updater (self ) -> Updater :
77
+ def _init_updater (self , skip_bootstrap : bool = False ) -> Updater :
90
78
"""Create a new Updater instance"""
91
79
if self .dump_dir is not None :
92
80
self .sim .write ()
@@ -97,6 +85,7 @@ def _init_updater(self) -> Updater:
97
85
self .targets_dir ,
98
86
"https://example.com/targets/" ,
99
87
self .sim ,
88
+ bootstrap = None if skip_bootstrap else self .sim .signed_roots [0 ]
100
89
)
101
90
102
91
def _assert_files_exist (self , roles : Iterable [str ]) -> None :
@@ -122,9 +111,6 @@ def _assert_version_equals(self, role: str, expected_version: int) -> None:
122
111
self .assertEqual (md .signed .version , expected_version )
123
112
124
113
def test_first_time_refresh (self ) -> None :
125
- # Metadata dir contains only the mandatory initial root.json
126
- self ._assert_files_exist ([Root .type ])
127
-
128
114
# Add one more root version to repository so that
129
115
# refresh() updates from local trusted root (v1) to
130
116
# remote root (v2)
@@ -138,10 +124,11 @@ def test_first_time_refresh(self) -> None:
138
124
version = 2 if role == Root .type else None
139
125
self ._assert_content_equals (role , version )
140
126
141
- def test_trusted_root_missing (self ) -> None :
142
- os .remove (os .path .join (self .metadata_dir , "root.json" ))
127
+ def test_cached_root_missing_without_bootstrap (self ) -> None :
128
+ # Run update without a bootstrap, with empty cache: this fails since there is no
129
+ # trusted root
143
130
with self .assertRaises (OSError ):
144
- self ._run_refresh ()
131
+ self ._run_refresh (skip_bootstrap = True )
145
132
146
133
# Metadata dir is empty
147
134
self .assertFalse (os .listdir (self .metadata_dir ))
@@ -174,15 +161,15 @@ def test_trusted_root_expired(self) -> None:
174
161
self ._assert_files_exist (TOP_LEVEL_ROLE_NAMES )
175
162
self ._assert_content_equals (Root .type , 3 )
176
163
177
- def test_trusted_root_unsigned (self ) -> None :
178
- # Local trusted root is not signed
164
+ def test_trusted_root_unsigned_without_bootstrap (self ) -> None :
165
+ # Cached root is not signed, bootstrap root is not used
179
166
root_path = os .path .join (self .metadata_dir , "root.json" )
180
- md_root = Metadata .from_file ( root_path )
167
+ md_root = Metadata .from_bytes ( self . sim . signed_roots [ 0 ] )
181
168
md_root .signatures .clear ()
182
169
md_root .to_file (root_path )
183
170
184
171
with self .assertRaises (UnsignedMetadataError ):
185
- self ._run_refresh ()
172
+ self ._run_refresh (skip_bootstrap = True )
186
173
187
174
# The update failed, no changes in metadata
188
175
self ._assert_files_exist ([Root .type ])
@@ -200,10 +187,7 @@ def test_max_root_rotations(self) -> None:
200
187
self .sim .root .version += 1
201
188
self .sim .publish_root ()
202
189
203
- md_root = Metadata .from_file (
204
- os .path .join (self .metadata_dir , "root.json" )
205
- )
206
- initial_root_version = md_root .signed .version
190
+ initial_root_version = 1
207
191
208
192
updater .refresh ()
209
193
@@ -708,26 +692,18 @@ def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None:
708
692
updater = self ._run_refresh ()
709
693
updater .get_targetinfo ("non_existent_target" )
710
694
711
- # Clean up calls to open during refresh()
695
+ # Clear statistics for calls and metadata requests
712
696
wrapped_open .reset_mock ()
713
- # Clean up fetch tracker metadata
714
697
self .sim .fetch_tracker .metadata .clear ()
715
698
716
699
# Create a new updater and perform a second update while
717
700
# the metadata is already stored in cache (metadata dir)
718
- updater = Updater (
719
- self .metadata_dir ,
720
- "https://example.com/metadata/" ,
721
- self .targets_dir ,
722
- "https://example.com/targets/" ,
723
- self .sim ,
724
- )
701
+ updater = self ._init_updater ()
725
702
updater .get_targetinfo ("non_existent_target" )
726
703
727
704
# Test that metadata is loaded from cache and not downloaded
728
705
wrapped_open .assert_has_calls (
729
706
[
730
- call (os .path .join (self .metadata_dir , "root.json" ), "rb" ),
731
707
call (os .path .join (self .metadata_dir , "root_history/2.root.json" ), "rb" ),
732
708
call (os .path .join (self .metadata_dir , "timestamp.json" ), "rb" ),
733
709
call (os .path .join (self .metadata_dir , "snapshot.json" ), "rb" ),
0 commit comments