allow for setting the path to the op command (#1)

.github/workflows/test.yml

@@ -53,19 +53,6 @@ jobs:
 
       - uses: 1password/install-cli-action@v1
 
-      - run: |
-          which op
-          op --version
-          echo ${OP_CLI_PATH}
-
-      - shell: python -u {0}
-        run: |
-          import shutil
-          import subprocess
-
-          subprocess.check_call(["op", "--version"])
-          shutil.which("op")
-
       - name: Run tests
         run: |
           python -m nox --session "tests(python='${{ matrix.python-version }}', django='${{ matrix.django-version }}')"

noxfile.py

@@ -71,16 +71,10 @@ def tests(session, django):
     else:
         session.install(f"django=={django}")
 
-    import shutil
-    import subprocess
-
-    subprocess.check_call(["op", "--version"])
-    shutil.which("op")
-
     if session.posargs:
-        session.run("python", "-m", "pytest", *session.posargs, external=True)
+        session.run("python", "-m", "pytest", *session.posargs)
     else:
-        session.run("python", "-m", "pytest", external=True)
+        session.run("python", "-m", "pytest")
 
 
 @nox.session

src/django_opfield/conf.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import os
+import shutil
 import sys
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any
 
 from django.conf import settings
@@ -24,6 +26,21 @@ def __getattribute__(self, __name: str) -> Any:
         user_settings = getattr(settings, OPFIELD_SETTINGS_NAME, {})
         return user_settings.get(__name, super().__getattribute__(__name))
 
+    @property
+    def OP_CLI_PATH(self) -> Path:
+        user_settings = getattr(settings, OPFIELD_SETTINGS_NAME, {})
+        if user_cli_path := user_settings.get("OP_CLI_PATH", None):
+            path = user_cli_path
+        elif env_cli_path := os.environ.get("OP_CLI_PATH", None):
+            path = env_cli_path
+        else:
+            path = shutil.which("op")
+
+        if not path:
+            raise ImportError("Could not find the 'op' CLI command")
+
+        return Path(path).resolve()
+
     @property
     def OP_SERVICE_ACCOUNT_TOKEN(self) -> str:
         return os.environ.get("OP_SERVICE_ACCOUNT_TOKEN", "")

src/django_opfield/fields.py

@@ -43,12 +43,11 @@ def get_secret(self: models.Model) -> str | None:
             if not app_settings.OP_SERVICE_ACCOUNT_TOKEN:
                 raise ValueError("OP_SERVICE_ACCOUNT_TOKEN is not set")
             try:
-                _ = subprocess.check_call(["op", "--version"])
-            except subprocess.CalledProcessError as err:
-                msg = "The 'op' CLI command is not available"
-                raise OSError(msg) from err
+                op = app_settings.OP_CLI_PATH
+            except ImportError as err:
+                raise err
             op_uri = getattr(self, name)
-            result = subprocess.run(["op", "read", op_uri], capture_output=True)
+            result = subprocess.run([op, "read", op_uri], capture_output=True)
             if result.returncode != 0:
                 raise ValueError(
                     f"Could not read secret from 1Password: {result.stderr.decode('utf-8')}"

tests/test_conf.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import os
+from unittest.mock import patch
+
+import pytest
+from django.test import override_settings
+
+from django_opfield.conf import app_settings
+
+
+@patch.dict(os.environ, {"OP_CLI_PATH": ""})
+class TestOPCliPath:
+    @patch("shutil.which")
+    def test_default(self, mock_which):
+        mock_which.return_value = None
+
+        with pytest.raises(ImportError):
+            assert app_settings.OP_CLI_PATH
+
+    @override_settings(DJANGO_OPFIELD={"OP_CLI_PATH": "path/to/op"})
+    def test_user_setting(self):
+        assert "path/to/op" in str(app_settings.OP_CLI_PATH)
+
+    @patch.dict(os.environ, {"OP_CLI_PATH": "path/to/op"})
+    def test_env_var(self):
+        assert "path/to/op" in str(app_settings.OP_CLI_PATH)
+
+    @patch("shutil.which")
+    def test_shutil_which(self, mock_which):
+        mock_which.return_value = "path/to/op"
+
+        assert "path/to/op" in str(app_settings.OP_CLI_PATH)

tests/test_fields.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-import subprocess
+from unittest.mock import ANY
 from unittest.mock import patch
 
 import pytest
@@ -78,7 +78,7 @@ def test_deconstruct_with_vaults():
 
 
 @patch("subprocess.run")
-@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"}, clear=True)
+@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"})
 def test_get_secret(mock_run):
     mock_run.return_value.returncode = 0
     mock_run.return_value.stdout = b"secret value"
@@ -88,7 +88,7 @@ def test_get_secret(mock_run):
     secret = model.op_uri_secret
 
     mock_run.assert_called_once_with(
-        ["op", "read", "op://vault/item/field"], capture_output=True
+        [ANY, "read", "op://vault/item/field"], capture_output=True
     )
     assert secret == "secret value"
 
@@ -107,7 +107,7 @@ def test_get_secret_no_token(mock_run):
 
 
 @patch("subprocess.run")
-@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"}, clear=True)
+@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"})
 def test_get_secret_error(mock_run):
     mock_run.return_value.returncode = 1
     mock_run.return_value.stderr = b"error message"
@@ -120,23 +120,21 @@ def test_get_secret_error(mock_run):
     assert "Could not read secret from 1Password" in str(exc_info.value)
 
 
-@patch("subprocess.check_call")
-@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"}, clear=True)
-def test_get_secret_command_not_available(mock_check_call, db):
-    mock_check_call.side_effect = subprocess.CalledProcessError(
-        returncode=1, cmd="op --version", output=b"Command not found"
-    )
+@patch("shutil.which")
+@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"})
+def test_get_secret_command_not_available(mock_which, db):
+    mock_which.return_value = None
 
     model = TestModel(op_uri="op://vault/item/field")
 
-    with pytest.raises(OSError) as excinfo:
+    with pytest.raises(ImportError) as excinfo:
         _ = model.op_uri_secret
 
-    assert "The 'op' CLI command is not available" in str(excinfo.value)
+    assert "Could not find the 'op' CLI command" in str(excinfo.value)
 
 
 @patch("subprocess.run")
-@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"}, clear=True)
+@patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"})
 def test_set_secret_failure(mock_run):
     model = TestModel(op_uri="op://vault/item/field")