Skip to content

Commit e752193

Browse files
sechkovasechkova
authored
Merge pull request #1728 from sechkova/test-fetch-target
ngtests: Test fetch target
2 parents ed15d11 + d1bc201 commit e752193

File tree

2 files changed

+191
-49
lines changed

2 files changed

+191
-49
lines changed

‎tests/test_updater_fetch_target.py

+190
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,190 @@
1+
#!/usr/bin/env python
2+
3+
# Copyright 2021, New York University and the TUF contributors
4+
# SPDX-License-Identifier: MIT OR Apache-2.0
5+
6+
"""Test 'Fetch target' from 'Detailed client workflow' as well as
7+
target files storing/loading from cache.
8+
"""
9+
import os
10+
import sys
11+
import tempfile
12+
import unittest
13+
from dataclasses import dataclass
14+
from typing import Optional
15+
16+
from tests import utils
17+
from tests.repository_simulator import RepositorySimulator
18+
from tuf.exceptions import RepositoryError
19+
from tuf.ngclient import Updater
20+
21+
22+
@dataclass
23+
class TestTarget:
24+
path: str
25+
content: bytes
26+
encoded_path: str
27+
28+
29+
class TestFetchTarget(unittest.TestCase):
30+
"""Test ngclient downloading and caching target files."""
31+
32+
# set dump_dir to trigger repository state dumps
33+
dump_dir: Optional[str] = None
34+
35+
def setUp(self) -> None:
36+
# pylint: disable-next=consider-using-with
37+
self.temp_dir = tempfile.TemporaryDirectory()
38+
self.metadata_dir = os.path.join(self.temp_dir.name, "metadata")
39+
self.targets_dir = os.path.join(self.temp_dir.name, "targets")
40+
os.mkdir(self.metadata_dir)
41+
os.mkdir(self.targets_dir)
42+
43+
# Setup the repository, bootstrap client root.json
44+
self.sim = RepositorySimulator()
45+
with open(os.path.join(self.metadata_dir, "root.json"), "bw") as f:
46+
f.write(self.sim.signed_roots[0])
47+
48+
if self.dump_dir is not None:
49+
# create test specific dump directory
50+
name = self.id().split(".")[-1]
51+
self.sim.dump_dir = os.path.join(self.dump_dir, name)
52+
os.mkdir(self.sim.dump_dir)
53+
54+
def tearDown(self) -> None:
55+
self.temp_dir.cleanup()
56+
57+
def _init_updater(self) -> Updater:
58+
"""Creates a new updater instance."""
59+
if self.sim.dump_dir is not None:
60+
self.sim.write()
61+
62+
updater = Updater(
63+
self.metadata_dir,
64+
"https://example.com/metadata/",
65+
self.targets_dir,
66+
"https://example.com/targets/",
67+
self.sim,
68+
)
69+
return updater
70+
71+
targets: utils.DataSet = {
72+
"standard case": TestTarget(
73+
path="targetpath",
74+
content=b"target content",
75+
encoded_path="targetpath",
76+
),
77+
"non-asci case": TestTarget(
78+
path="åäö",
79+
content=b"more content",
80+
encoded_path="%C3%A5%C3%A4%C3%B6",
81+
),
82+
"subdirectory case": TestTarget(
83+
path="a/b/c/targetpath",
84+
content=b"dir target content",
85+
encoded_path="a%2Fb%2Fc%2Ftargetpath",
86+
),
87+
}
88+
89+
@utils.run_sub_tests_with_dataset(targets)
90+
def test_fetch_target(self, target: TestTarget) -> None:
91+
path = os.path.join(self.targets_dir, target.encoded_path)
92+
93+
updater = self._init_updater()
94+
# target does not exist yet
95+
self.assertIsNone(updater.get_targetinfo(target.path))
96+
97+
# Add targets to repository
98+
self.sim.targets.version += 1
99+
self.sim.add_target("targets", target.content, target.path)
100+
self.sim.update_snapshot()
101+
102+
updater = self._init_updater()
103+
# target now exists, is not in cache yet
104+
info = updater.get_targetinfo(target.path)
105+
assert info is not None
106+
# Test without and with explicit local filepath
107+
self.assertIsNone(updater.find_cached_target(info))
108+
self.assertIsNone(updater.find_cached_target(info, path))
109+
110+
# download target, assert it is in cache and content is correct
111+
self.assertEqual(path, updater.download_target(info))
112+
self.assertEqual(path, updater.find_cached_target(info))
113+
self.assertEqual(path, updater.find_cached_target(info, path))
114+
115+
with open(path, "rb") as f:
116+
self.assertEqual(f.read(), target.content)
117+
118+
# download using explicit filepath as well
119+
os.remove(path)
120+
self.assertEqual(path, updater.download_target(info, path))
121+
self.assertEqual(path, updater.find_cached_target(info))
122+
self.assertEqual(path, updater.find_cached_target(info, path))
123+
124+
def test_invalid_target_download(self) -> None:
125+
target = TestTarget("targetpath", b"content", "targetpath")
126+
127+
# Add target to repository
128+
self.sim.targets.version += 1
129+
self.sim.add_target("targets", target.content, target.path)
130+
self.sim.update_snapshot()
131+
132+
updater = self._init_updater()
133+
info = updater.get_targetinfo(target.path)
134+
assert info is not None
135+
136+
# Corrupt the file content to not match the hash
137+
self.sim.target_files[target.path].data = b"conten@"
138+
with self.assertRaises(RepositoryError):
139+
updater.download_target(info)
140+
141+
# Corrupt the file content to not match the length
142+
self.sim.target_files[target.path].data = b"cont"
143+
with self.assertRaises(RepositoryError):
144+
updater.download_target(info)
145+
146+
# Verify the file is not persisted in cache
147+
self.assertIsNone(updater.find_cached_target(info))
148+
149+
def test_invalid_target_cache(self) -> None:
150+
target = TestTarget("targetpath", b"content", "targetpath")
151+
152+
# Add target to repository
153+
self.sim.targets.version += 1
154+
self.sim.add_target("targets", target.content, target.path)
155+
self.sim.update_snapshot()
156+
157+
# Download the target
158+
updater = self._init_updater()
159+
info = updater.get_targetinfo(target.path)
160+
assert info is not None
161+
path = updater.download_target(info)
162+
self.assertEqual(path, updater.find_cached_target(info))
163+
164+
# Add newer content to the same targetpath
165+
target.content = b"contentv2"
166+
self.sim.targets.version += 1
167+
self.sim.add_target("targets", target.content, target.path)
168+
self.sim.update_snapshot()
169+
170+
# Newer content is detected, old cached version is not used
171+
updater = self._init_updater()
172+
info = updater.get_targetinfo(target.path)
173+
assert info is not None
174+
self.assertIsNone(updater.find_cached_target(info))
175+
176+
# Download target, assert it is in cache and content is the newer
177+
path = updater.download_target(info)
178+
self.assertEqual(path, updater.find_cached_target(info))
179+
with open(path, "rb") as f:
180+
self.assertEqual(f.read(), target.content)
181+
182+
183+
if __name__ == "__main__":
184+
if "--dump" in sys.argv:
185+
TestFetchTarget.dump_dir = tempfile.mkdtemp()
186+
print(f"Repository Simulator dumps in {TestFetchTarget.dump_dir}")
187+
sys.argv.remove("--dump")
188+
189+
utils.configure_test_logging(sys.argv)
190+
unittest.main()

‎tests/test_updater_with_simulator.py

+1-49
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@
1111
import sys
1212
import tempfile
1313
import unittest
14-
from typing import Optional, Tuple
14+
from typing import Optional
1515
from unittest.mock import MagicMock, patch
1616

1717
from tests import utils
@@ -85,54 +85,6 @@ def test_refresh(self) -> None:
8585

8686
self._run_refresh()
8787

88-
targets: utils.DataSet = {
89-
"standard case": ("targetpath", b"content", "targetpath"),
90-
"non-asci case": ("åäö", b"more content", "%C3%A5%C3%A4%C3%B6"),
91-
"subdirectory case": (
92-
"a/b/c/targetpath",
93-
b"dir target content",
94-
"a%2Fb%2Fc%2Ftargetpath",
95-
),
96-
}
97-
98-
@utils.run_sub_tests_with_dataset(targets)
99-
def test_targets(self, test_case_data: Tuple[str, bytes, str]) -> None:
100-
targetpath, content, encoded_path = test_case_data
101-
path = os.path.join(self.targets_dir, encoded_path)
102-
103-
updater = self._run_refresh()
104-
# target does not exist yet, configuration is what we expect
105-
self.assertIsNone(updater.get_targetinfo(targetpath))
106-
self.assertTrue(self.sim.root.consistent_snapshot)
107-
self.assertTrue(updater.config.prefix_targets_with_hash)
108-
109-
# Add targets to repository
110-
self.sim.targets.version += 1
111-
self.sim.add_target("targets", content, targetpath)
112-
self.sim.update_snapshot()
113-
114-
updater = self._run_refresh()
115-
# target now exists, is not in cache yet
116-
info = updater.get_targetinfo(targetpath)
117-
assert info is not None
118-
# Test without and with explicit local filepath
119-
self.assertIsNone(updater.find_cached_target(info))
120-
self.assertIsNone(updater.find_cached_target(info, path))
121-
122-
# download target, assert it is in cache and content is correct
123-
self.assertEqual(path, updater.download_target(info))
124-
self.assertEqual(path, updater.find_cached_target(info))
125-
self.assertEqual(path, updater.find_cached_target(info, path))
126-
127-
with open(path, "rb") as f:
128-
self.assertEqual(f.read(), content)
129-
130-
# download using explicit filepath as well
131-
os.remove(path)
132-
self.assertEqual(path, updater.download_target(info, path))
133-
self.assertEqual(path, updater.find_cached_target(info))
134-
self.assertEqual(path, updater.find_cached_target(info, path))
135-
13688
def test_fishy_rolenames(self) -> None:
13789
roles_to_filenames = {
13890
"../a": "..%2Fa.json",

0 commit comments

Comments
 (0)