diff --git a/src/django_opfield/conf.py b/src/django_opfield/conf.py index 3cd8fb9..7b79d04 100644 --- a/src/django_opfield/conf.py +++ b/src/django_opfield/conf.py @@ -34,15 +34,16 @@ def _get_user_setting(setting: str, fallback: Any = None) -> Any: @dataclass(frozen=True) class AppSettings: OP_COMMAND_TIMEOUT: int = 5 # in seconds + OP_CLI_PATH: str = "" + OP_SERVICE_ACCOUNT_TOKEN: str = "" @override def __getattribute__(self, __name: str) -> Any: user_setting = _get_user_setting(__name) return user_setting or super().__getattribute__(__name) - @property - def OP_CLI_PATH(self) -> Path: - if user_cli_path := _get_user_setting("OP_CLI_PATH"): + def get_op_cli_path(self) -> Path: + if user_cli_path := self.OP_CLI_PATH: path = user_cli_path else: path = shutil.which("op") @@ -52,9 +53,8 @@ def OP_CLI_PATH(self) -> Path: return Path(path).resolve() - @property - def OP_SERVICE_ACCOUNT_TOKEN(self) -> str: - token = _get_user_setting("OP_SERVICE_ACCOUNT_TOKEN") + def get_op_service_account_token(self) -> str: + token = self.OP_SERVICE_ACCOUNT_TOKEN if not token: raise ImproperlyConfigured("OP_SERVICE_ACCOUNT_TOKEN is not set") diff --git a/src/django_opfield/fields.py b/src/django_opfield/fields.py index 0485c03..800c733 100644 --- a/src/django_opfield/fields.py +++ b/src/django_opfield/fields.py @@ -40,9 +40,9 @@ def contribute_to_class( super().contribute_to_class(cls, name, private_only) def get_secret(self: models.Model) -> str | None: - op = app_settings.OP_CLI_PATH + op = app_settings.get_op_cli_path() op_uri = getattr(self, name) - op_token = app_settings.OP_SERVICE_ACCOUNT_TOKEN + op_token = app_settings.get_op_service_account_token() op_timeout = app_settings.OP_COMMAND_TIMEOUT result = subprocess.run( [op, "read", op_uri], diff --git a/tests/conftest.py b/tests/conftest.py index 246eea3..1084a00 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,6 +17,7 @@ def pytest_configure(config): TEST_SETTINGS = { "INSTALLED_APPS": [ + "django_opfield", "tests", ] } diff --git a/tests/models.py b/tests/models.py index f46c4ad..a800d92 100644 --- a/tests/models.py +++ b/tests/models.py @@ -5,5 +5,5 @@ from django_opfield.fields import OPField -class TestModel(models.Model): +class OPFieldModel(models.Model): op_uri = OPField() diff --git a/tests/test_conf.py b/tests/test_conf.py index 0ee1156..b9c6217 100644 --- a/tests/test_conf.py +++ b/tests/test_conf.py @@ -4,6 +4,7 @@ from unittest.mock import patch import pytest +from django.core.exceptions import ImproperlyConfigured from django.test import override_settings from django_opfield.conf import OPFIELD_SETTINGS_NAME @@ -24,28 +25,35 @@ def test_set_in_env(self): assert _get_user_setting("TEST_SETTING") == "test value" -@patch.dict(os.environ, {"OP_CLI_PATH": ""}) -class TestOPCliPath: +class TestGetOpCLIPath: + @pytest.fixture(autouse=True) + def setup(self): + env = {k: v for k, v in os.environ.items() if k not in {"OP_CLI_PATH"}} + with patch.dict(os.environ, env, clear=True): + yield + @patch("shutil.which") def test_default(self, mock_which): mock_which.return_value = None - with pytest.raises(ImportError): - assert app_settings.OP_CLI_PATH + with pytest.raises(ImportError) as exc_info: + assert app_settings.get_op_cli_path() + + assert "Could not find the 'op' CLI command" in str(exc_info.value) @override_settings(**{OPFIELD_SETTINGS_NAME: {"OP_CLI_PATH": "path/to/op"}}) def test_user_setting(self): - assert "path/to/op" in str(app_settings.OP_CLI_PATH) + assert "path/to/op" in str(app_settings.get_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) + assert "path/to/op" in str(app_settings.get_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) + assert "path/to/op" in str(app_settings.get_op_cli_path()) class TestOpCommandTimeout: @@ -55,3 +63,19 @@ def test_default(self): @override_settings(**{OPFIELD_SETTINGS_NAME: {"OP_COMMAND_TIMEOUT": 10}}) def test_user_setting(self): assert app_settings.OP_COMMAND_TIMEOUT == 10 + + +class TestGetOpServiceAccountToken: + def test_default(self): + with pytest.raises(ImproperlyConfigured) as exc_info: + assert app_settings.get_op_service_account_token() + + assert "OP_SERVICE_ACCOUNT_TOKEN is not set" in str(exc_info.value) + + @override_settings(**{OPFIELD_SETTINGS_NAME: {"OP_SERVICE_ACCOUNT_TOKEN": "token"}}) + def test_user_setting(self): + assert app_settings.get_op_service_account_token() == "token" + + @patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"}) + def test_env_var(self): + assert app_settings.get_op_service_account_token() == "token" diff --git a/tests/test_fields.py b/tests/test_fields.py index 068117d..5230b5d 100644 --- a/tests/test_fields.py +++ b/tests/test_fields.py @@ -15,7 +15,7 @@ from django_opfield.fields import OPField from django_opfield.validators import OPURIValidator -from .models import TestModel +from .models import OPFieldModel def test_init_with_defaults(): @@ -87,7 +87,7 @@ def test_get_secret(mock_run): mock_run.return_value.returncode = 0 mock_run.return_value.stdout = b"secret value" - model = TestModel(op_uri="op://vault/item/field") + model = OPFieldModel(op_uri="op://vault/item/field") secret = model.op_uri_secret @@ -105,7 +105,7 @@ def test_get_secret_no_token(mock_run): mock_run.return_value.returncode = 1 mock_run.return_value.stderr = b"error message" - model = TestModel(op_uri="op://vault/item/field") + model = OPFieldModel(op_uri="op://vault/item/field") with pytest.raises(ImproperlyConfigured) as exc_info: _ = model.op_uri_secret @@ -119,7 +119,7 @@ def test_get_secret_error(mock_run): mock_run.return_value.returncode = 1 mock_run.return_value.stderr = b"error message" - model = TestModel(op_uri="op://vault/item/field") + model = OPFieldModel(op_uri="op://vault/item/field") with pytest.raises(ValueError) as exc_info: _ = model.op_uri_secret @@ -132,7 +132,7 @@ def test_get_secret_error(mock_run): def test_get_secret_command_not_available(mock_which, db): mock_which.return_value = None - model = TestModel(op_uri="op://vault/item/field") + model = OPFieldModel(op_uri="op://vault/item/field") with pytest.raises(ImportError) as excinfo: _ = model.op_uri_secret @@ -143,7 +143,7 @@ def test_get_secret_command_not_available(mock_which, db): @patch("subprocess.run") @patch.dict(os.environ, {"OP_SERVICE_ACCOUNT_TOKEN": "token"}) def test_set_secret_failure(mock_run): - model = TestModel(op_uri="op://vault/item/field") + model = OPFieldModel(op_uri="op://vault/item/field") with pytest.raises(NotImplementedError) as exc_info: model.op_uri_secret = "new secret" @@ -160,7 +160,7 @@ def test_set_secret_failure(mock_run): ], ) def test_model_with_valid_op_uri(valid_uri, db): - model = TestModel(op_uri=valid_uri) + model = OPFieldModel(op_uri=valid_uri) model.full_clean() model.save() @@ -178,7 +178,7 @@ def test_model_with_valid_op_uri(valid_uri, db): ], ) def test_model_with_invalid_op_uri(invalid_uri, db): - model = TestModel(op_uri=invalid_uri) + model = OPFieldModel(op_uri=invalid_uri) with pytest.raises(ValidationError) as excinfo: model.full_clean() @@ -193,7 +193,7 @@ def test_model_with_invalid_op_uri(invalid_uri, db): def test_command_timeout(mock_run): mock_run.side_effect = TimeoutExpired(ANY, timeout=1) - model = TestModel(op_uri="op://vault/item/field") + model = OPFieldModel(op_uri="op://vault/item/field") with pytest.raises(TimeoutExpired): _ = model.op_uri_secret