-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
261 additions
and
65 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,104 @@ | ||
from typing import Callable, Optional, Union, overload | ||
|
||
from .resources import Resource, ResourceParameters, Resources | ||
|
||
AfterDestroyCallback = Callable[[], None] | ||
|
||
|
||
class Vanity(Resource): | ||
"""Represents a Vanity resource with the ability to destroy itself.""" | ||
|
||
def __init__( | ||
self, | ||
/, | ||
params: ResourceParameters, | ||
*, | ||
after_destroy: AfterDestroyCallback = lambda: None, | ||
**kwargs, | ||
): | ||
super().__init__(params, **kwargs) | ||
self._after_destroy = after_destroy | ||
|
||
def destroy(self) -> None: | ||
"""Destroy the vanity resource.""" | ||
content_guid = self.get("content_guid") | ||
if content_guid is None: | ||
raise ValueError( | ||
"The 'content_guid' is missing. Unable to perform the destroy operation." | ||
) | ||
endpoint = self.params.url + f"v1/content/{content_guid}/vanity" | ||
self.params.session.delete(endpoint) | ||
self._after_destroy() | ||
|
||
|
||
class Vanities(Resources): | ||
"""Manages a collection of Vanity resources.""" | ||
|
||
def all(self) -> list[Vanity]: | ||
"""Retrieve all vanity resources.""" | ||
endpoint = self.params.url + "v1/vanities" | ||
response = self.params.session.get(endpoint) | ||
results = response.json() | ||
return [Vanity(self.params, **result) for result in results] | ||
|
||
|
||
class VanityMixin(Resource): | ||
"""Mixin class to add vanity management capabilities to a resource.""" | ||
|
||
def __init__(self, /, params: ResourceParameters, **kwargs): | ||
super().__init__(params, **kwargs) | ||
self._vanity: Optional[Vanity] = None | ||
|
||
@property | ||
def vanity(self) -> Vanity: | ||
"""Retrieve or lazily load the associated vanity resource.""" | ||
if self._vanity is None: | ||
uid = self.get("guid") | ||
if uid is None: | ||
raise ValueError( | ||
"The 'guid' is missing. Unable to perform the get vanity operation." | ||
) | ||
endpoint = self.params.url + f"v1/content/{uid}/vanity" | ||
response = self.params.session.get(endpoint) | ||
result = response.json() | ||
self._vanity = Vanity( | ||
self.params, after_destroy=lambda: setattr(self, "_vanity", None), **result | ||
) | ||
return self._vanity | ||
|
||
@vanity.setter | ||
def vanity(self, value: Union[str, dict]) -> None: | ||
"""Set the vanity using a path or dictionary of attributes.""" | ||
if isinstance(value, str): | ||
self.set_vanity(path=value) | ||
elif isinstance(value, dict): | ||
self.set_vanity(**value) | ||
self.reset() | ||
|
||
@vanity.deleter | ||
def vanity(self) -> None: | ||
"""Delete the vanity resource.""" | ||
if self._vanity: | ||
self._vanity.destroy() | ||
self.reset() | ||
|
||
def reset(self) -> None: | ||
"""Reset the cached vanity resource.""" | ||
self._vanity = None | ||
|
||
@overload | ||
def set_vanity(self, *, path: str) -> None: ... | ||
|
||
@overload | ||
def set_vanity(self, *, path: str, force: bool) -> None: ... | ||
|
||
@overload | ||
def set_vanity(self, **attributes) -> None: ... | ||
|
||
def set_vanity(self, **attributes) -> None: | ||
"""Set or update the vanity resource with given attributes.""" | ||
uid = self.get("guid") | ||
if uid is None: | ||
raise ValueError("The 'guid' is missing. Unable to perform the set vanity operation.") | ||
endpoint = self.params.url + f"v1/content/{uid}/vanity" | ||
self.params.session.put(endpoint, json=attributes) |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,153 @@ | ||
from unittest.mock import Mock | ||
|
||
import pytest | ||
import requests | ||
import responses | ||
from responses.matchers import json_params_matcher | ||
|
||
from posit.connect.resources import ResourceParameters | ||
from posit.connect.urls import Url | ||
from posit.connect.vanities import Vanities, Vanity, VanityMixin | ||
|
||
|
||
class TestVanityDestroy: | ||
@responses.activate | ||
def test_destroy_sends_delete_request(self): | ||
content_guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/content/{content_guid}/vanity" | ||
mock_delete = responses.delete(endpoint) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
params = ResourceParameters(session, url) | ||
vanity = Vanity(params, content_guid=content_guid) | ||
|
||
vanity.destroy() | ||
|
||
assert mock_delete.call_count == 1 | ||
|
||
def test_destroy_without_content_guid_raises_value_error(self): | ||
vanity = Vanity(params=Mock()) | ||
with pytest.raises(ValueError): | ||
vanity.destroy() | ||
|
||
def test_destroy_with_none_content_guid_raises_value_error(self): | ||
vanity = Vanity(params=Mock(), content_guid=None) | ||
with pytest.raises(ValueError): | ||
vanity.destroy() | ||
|
||
@responses.activate | ||
def test_destroy_calls_after_destroy_callback(self): | ||
content_guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/content/{content_guid}/vanity" | ||
responses.delete(endpoint) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
after_destroy = Mock() | ||
params = ResourceParameters(session, url) | ||
vanity = Vanity(params, after_destroy=after_destroy, content_guid=content_guid) | ||
|
||
vanity.destroy() | ||
|
||
assert after_destroy.call_count == 1 | ||
|
||
|
||
class TestVanitiesAll: | ||
@responses.activate | ||
def test_all_sends_get_request(self): | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/vanities" | ||
mock_get = responses.get(endpoint, json=[]) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
params = ResourceParameters(session, url) | ||
vanities = Vanities(params) | ||
|
||
vanities.all() | ||
|
||
assert mock_get.call_count == 1 | ||
|
||
|
||
class TestVanityMixin: | ||
@responses.activate | ||
def test_vanity_getter_returns_vanity(self): | ||
guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/content/{guid}/vanity" | ||
vanity_data = {"content_guid": guid} | ||
mock_get = responses.get(endpoint, json=vanity_data) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
params = ResourceParameters(session, url) | ||
content = VanityMixin(params, guid=guid) | ||
|
||
assert content.vanity == vanity_data | ||
assert mock_get.call_count == 1 | ||
|
||
@responses.activate | ||
def test_vanity_setter_with_string(self): | ||
guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/content/{guid}/vanity" | ||
path = "example" | ||
mock_put = responses.put(endpoint, match=[json_params_matcher({"path": path})]) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
params = ResourceParameters(session, url) | ||
content = VanityMixin(params, guid=guid) | ||
content.vanity = path | ||
|
||
assert mock_put.call_count == 1 | ||
|
||
@responses.activate | ||
def test_vanity_setter_with_dict(self): | ||
guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/content/{guid}/vanity" | ||
vanity_attrs = {"path": "example", "locked": True} | ||
mock_put = responses.put(endpoint, match=[json_params_matcher(vanity_attrs)]) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
params = ResourceParameters(session, url) | ||
content = VanityMixin(params, guid=guid) | ||
content.vanity = vanity_attrs | ||
|
||
assert mock_put.call_count == 1 | ||
|
||
@responses.activate | ||
def test_vanity_deleter_sends_delete_request(self): | ||
guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/content/{guid}/vanity" | ||
mock_delete = responses.delete(endpoint) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
params = ResourceParameters(session, url) | ||
content = VanityMixin(params, guid=guid) | ||
content._vanity = Vanity(params, content_guid=guid) | ||
del content.vanity | ||
|
||
assert mock_delete.call_count == 1 | ||
|
||
@responses.activate | ||
def test_set_vanity(self): | ||
guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" | ||
base_url = "http://connect.example/__api__" | ||
endpoint = f"{base_url}/v1/content/{guid}/vanity" | ||
mock_put = responses.put(endpoint) | ||
|
||
session = requests.Session() | ||
url = Url(base_url) | ||
params = ResourceParameters(session, url) | ||
content = VanityMixin(params, guid=guid) | ||
content.set_vanity(path="example") | ||
|
||
assert mock_put.call_count == 1 |