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
+ 32
29
Compare changes
  • Side-by-side
  • Inline
@@ -21,6 +21,7 @@
from pathlib import Path
from typing import List, Union
from unittest.mock import patch
import pytest
from productfetcher.fetchers import DryRunFakeFileFetcher, ForceMode
@@ -117,32 +118,34 @@ def test_argument_parsing(capsys):
:param capsys:
:return:
"""
# ensure that various combinations do not work
with pytest.raises(SystemExit):
# must have an SPL or a file
FetchContext.parse_commandline([])
# check the dry run value
fc = FetchContext.parse_commandline([CLIParam.DRY.value, CLIParam.FILE.value, "foo"])
assert fc.dry_run
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo"])
assert not fc.dry_run
# check the force value
fc = FetchContext.parse_commandline([CLIParam.FORCE.value, CLIParam.FILE.value, "foo"])
assert fc.force == ForceMode.FORCE
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo"])
assert fc.force == ForceMode.NORMAL
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo"])
assert isinstance(fc.locators[0], FileLocator)
assert fc.locators[0].file == Path("foo")
fc = FetchContext.parse_commandline([CLIParam.SPL.value, "uid://this/is/a/fakesy"])
assert isinstance(fc.locators[0], ServiceLocator)
assert fc.locators[0].spl == "uid://this/is/a/fakesy"
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo", CLIParam.CONCURRENCY.value, "732"])
assert fc.concurrency == 732
capsys.readouterr()
with patch("productfetcher.product_fetcher.CapoConfig") as mocked_capo:
mocked_capo.return_value.settings.return_value.defaultThreadsPerHost = 1
# ensure that various combinations do not work
with pytest.raises(SystemExit):
# must have an SPL or a file
FetchContext.parse_commandline([])
# check the dry run value
fc = FetchContext.parse_commandline([CLIParam.DRY.value, CLIParam.FILE.value, "foo"])
assert fc.dry_run
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo"])
assert not fc.dry_run
# check the force value
fc = FetchContext.parse_commandline([CLIParam.FORCE.value, CLIParam.FILE.value, "foo"])
assert fc.force == ForceMode.FORCE
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo"])
assert fc.force == ForceMode.NORMAL
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo"])
assert isinstance(fc.locators[0], FileLocator)
assert fc.locators[0].file == Path("foo")
fc = FetchContext.parse_commandline([CLIParam.SPL.value, "uid://this/is/a/fakesy"])
assert isinstance(fc.locators[0], ServiceLocator)
assert fc.locators[0].spl == "uid://this/is/a/fakesy"
fc = FetchContext.parse_commandline([CLIParam.FILE.value, "foo", CLIParam.CONCURRENCY.value, "732"])
assert fc.concurrency == 732
capsys.readouterr()
Loading