Skip to content
Snippets Groups Projects

make productfectcher use default threads capo property

Merged Charlotte Hausman requested to merge productfetcher_tweaks into main
1 file
+ 23
21
Compare changes
  • Side-by-side
  • Inline
@@ -88,27 +88,29 @@ def test_multiple_locator_fetching(capsys, resource_path_root):
:param capsys:
:return:
"""
img = resource_path_root / "location_files" / "IMG.json"
cal = resource_path_root / "location_files" / "CALIBRATION.json"
# parse the command line with these two
fc = FetchContext.parse_commandline([CLIParam.FILE.value, str(img), CLIParam.FILE.value, str(cal)])
assert len(fc.locators) == 2
# let's make the plan and ensure we have all the stuff we expect from both
plan = fc.generate_plan()
# we'll need to open these files ourselves to figure out what fetchers we expect
for locator_file in [img, cal]:
for file in FileLocator(locator_file).locate().files:
seen = False
# there may be a more "test friendly" way of doing this, such as by asking the plan
# if it is fetching a certain file, but it seems like a lot of refactoring for this
# one test, so I'm going to leave it alone for now
for hostgroup_fetcher in plan.fetchers:
seen = seen or file in [fetcher.file for fetcher in hostgroup_fetcher.fetchers]
assert seen
with patch("productfetcher.product_fetcher.CapoConfig") as mocked_capo:
mocked_capo.return_value.settings.return_value.defaultThreadsPerHost = 1
img = resource_path_root / "location_files" / "IMG.json"
cal = resource_path_root / "location_files" / "CALIBRATION.json"
# parse the command line with these two
fc = FetchContext.parse_commandline([CLIParam.FILE.value, str(img), CLIParam.FILE.value, str(cal)])
assert len(fc.locators) == 2
# let's make the plan and ensure we have all the stuff we expect from both
plan = fc.generate_plan()
# we'll need to open these files ourselves to figure out what fetchers we expect
for locator_file in [img, cal]:
for file in FileLocator(locator_file).locate().files:
seen = False
# there may be a more "test friendly" way of doing this, such as by asking the plan
# if it is fetching a certain file, but it seems like a lot of refactoring for this
# one test, so I'm going to leave it alone for now
for hostgroup_fetcher in plan.fetchers:
seen = seen or file in [fetcher.file for fetcher in hostgroup_fetcher.fetchers]
assert seen
def test_argument_parsing(capsys):
Loading