Skip to content

Commit

Permalink
fix: match os.PathLike too
Browse files Browse the repository at this point in the history
Signed-off-by: Henry Schreiner <[email protected]>
  • Loading branch information
henryiii committed Dec 2, 2022
1 parent 1f6bd5f commit 2a9f934
Show file tree
Hide file tree
Showing 3 changed files with 34 additions and 19 deletions.
3 changes: 2 additions & 1 deletion pytest_subprocess/types.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import asyncio
import io
import os
from typing import Sequence
from typing import Union

Expand All @@ -14,5 +15,5 @@
Sequence[Union[str, bytes]],
]
BUFFER = Union[io.BytesIO, io.StringIO, asyncio.StreamReader]
ARGUMENT = Union[str, Any]
ARGUMENT = Union[str, Any, os.PathLike]
COMMAND = Union[Sequence[ARGUMENT], str, Command]
39 changes: 21 additions & 18 deletions pytest_subprocess/utils.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,13 @@
import os
import threading
from typing import Any as AnyType
from typing import Iterator
from typing import List
from typing import Optional
from typing import Sequence
from typing import Tuple
from typing import Union

ARGUMENT = Union[str, "Any"]
ARGUMENT = Union[str, "Any", os.PathLike]


class Thread(threading.Thread):
Expand All @@ -33,11 +33,12 @@ def __init__(
):
if isinstance(command, str):
command = tuple(command.split(" "))
if isinstance(command, list):
command = tuple(command)
elif not isinstance(command, tuple):
if not isinstance(command, (list, tuple)):
raise TypeError("Command can be only of type string, list or tuple.")
self.command: Tuple[Union[str, Any], ...] = command

self.command: Tuple[ARGUMENT, ...] = tuple(
os.fspath(c) if isinstance(c, os.PathLike) else c for c in command
)

for (i, command_elem) in enumerate(self.command):
if isinstance(command_elem, Any) and isinstance(
Expand All @@ -48,24 +49,26 @@ def __init__(
def __eq__(self, other: AnyType) -> bool:
if isinstance(other, str):
other = other.split(" ")
elif isinstance(other, tuple):
other = list(other)

if other == list(self.command):
norm_command = [
os.fspath(c) if isinstance(c, os.PathLike) else c for c in self.command
]
norm_other = [os.fspath(c) if isinstance(c, os.PathLike) else c for c in other]

if norm_other == norm_command:
# straightforward matching
return True

other = list(other)
for (i, command_elem) in enumerate(self.command):
for (i, command_elem) in enumerate(norm_command):
if isinstance(command_elem, Any):
next_command_elem = self._get_next_command_elem(i)
if next_command_elem is None:
if not self._are_thresholds_ok(command_elem, len(other)):
if not self._are_thresholds_ok(command_elem, len(norm_other)):
return False
return True
else:
next_matching_elem = self._get_next_matching_elem_index(
other, next_command_elem
norm_other, next_command_elem
)
if next_matching_elem is None:
return False
Expand All @@ -75,12 +78,12 @@ def __eq__(self, other: AnyType) -> bool:
):
return False

other = other[next_matching_elem:]
norm_other = norm_other[next_matching_elem:]
else:
if len(other) == 0 or other.pop(0) != command_elem:
if len(norm_other) == 0 or norm_other.pop(0) != command_elem:
return False

return len(other) == 0
return len(norm_other) == 0

def __iter__(self) -> Iterator:
return iter(self.command)
Expand All @@ -93,15 +96,15 @@ def _are_thresholds_ok(command_elem: "Any", value: int) -> bool:
return False
return True

def _get_next_command_elem(self, index: int) -> Optional[Union[str, "Any"]]:
def _get_next_command_elem(self, index: int) -> Optional[ARGUMENT]:
try:
return self.command[index + 1]
except IndexError:
return None

@staticmethod
def _get_next_matching_elem_index(
other: List[Union[str, "Any"]], elem: Union[str, "Any"]
other: Sequence[ARGUMENT], elem: ARGUMENT
) -> Optional[int]:
return next(
(i for i, other_elem in enumerate(other) if elem == other_elem), None
Expand Down
11 changes: 11 additions & 0 deletions tests/test_subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import subprocess
import sys
import time
from pathlib import Path

import pytest

Expand Down Expand Up @@ -43,6 +44,16 @@ def test_completedprocess_args(fp, cmd):
assert isinstance(proc.args, type(cmd))


@pytest.mark.parametrize("p1,p2", [(Path, str), (str, Path), (Path, Path)])
def test_completedprocess_args_path(fp, p1, p2):
fp.register([p1("cmd")])

proc = subprocess.run([p2("cmd")], check=True)

assert proc.args == [p2("cmd")]
assert isinstance(proc.args[0], p2)


@pytest.mark.parametrize("cmd", [("cmd"), ["cmd"]])
def test_called_process_error(fp, cmd):
fp.register(cmd, returncode=1)
Expand Down

0 comments on commit 2a9f934

Please sign in to comment.