Skip to content
Closed
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
2 changes: 2 additions & 0 deletions USAGE.md
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,8 @@ Users can create `classes` or `dicts` that describe fields they wish to get the
- **load_dict**: Takes a dictionary with keys specifying the user desired naming scheme of the values to return. Each key's value is a dictionary that includes information on where to find the item field value in 1Password. This returns a dictionary of user specified keys with values retrieved from 1Password.
- **load**: Takes an object with class attributes annotated with tags describing where to find desired fields in 1Password. Manipulates given object and fills attributes in with 1Password item field values.

The `opfield` tag uses `section.field` syntax. The section can be a section label or ID, and the field can be a field label or ID. Labels are tried first; IDs are used as a fallback when no matching label is found.

```python
# example dict configuration for onepasswordconnectsdk.load_dict(connect_client, CONFIG)
CONFIG = {
Expand Down
51 changes: 30 additions & 21 deletions src/onepasswordconnectsdk/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,32 +230,41 @@ def _set_values_for_item(
{parsed_field.name}"
)

value_found = False
for field in item.fields:
try:
section_id = field.section.id
except AttributeError:
section_id = None

if field.label == path_parts[1]:
if (
section_id is None
or (section_id == sections.get(path_parts[0]))
or path_parts[0] in sections.values()
):
value_found = True

if config_object:
setattr(config_object, parsed_field.name, field.value)
else:
config_dict[parsed_field.name] = field.value
break
if not value_found:
matched_field = _find_field(item.fields, sections, path_parts[0], path_parts[1])
if matched_field is None:
raise UnknownSectionAndFieldTag(
f"There is no section {path_parts[0]} \
for field {path_parts[1]}"
)

if config_object:
setattr(config_object, parsed_field.name, matched_field.value)
else:
config_dict[parsed_field.name] = matched_field.value


def _find_field(fields, sections: Dict[str, str], section_tag: str, field_tag: str):
for attr in ("label", "id"):
for field in fields:
if getattr(field, attr, None) == field_tag and _field_section_matches(
field, sections, section_tag
):
return field
return None


def _field_section_matches(field, sections: Dict[str, str], section_tag: str):
try:
section_id = field.section.id
except AttributeError:
section_id = None

return (
section_id is None
or section_id == sections.get(section_tag)
or section_tag in sections.values()
)


def _convert_sections_to_dict(sections: List[Section]):
if not sections:
Expand Down
71 changes: 71 additions & 0 deletions src/tests/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,56 @@ def test_load_dict_empty_field_returns_none(respx_mock):
assert config_with_values['empty'] is None


def test_load_dict_resolves_field_by_id_when_label_differs(respx_mock):
config_dict = {
"host": {
"opitem": ITEM_NAME2,
"opfield": ".716C5B0E95A84092B2FE2CC402E0DDDF",
"opvault": VAULT_ID,
}
}

respx_mock.get(f"v1/vaults/{VAULT_ID}/items?filter=title eq \"{ITEM_NAME2}\"").mock(
return_value=Response(200, json=[item2]))
respx_mock.get(f"v1/vaults/{VAULT_ID}/items/{ITEM_ID2}").mock(return_value=Response(200, json=item2))

config_with_values = onepasswordconnectsdk.load_dict(SS_CLIENT, config_dict)

assert config_with_values["host"] == HOST_VALUE


def test_load_resolves_field_by_id_when_label_differs(respx_mock):
class ConfigByFieldID:
host: f'opitem:"{ITEM_NAME2}" opfield:.716C5B0E95A84092B2FE2CC402E0DDDF opvault:{VAULT_ID}' = None

respx_mock.get(f"v1/vaults/{VAULT_ID}/items?filter=title eq \"{ITEM_NAME2}\"").mock(
return_value=Response(200, json=[item2]))
respx_mock.get(f"v1/vaults/{VAULT_ID}/items/{ITEM_ID2}").mock(return_value=Response(200, json=item2))

config_with_values = onepasswordconnectsdk.load(SS_CLIENT, ConfigByFieldID())

assert config_with_values.host == HOST_VALUE


def test_load_dict_prefers_field_label_when_label_and_id_both_match(respx_mock):
config_dict = {
"selected": {
"opitem": ITEM_NAME1,
"opfield": ".username",
"opvault": VAULT_ID,
}
}

respx_mock.get(f"v1/vaults/{VAULT_ID}/items?filter=title eq \"{ITEM_NAME1}\"").mock(
return_value=Response(200, json=[item_with_label_id_collision]))
respx_mock.get(f"v1/vaults/{VAULT_ID}/items/{ITEM_ID1}").mock(
return_value=Response(200, json=item_with_label_id_collision))

config_with_values = onepasswordconnectsdk.load_dict(SS_CLIENT, config_dict)

assert config_with_values["selected"] == USERNAME_VALUE


item = {
"id": ITEM_ID1,
"title": ITEM_NAME1,
Expand Down Expand Up @@ -148,6 +198,27 @@ def test_load_dict_empty_field_returns_none(respx_mock):
]
}

item_with_label_id_collision = {
"id": ITEM_ID1,
"title": ITEM_NAME1,
"vault": {
"id": VAULT_ID
},
"category": "LOGIN",
"fields": [
{
"id": "primary_username_id",
"label": "username",
"value": USERNAME_VALUE
},
{
"id": "username",
"label": "api_username",
"value": "id fallback value"
}
]
}

item2 = {
"id": ITEM_ID2,
"title": ITEM_NAME2,
Expand Down
Loading