arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-04-24 07:17:17 +00:00
Code Issues Projects Releases Wiki Activity

make exception optional argument for check_output method

Browse Source
This commit is contained in:
Evgenii Alekseev 2022-11-10 18:05:32 +02:00
parent b75bd30658
commit 791ce4f242
10 changed files with 52 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/ahriman/core/alpm/repo.py
View File

@ -81,12 +81,8 @@ class Repo(LazyLogging):
""" """
create empty repository database create empty repository database
""" """
Repo._check_output( Repo._check_output("repo-add", *self.sign_args, str(self.repo_path),
"repo-add", *self.sign_args, str(self.repo_path), cwd=self.paths.repository, logger=self.logger, user=self.uid)
exception=None,
cwd=self.paths.repository,
logger=self.logger,
user=self.uid)
def remove(self, package: str, filename: Path) -> None: def remove(self, package: str, filename: Path) -> None:
""" """

27
src/ahriman/core/build_tools/sources.py
View File

@ -82,22 +82,18 @@ class Sources(LazyLogging):
branch = remote.branch if remote is not None else instance.DEFAULT_BRANCH branch = remote.branch if remote is not None else instance.DEFAULT_BRANCH
if is_initialized_git: if is_initialized_git:
instance.logger.info("update HEAD to remote at %s using branch %s", sources_dir, branch) instance.logger.info("update HEAD to remote at %s using branch %s", sources_dir, branch)
Sources._check_output("git", "fetch", "origin", branch, Sources._check_output("git", "fetch", "origin", branch, cwd=sources_dir, logger=instance.logger)
exception=None, cwd=sources_dir, logger=instance.logger)
elif remote is not None: elif remote is not None:
instance.logger.info("clone remote %s to %s using branch %s", remote.git_url, sources_dir, branch) instance.logger.info("clone remote %s to %s using branch %s", remote.git_url, sources_dir, branch)
Sources._check_output("git", "clone", "--branch", branch, "--single-branch", Sources._check_output("git", "clone", "--branch", branch, "--single-branch",
remote.git_url, str(sources_dir), remote.git_url, str(sources_dir), cwd=sources_dir, logger=instance.logger)
exception=None, cwd=sources_dir, logger=instance.logger)
else: else:
# it will cause an exception later # it will cause an exception later
instance.logger.error("%s is not initialized, but no remote provided", sources_dir) instance.logger.error("%s is not initialized, but no remote provided", sources_dir)
# and now force reset to our branch # and now force reset to our branch
Sources._check_output("git", "checkout", "--force", branch, Sources._check_output("git", "checkout", "--force", branch, cwd=sources_dir, logger=instance.logger)
exception=None, cwd=sources_dir, logger=instance.logger) Sources._check_output("git", "reset", "--hard", f"origin/{branch}", cwd=sources_dir, logger=instance.logger)
Sources._check_output("git", "reset", "--hard", f"origin/{branch}",
exception=None, cwd=sources_dir, logger=instance.logger)
# move content if required # move content if required
# we are using full path to source directory in order to make append possible # we are using full path to source directory in order to make append possible
@ -116,7 +112,7 @@ class Sources(LazyLogging):
bool: True in case if there is any remote and false otherwise bool: True in case if there is any remote and false otherwise
""" """
instance = Sources() instance = Sources()
remotes = Sources._check_output("git", "remote", exception=None, cwd=sources_dir, logger=instance.logger) remotes = Sources._check_output("git", "remote", cwd=sources_dir, logger=instance.logger)
return bool(remotes) return bool(remotes)
@staticmethod @staticmethod
@ -129,7 +125,7 @@ class Sources(LazyLogging):
""" """
instance = Sources() instance = Sources()
Sources._check_output("git", "init", "--initial-branch", instance.DEFAULT_BRANCH, Sources._check_output("git", "init", "--initial-branch", instance.DEFAULT_BRANCH,
exception=None, cwd=sources_dir, logger=instance.logger) cwd=sources_dir, logger=instance.logger)
@staticmethod @staticmethod
def load(sources_dir: Path, package: Package, patches: List[PkgbuildPatch], paths: RepositoryPaths) -> None: def load(sources_dir: Path, package: Package, patches: List[PkgbuildPatch], paths: RepositoryPaths) -> None:
@ -182,8 +178,7 @@ class Sources(LazyLogging):
instance = Sources() instance = Sources()
instance.add(sources_dir, *pattern) instance.add(sources_dir, *pattern)
instance.commit(sources_dir) instance.commit(sources_dir)
Sources._check_output("git", "push", remote.git_url, remote.branch, Sources._check_output("git", "push", remote.git_url, remote.branch, cwd=sources_dir, logger=instance.logger)
exception=None, cwd=sources_dir, logger=instance.logger)
def add(self, sources_dir: Path, *pattern: str) -> None: def add(self, sources_dir: Path, *pattern: str) -> None:
""" """
@ -203,7 +198,7 @@ class Sources(LazyLogging):
# add them to index # add them to index
Sources._check_output("git", "add", "--intent-to-add", Sources._check_output("git", "add", "--intent-to-add",
*[str(fn.relative_to(sources_dir)) for fn in found_files], *[str(fn.relative_to(sources_dir)) for fn in found_files],
exception=None, cwd=sources_dir, logger=self.logger) cwd=sources_dir, logger=self.logger)
def commit(self, sources_dir: Path, commit_message: Optional[str] = None) -> None: def commit(self, sources_dir: Path, commit_message: Optional[str] = None) -> None:
""" """
@ -217,7 +212,7 @@ class Sources(LazyLogging):
if commit_message is None: if commit_message is None:
commit_message = f"Autogenerated commit at {datetime.datetime.utcnow()}" commit_message = f"Autogenerated commit at {datetime.datetime.utcnow()}"
Sources._check_output("git", "commit", "--allow-empty", "--message", commit_message, Sources._check_output("git", "commit", "--allow-empty", "--message", commit_message,
exception=None, cwd=sources_dir, logger=self.logger) cwd=sources_dir, logger=self.logger)
def diff(self, sources_dir: Path) -> str: def diff(self, sources_dir: Path) -> str:
""" """
@ -229,7 +224,7 @@ class Sources(LazyLogging):
Returns: Returns:
str: patch as plain string str: patch as plain string
""" """
return Sources._check_output("git", "diff", exception=None, cwd=sources_dir, logger=self.logger) return Sources._check_output("git", "diff", cwd=sources_dir, logger=self.logger)
def move(self, pkgbuild_dir: Path, sources_dir: Path) -> None: def move(self, pkgbuild_dir: Path, sources_dir: Path) -> None:
""" """
@ -258,6 +253,6 @@ class Sources(LazyLogging):
self.logger.info("apply patch %s from database at %s", patch.key, sources_dir) self.logger.info("apply patch %s from database at %s", patch.key, sources_dir)
if patch.is_plain_diff: if patch.is_plain_diff:
Sources._check_output("git", "apply", "--ignore-space-change", "--ignore-whitespace", Sources._check_output("git", "apply", "--ignore-space-change", "--ignore-whitespace",
exception=None, cwd=sources_dir, input_data=patch.serialize(), logger=self.logger) cwd=sources_dir, input_data=patch.serialize(), logger=self.logger)
else: else:
patch.write(sources_dir / "PKGBUILD") patch.write(sources_dir / "PKGBUILD")

2
src/ahriman/core/sign/gpg.py
View File

@ -138,7 +138,7 @@ class GPG(LazyLogging):
key(str): key ID to import key(str): key ID to import
""" """
key_body = self.key_download(server, key) key_body = self.key_download(server, key)
GPG._check_output("gpg", "--import", input_data=key_body, exception=None, logger=self.logger) GPG._check_output("gpg", "--import", input_data=key_body, logger=self.logger)
def process(self, path: Path, key: str) -> List[Path]: def process(self, path: Path, key: str) -> List[Path]:
""" """

2
src/ahriman/core/upload/rsync.py
View File

@ -58,4 +58,4 @@ class Rsync(Upload):
path(Path): local path to sync path(Path): local path to sync
built_packages(Iterable[Package]): list of packages which has just been built built_packages(Iterable[Package]): list of packages which has just been built
""" """
Rsync._check_output(*self.command, str(path), self.remote, exception=None, logger=self.logger) Rsync._check_output(*self.command, str(path), self.remote, logger=self.logger)

6
src/ahriman/core/util.py
View File

@ -37,7 +37,7 @@ __all__ = ["check_output", "check_user", "exception_response_text", "filter_json
"package_like", "pretty_datetime", "pretty_size", "safe_filename", "walk"] "package_like", "pretty_datetime", "pretty_size", "safe_filename", "walk"]
def check_output(*args: str, exception: Optional[Exception], cwd: Optional[Path] = None, def check_output(*args: str, exception: Optional[Exception] = None, cwd: Optional[Path] = None,
input_data: Optional[str] = None, logger: Optional[Logger] = None, user: Optional[int] = None) -> str: input_data: Optional[str] = None, logger: Optional[Logger] = None, user: Optional[int] = None) -> str:
""" """
subprocess wrapper subprocess wrapper
@ -59,7 +59,7 @@ def check_output(*args: str, exception: Optional[Exception], cwd: Optional[Path]
Examples: Examples:
Simply call the function:: Simply call the function::
>>> check_output("echo", "hello world", exception=None) >>> check_output("echo", "hello world")
The more complicated calls which include result logging and input data are also possible:: The more complicated calls which include result logging and input data are also possible::
@ -67,7 +67,7 @@ def check_output(*args: str, exception: Optional[Exception], cwd: Optional[Path]
>>> >>>
>>> logger = logging.getLogger() >>> logger = logging.getLogger()
>>> check_output("python", "-c", "greeting = input('say hello: '); print(); print(greeting)", >>> check_output("python", "-c", "greeting = input('say hello: '); print(); print(greeting)",
>>> exception=None, input_data="hello world", logger=logger) >>> input_data="hello world", logger=logger)
An additional argument ``exception`` can be supplied in order to override the default exception:: An additional argument ``exception`` can be supplied in order to override the default exception::

10
src/ahriman/models/package.py
View File

@ -183,7 +183,7 @@ class Package(LazyLogging):
Raises: Raises:
InvalidPackageInfo: if there are parsing errors InvalidPackageInfo: if there are parsing errors
""" """
srcinfo_source = Package._check_output("makepkg", "--printsrcinfo", exception=None, cwd=path) srcinfo_source = Package._check_output("makepkg", "--printsrcinfo", cwd=path)
srcinfo, errors = parse_srcinfo(srcinfo_source) srcinfo, errors = parse_srcinfo(srcinfo_source)
if errors: if errors:
raise PackageInfoError(errors) raise PackageInfoError(errors)
@ -254,7 +254,7 @@ class Package(LazyLogging):
package_name = package_name.split(symbol)[0] package_name = package_name.split(symbol)[0]
return package_name return package_name
srcinfo_source = Package._check_output("makepkg", "--printsrcinfo", exception=None, cwd=path) srcinfo_source = Package._check_output("makepkg", "--printsrcinfo", cwd=path)
srcinfo, errors = parse_srcinfo(srcinfo_source) srcinfo, errors = parse_srcinfo(srcinfo_source)
if errors: if errors:
raise PackageInfoError(errors) raise PackageInfoError(errors)
@ -281,7 +281,7 @@ class Package(LazyLogging):
Raises: Raises:
InvalidPackageInfo: if there are parsing errors InvalidPackageInfo: if there are parsing errors
""" """
srcinfo_source = Package._check_output("makepkg", "--printsrcinfo", exception=None, cwd=path) srcinfo_source = Package._check_output("makepkg", "--printsrcinfo", cwd=path)
srcinfo, errors = parse_srcinfo(srcinfo_source) srcinfo, errors = parse_srcinfo(srcinfo_source)
if errors: if errors:
raise PackageInfoError(errors) raise PackageInfoError(errors)
@ -310,10 +310,10 @@ class Package(LazyLogging):
try: try:
# update pkgver first # update pkgver first
Package._check_output("makepkg", "--nodeps", "--nobuild", Package._check_output("makepkg", "--nodeps", "--nobuild",
exception=None, cwd=paths.cache_for(self.base), logger=self.logger) cwd=paths.cache_for(self.base), logger=self.logger)
# generate new .SRCINFO and put it to parser # generate new .SRCINFO and put it to parser
srcinfo_source = Package._check_output("makepkg", "--printsrcinfo", srcinfo_source = Package._check_output("makepkg", "--printsrcinfo",
exception=None, cwd=paths.cache_for(self.base), logger=self.logger) cwd=paths.cache_for(self.base), logger=self.logger)
srcinfo, errors = parse_srcinfo(srcinfo_source) srcinfo, errors = parse_srcinfo(srcinfo_source)
if errors: if errors:
raise PackageInfoError(errors) raise PackageInfoError(errors)

41
tests/ahriman/core/build_tools/test_sources.py
View File

@ -55,12 +55,10 @@ def test_fetch_existing(remote_source: RemoteSource, mocker: MockerFixture) -> N
local = Path("local") local = Path("local")
Sources.fetch(local, remote_source) Sources.fetch(local, remote_source)
check_output_mock.assert_has_calls([ check_output_mock.assert_has_calls([
MockCall("git", "fetch", "origin", remote_source.branch, MockCall("git", "fetch", "origin", remote_source.branch, cwd=local, logger=pytest.helpers.anyvar(int)),
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)), MockCall("git", "checkout", "--force", remote_source.branch, cwd=local, logger=pytest.helpers.anyvar(int)),
MockCall("git", "checkout", "--force", remote_source.branch,
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)),
MockCall("git", "reset", "--hard", f"origin/{remote_source.branch}", MockCall("git", "reset", "--hard", f"origin/{remote_source.branch}",
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)) cwd=local, logger=pytest.helpers.anyvar(int)),
]) ])
move_mock.assert_called_once_with(local.resolve(), local) move_mock.assert_called_once_with(local.resolve(), local)
@ -77,11 +75,10 @@ def test_fetch_new(remote_source: RemoteSource, mocker: MockerFixture) -> None:
Sources.fetch(local, remote_source) Sources.fetch(local, remote_source)
check_output_mock.assert_has_calls([ check_output_mock.assert_has_calls([
MockCall("git", "clone", "--branch", remote_source.branch, "--single-branch", MockCall("git", "clone", "--branch", remote_source.branch, "--single-branch",
remote_source.git_url, str(local), exception=None, cwd=local, logger=pytest.helpers.anyvar(int)), remote_source.git_url, str(local), cwd=local, logger=pytest.helpers.anyvar(int)),
MockCall("git", "checkout", "--force", remote_source.branch, MockCall("git", "checkout", "--force", remote_source.branch, cwd=local, logger=pytest.helpers.anyvar(int)),
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)),
MockCall("git", "reset", "--hard", f"origin/{remote_source.branch}", MockCall("git", "reset", "--hard", f"origin/{remote_source.branch}",
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)) cwd=local, logger=pytest.helpers.anyvar(int))
]) ])
move_mock.assert_called_once_with(local.resolve(), local) move_mock.assert_called_once_with(local.resolve(), local)
@ -97,10 +94,9 @@ def test_fetch_new_without_remote(mocker: MockerFixture) -> None:
local = Path("local") local = Path("local")
Sources.fetch(local, None) Sources.fetch(local, None)
check_output_mock.assert_has_calls([ check_output_mock.assert_has_calls([
MockCall("git", "checkout", "--force", Sources.DEFAULT_BRANCH, MockCall("git", "checkout", "--force", Sources.DEFAULT_BRANCH, cwd=local, logger=pytest.helpers.anyvar(int)),
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)),
MockCall("git", "reset", "--hard", f"origin/{Sources.DEFAULT_BRANCH}", MockCall("git", "reset", "--hard", f"origin/{Sources.DEFAULT_BRANCH}",
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)) cwd=local, logger=pytest.helpers.anyvar(int))
]) ])
move_mock.assert_called_once_with(local.resolve(), local) move_mock.assert_called_once_with(local.resolve(), local)
@ -124,8 +120,7 @@ def test_has_remotes(mocker: MockerFixture) -> None:
local = Path("local") local = Path("local")
assert Sources.has_remotes(local) assert Sources.has_remotes(local)
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with("git", "remote", cwd=local, logger=pytest.helpers.anyvar(int))
"git", "remote", exception=None, cwd=local, logger=pytest.helpers.anyvar(int))
def test_has_remotes_empty(mocker: MockerFixture) -> None: def test_has_remotes_empty(mocker: MockerFixture) -> None:
@ -145,7 +140,7 @@ def test_init(mocker: MockerFixture) -> None:
local = Path("local") local = Path("local")
Sources.init(local) Sources.init(local)
check_output_mock.assert_called_once_with("git", "init", "--initial-branch", Sources.DEFAULT_BRANCH, check_output_mock.assert_called_once_with("git", "init", "--initial-branch", Sources.DEFAULT_BRANCH,
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)) cwd=local, logger=pytest.helpers.anyvar(int))
def test_load(package_ahriman: Package, repository_paths: RepositoryPaths, mocker: MockerFixture) -> None: def test_load(package_ahriman: Package, repository_paths: RepositoryPaths, mocker: MockerFixture) -> None:
@ -225,7 +220,7 @@ def test_push(package_ahriman: Package, mocker: MockerFixture) -> None:
commit_mock.assert_called_once_with(local) commit_mock.assert_called_once_with(local)
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with(
"git", "push", package_ahriman.remote.git_url, package_ahriman.remote.branch, "git", "push", package_ahriman.remote.git_url, package_ahriman.remote.branch,
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)) cwd=local, logger=pytest.helpers.anyvar(int))
def test_add(sources: Sources, mocker: MockerFixture) -> None: def test_add(sources: Sources, mocker: MockerFixture) -> None:
@ -239,8 +234,8 @@ def test_add(sources: Sources, mocker: MockerFixture) -> None:
sources.add(local, "pattern1", "pattern2") sources.add(local, "pattern1", "pattern2")
glob_mock.assert_has_calls([MockCall("pattern1"), MockCall("pattern2")]) glob_mock.assert_has_calls([MockCall("pattern1"), MockCall("pattern2")])
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with(
"git", "add", "--intent-to-add", "1", "2", "1", "2", "git", "add", "--intent-to-add", "1", "2", "1", "2", cwd=local, logger=pytest.helpers.anyvar(int)
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)) )
def test_add_skip(sources: Sources, mocker: MockerFixture) -> None: def test_add_skip(sources: Sources, mocker: MockerFixture) -> None:
@ -264,8 +259,7 @@ def test_commit(sources: Sources, mocker: MockerFixture) -> None:
commit_message = "Commit message" commit_message = "Commit message"
sources.commit(local, commit_message=commit_message) sources.commit(local, commit_message=commit_message)
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with(
"git", "commit", "--allow-empty", "--message", commit_message, "git", "commit", "--allow-empty", "--message", commit_message, cwd=local, logger=pytest.helpers.anyvar(int)
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)
) )
@ -279,7 +273,7 @@ def test_commit_autogenerated(sources: Sources, mocker: MockerFixture) -> None:
sources.commit(Path("local")) sources.commit(Path("local"))
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with(
"git", "commit", "--allow-empty", "--message", pytest.helpers.anyvar(str, strict=True), "git", "commit", "--allow-empty", "--message", pytest.helpers.anyvar(str, strict=True),
exception=None, cwd=local, logger=pytest.helpers.anyvar(int) cwd=local, logger=pytest.helpers.anyvar(int)
) )
@ -291,8 +285,7 @@ def test_diff(sources: Sources, mocker: MockerFixture) -> None:
local = Path("local") local = Path("local")
assert sources.diff(local) assert sources.diff(local)
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with("git", "diff", cwd=local, logger=pytest.helpers.anyvar(int))
"git", "diff", exception=None, cwd=local, logger=pytest.helpers.anyvar(int))
def test_move(sources: Sources, mocker: MockerFixture) -> None: def test_move(sources: Sources, mocker: MockerFixture) -> None:
@ -326,7 +319,7 @@ def test_patch_apply(sources: Sources, mocker: MockerFixture) -> None:
sources.patch_apply(local, patch) sources.patch_apply(local, patch)
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with(
"git", "apply", "--ignore-space-change", "--ignore-whitespace", "git", "apply", "--ignore-space-change", "--ignore-whitespace",
exception=None, cwd=local, input_data=patch.value, logger=pytest.helpers.anyvar(int) cwd=local, input_data=patch.value, logger=pytest.helpers.anyvar(int)
) )

3
tests/ahriman/core/sign/test_gpg.py
View File

@ -105,8 +105,7 @@ def test_key_import(gpg: GPG, mocker: MockerFixture) -> None:
check_output_mock = mocker.patch("ahriman.core.sign.gpg.GPG._check_output") check_output_mock = mocker.patch("ahriman.core.sign.gpg.GPG._check_output")
gpg.key_import("pgp.mit.edu", "0xE989490C") gpg.key_import("pgp.mit.edu", "0xE989490C")
check_output_mock.assert_called_once_with( check_output_mock.assert_called_once_with("gpg", "--import", input_data="key", logger=pytest.helpers.anyvar(int))
"gpg", "--import", input_data="key", exception=None, logger=pytest.helpers.anyvar(int))
def test_process(gpg_with_key: GPG, mocker: MockerFixture) -> None: def test_process(gpg_with_key: GPG, mocker: MockerFixture) -> None:

20
tests/ahriman/core/test_util.py
View File

@ -22,10 +22,10 @@ def test_check_output(mocker: MockerFixture) -> None:
""" """
logger_mock = mocker.patch("logging.Logger.debug") logger_mock = mocker.patch("logging.Logger.debug")
assert check_output("echo", "hello", exception=None) == "hello" assert check_output("echo", "hello") == "hello"
logger_mock.assert_not_called() logger_mock.assert_not_called()
assert check_output("echo", "hello", exception=None, logger=logging.getLogger("")) == "hello" assert check_output("echo", "hello", logger=logging.getLogger("")) == "hello"
logger_mock.assert_called_once_with("hello") logger_mock.assert_called_once_with("hello")
@ -35,11 +35,11 @@ def test_check_output_stderr(mocker: MockerFixture) -> None:
""" """
logger_mock = mocker.patch("logging.Logger.debug") logger_mock = mocker.patch("logging.Logger.debug")
assert check_output("python", "-c", """import sys; print("hello", file=sys.stderr)""", exception=None) == "" assert check_output("python", "-c", """import sys; print("hello", file=sys.stderr)""") == ""
logger_mock.assert_not_called() logger_mock.assert_not_called()
assert check_output("python", "-c", """import sys; print("hello", file=sys.stderr)""", assert check_output("python", "-c", """import sys; print("hello", file=sys.stderr)""",
exception=None, logger=logging.getLogger("")) == "" logger=logging.getLogger("")) == ""
logger_mock.assert_called_once_with("hello") logger_mock.assert_called_once_with("hello")
@ -48,7 +48,7 @@ def test_check_output_with_stdin() -> None:
must run command and put string to stdin must run command and put string to stdin
""" """
assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)", assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)",
exception=None, input_data="single line") == "single line" input_data="single line") == "single line"
def test_check_output_with_stdin_newline() -> None: def test_check_output_with_stdin_newline() -> None:
@ -56,7 +56,7 @@ def test_check_output_with_stdin_newline() -> None:
must run command and put string to stdin ending with new line must run command and put string to stdin ending with new line
""" """
assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)", assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)",
exception=None, input_data="single line\n") == "single line" input_data="single line\n") == "single line"
def test_check_output_multiple_with_stdin() -> None: def test_check_output_multiple_with_stdin() -> None:
@ -64,7 +64,7 @@ def test_check_output_multiple_with_stdin() -> None:
must run command and put multiple lines to stdin must run command and put multiple lines to stdin
""" """
assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)", assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)",
exception=None, input_data="multiple\nlines") == "multiple\nlines" input_data="multiple\nlines") == "multiple\nlines"
def test_check_output_multiple_with_stdin_newline() -> None: def test_check_output_multiple_with_stdin_newline() -> None:
@ -72,7 +72,7 @@ def test_check_output_multiple_with_stdin_newline() -> None:
must run command and put multiple lines to stdin with new line at the end must run command and put multiple lines to stdin with new line at the end
""" """
assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)", assert check_output("python", "-c", "import sys; value = sys.stdin.read(); print(value)",
exception=None, input_data="multiple\nlines\n") == "multiple\nlines" input_data="multiple\nlines\n") == "multiple\nlines"
def test_check_output_failure(mocker: MockerFixture) -> None: def test_check_output_failure(mocker: MockerFixture) -> None:
@ -82,10 +82,10 @@ def test_check_output_failure(mocker: MockerFixture) -> None:
mocker.patch("subprocess.Popen.wait", return_value=1) mocker.patch("subprocess.Popen.wait", return_value=1)
with pytest.raises(subprocess.CalledProcessError): with pytest.raises(subprocess.CalledProcessError):
check_output("echo", "hello", exception=None) check_output("echo", "hello")
with pytest.raises(subprocess.CalledProcessError): with pytest.raises(subprocess.CalledProcessError):
check_output("echo", "hello", exception=None, logger=logging.getLogger("")) check_output("echo", "hello", logger=logging.getLogger(""))
def test_check_output_failure_exception(mocker: MockerFixture) -> None: def test_check_output_failure_exception(mocker: MockerFixture) -> None:

2
tests/ahriman/core/upload/test_rsync.py
View File

@ -10,4 +10,4 @@ def test_sync(rsync: Rsync, mocker: MockerFixture) -> None:
""" """
check_output_mock = mocker.patch("ahriman.core.upload.rsync.Rsync._check_output") check_output_mock = mocker.patch("ahriman.core.upload.rsync.Rsync._check_output")
rsync.sync(Path("path"), []) rsync.sync(Path("path"), [])
check_output_mock.assert_called_once_with(*rsync.command, "path", rsync.remote, exception=None, logger=rsync.logger) check_output_mock.assert_called_once_with(*rsync.command, "path", rsync.remote, logger=rsync.logger)