Skip to content

Commit

Permalink
feat: allow sops binary path customization (#75)
Browse files Browse the repository at this point in the history
Refs #62
  • Loading branch information
nikaro authored Nov 15, 2024
1 parent de15909 commit c813826
Show file tree
Hide file tree
Showing 2 changed files with 37 additions and 6 deletions.
22 changes: 16 additions & 6 deletions src/sopsy/sopsy.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ def __init__(
self,
file: str | Path,
*,
binary_path: str | Path | None = None,
config: str | Path | None = None,
config_dict: dict[str, Any] | None = None,
extract: str | None = None,
Expand All @@ -64,7 +65,11 @@ def __init__(
Examples:
>>> from pathlib import Path
>>> from sopsy import Sops
>>> sops = Sops(Path("secrets.json"), config=Path(".config/sops.yml"))
>>> sops = Sops(
>>> binary_path="/app/bin/my_custom_sops",
>>> config=Path(".config/sops.yml"),
>>> file=Path("secrets.json"),
>>> )
Args:
file: Path to the SOPS file.
Expand All @@ -78,9 +83,14 @@ def __init__(
specified.
output_type: If not set, sops will use the input file's extension to
determine the output format.
binary_path: Path to the SOPS binary. If not defined it will search for it
in the PATH environment variable.
"""
self.bin: Path = Path("sops")
self.file: Path = Path(file).resolve(strict=True)
self.global_args: list[str] = []
if binary_path:
self.bin = Path(binary_path)
if extract:
self.global_args.extend(["--extract", extract])
if in_place:
Expand All @@ -102,9 +112,9 @@ def __init__(
config_tmp = fp.name
self.global_args.extend(["--config", config_tmp])

if not shutil.which("sops"):
if not shutil.which(self.bin):
msg = (
"sops command not found, "
f"{self.bin} command not found, "
"you may need to install it and/or add it to your PATH"
)
raise SopsyCommandNotFoundError(msg)
Expand All @@ -124,7 +134,7 @@ def decrypt(self, *, to_dict: bool = True) -> bytes | dict[str, Any] | None:
Returns:
The output of the sops command.
"""
cmd = ["sops", "--decrypt", *self.global_args, str(self.file)]
cmd = [str(self.bin), "--decrypt", *self.global_args, str(self.file)]
return run_cmd(cmd, to_dict=to_dict)

def encrypt(self, *, to_dict: bool = True) -> bytes | dict[str, Any] | None:
Expand All @@ -145,7 +155,7 @@ def encrypt(self, *, to_dict: bool = True) -> bytes | dict[str, Any] | None:
Returns:
The output of the sops command.
"""
cmd = ["sops", "--encrypt", *self.global_args, str(self.file)]
cmd = [str(self.bin), "--encrypt", *self.global_args, str(self.file)]
return run_cmd(cmd, to_dict=to_dict)

def get(self, key: str, *, default: Any = None) -> Any: # noqa: ANN401
Expand Down Expand Up @@ -183,5 +193,5 @@ def rotate(self, *, to_dict: bool = True) -> bytes | dict[str, Any] | None:
Returns:
The output of the sops command.
"""
cmd = ["sops", "--rotate", *self.global_args, str(self.file)]
cmd = [str(self.bin), "--rotate", *self.global_args, str(self.file)]
return run_cmd(cmd, to_dict=to_dict)
21 changes: 21 additions & 0 deletions tests/test_sopsy.py
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,23 @@ def test_sops_init_outputtype(tmp_path: Path) -> None:
assert "yaml" in s.global_args


def test_sops_init_default_bin(tmp_path: Path) -> None:
"""Test sops.Sops.__init__ function with binary_path arg."""
sops_file = tmp_path / "secret.json"
_ = sops_file.write_text('{"hello":"world"}')
s = sopsy.Sops(sops_file)
assert s.bin.name == "sops"


def test_sops_init_custom_bin(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test sops.Sops.__init__ function with binary_path arg."""
sops_file = tmp_path / "secret.json"
_ = sops_file.write_text('{"hello":"world"}')
monkeypatch.setattr(shutil, "which", _return_sops_path)
s = sopsy.Sops(sops_file, binary_path="/path/to/mysops")
assert s.bin.name == "mysops"


def test_sops_not_found(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test sops.Sops.__init__ function command not found."""
sops_file = tmp_path / "secret.json"
Expand Down Expand Up @@ -401,6 +418,10 @@ def _not_return_sops_path(*_args: Any, **_kwargs: Any) -> None:
return None


def _return_sops_path(*args: Any, **_kwargs: Any) -> Path:
return Path(*args)


def _mock_subprocess_run(*_args: Any, **_kwargs: Any) -> object:
return subprocess.CompletedProcess(
args=[], returncode=0, stdout=b'{"hello": "world"}'
Expand Down

0 comments on commit c813826

Please sign in to comment.