Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

change a few conf settings to getters #8

Merged
merged 2 commits into from
May 13, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions src/django_opfield/conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,15 +34,18 @@ 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:
path: str | None = None

if user_cli_path := self.OP_CLI_PATH:
path = user_cli_path
else:
path = shutil.which("op")
Expand All @@ -52,9 +55,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")
Expand Down
4 changes: 2 additions & 2 deletions src/django_opfield/fields.py
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand Down
1 change: 1 addition & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ def pytest_configure(config):

TEST_SETTINGS = {
"INSTALLED_APPS": [
"django_opfield",
"tests",
]
}
2 changes: 1 addition & 1 deletion tests/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,5 +5,5 @@
from django_opfield.fields import OPField


class TestModel(models.Model):
class OPFieldModel(models.Model):
op_uri = OPField()
38 changes: 31 additions & 7 deletions tests/test_conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -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"
18 changes: 9 additions & 9 deletions tests/test_fields.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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"
Expand All @@ -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()

Expand All @@ -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()
Expand All @@ -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
Expand Down