Skip to content

Commit

Permalink
refactor with_secret to contribute_to_class
Browse files Browse the repository at this point in the history
  • Loading branch information
joshuadavidthomas committed May 11, 2024
1 parent 482b941 commit e4a01c9
Show file tree
Hide file tree
Showing 3 changed files with 33 additions and 32 deletions.
44 changes: 26 additions & 18 deletions src/django_opfield/fields.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,38 +22,46 @@ class OPField(models.CharField):
description = "1Password secret"

def __init__(
self, vaults: list[str] | None = None, *args: Any, **kwargs: Any
self,
vaults: list[str] | None = None,
secret_name: str | None = None,
*args: Any,
**kwargs: Any,
) -> None:
self.vaults = vaults
self.secret_name = secret_name
kwargs.setdefault("max_length", 255)
super().__init__(*args, **kwargs)
self.validators.append(OPURIValidator(vaults=self.vaults))

@classmethod
def with_secret(cls, *args: Any, **kwargs: Any) -> tuple[OPField, property]:
op_uri = cls(*args, **kwargs)
@override
def contribute_to_class(
self, cls: type[models.Model], name: str, private_only: bool = False
):
super().contribute_to_class(cls, name, private_only)

def secret_getter(self: models.Model) -> str | None:
def get_secret(self: models.Model) -> str | None:
if not app_settings.OP_SERVICE_ACCOUNT_TOKEN:
msg = "OP_SERVICE_ACCOUNT_TOKEN is not set"
raise ValueError(msg)
raise ValueError("OP_SERVICE_ACCOUNT_TOKEN is not set")
if shutil.which("op") is None:
msg = "The 'op' CLI command is not available"
raise OSError(msg)
field_value = getattr(self, op_uri.name)
result = subprocess.run(["op", "read", field_value], capture_output=True)
op_uri = getattr(self, name)
result = subprocess.run(["op", "read", op_uri], capture_output=True)
if result.returncode != 0:
err = result.stderr.decode("utf-8")
msg = f"Could not read secret from 1Password: {err}"
raise ValueError(msg)
secret = result.stdout.decode("utf-8").strip()
return secret
raise ValueError(
f"Could not read secret from 1Password: {result.stderr.decode('utf-8')}"
)
return result.stdout.decode("utf-8").strip()

def set_secret(self: models.Model, value: str) -> None:
raise NotImplementedError("OPField does not support setting secret value")

def secret_setter(self: models.Model, value: str) -> None:
raise NotImplementedError("OPField does not support setting secret values")
property_name = (
f"{name}_secret" if self.secret_name is None else self.secret_name
)

secret = property(secret_getter, secret_setter)
return op_uri, secret
setattr(cls, property_name, property(get_secret, set_secret))

@override
def deconstruct(self):
Expand Down
2 changes: 1 addition & 1 deletion tests/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,4 +6,4 @@


class TestModel(models.Model):
op_uri, op_secret = OPField.with_secret()
op_uri = OPField()
19 changes: 6 additions & 13 deletions tests/test_fields.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,13 +49,6 @@ def test_init_validator():
assert OPURIValidator in validators


def test_field_return():
field, secret = OPField.with_secret()

assert isinstance(field, OPField)
assert isinstance(secret, property)


def test_deconstruct_default():
field = OPField()

Expand Down Expand Up @@ -91,7 +84,7 @@ def test_get_secret(mock_run):

model = TestModel(op_uri="op://vault/item/field")

secret = model.op_secret
secret = model.op_uri_secret

mock_run.assert_called_once_with(
["op", "read", "op://vault/item/field"], capture_output=True
Expand All @@ -107,7 +100,7 @@ def test_get_secret_no_token(mock_run):
model = TestModel(op_uri="op://vault/item/field")

with pytest.raises(ValueError) as exc_info:
_ = model.op_secret
_ = model.op_uri_secret

assert "OP_SERVICE_ACCOUNT_TOKEN is not set" in str(exc_info.value)

Expand All @@ -121,7 +114,7 @@ def test_get_secret_error(mock_run):
model = TestModel(op_uri="op://vault/item/field")

with pytest.raises(ValueError) as exc_info:
_ = model.op_secret
_ = model.op_uri_secret

assert "Could not read secret from 1Password" in str(exc_info.value)

Expand All @@ -132,7 +125,7 @@ def test_get_secret_command_not_available(mock_which, db):
model = TestModel(op_uri="op://vault/item/field")

with pytest.raises(OSError) as excinfo:
_ = model.op_secret
_ = model.op_uri_secret

assert "The 'op' CLI command is not available" in str(excinfo.value)

Expand All @@ -143,10 +136,10 @@ def test_set_secret_failure(mock_run):
model = TestModel(op_uri="op://vault/item/field")

with pytest.raises(NotImplementedError) as exc_info:
model.op_secret = "new secret"
model.op_uri_secret = "new secret"
model.save()

assert "OPField does not support setting secret values" in str(exc_info.value)
assert "OPField does not support setting secret value" in str(exc_info.value)


@pytest.mark.parametrize(
Expand Down

0 comments on commit e4a01c9

Please sign in to comment.