diff options
-rw-r--r-- | src/onapsdk/cps/dataspace.py | 49 | ||||
-rw-r--r-- | tests/test_cps.py | 67 | ||||
-rw-r--r-- | tests/test_onap_service.py | 12 |
3 files changed, 117 insertions, 11 deletions
diff --git a/src/onapsdk/cps/dataspace.py b/src/onapsdk/cps/dataspace.py index d997f81..282880f 100644 --- a/src/onapsdk/cps/dataspace.py +++ b/src/onapsdk/cps/dataspace.py @@ -13,7 +13,9 @@ # See the License for the specific language governing permissions and # limitations under the License. +from functools import wraps from typing import Any, Dict, Iterable +from onapsdk.exceptions import (APIError, ResourceNotFound) from .anchor import Anchor from .cps_element import CpsElement @@ -52,6 +54,22 @@ class Dataspace(CpsElement): """ return f"{self._url}/dataspaces/{self.name}" + def exception_handler(function): # pylint: disable= no-self-argument + """Exception handler. + + Handling APIError and throwing ResourceNotFound if Data space does not exist. + + """ + @wraps(function) + def wrapper(*args): + try: + return function(*args) # pylint: disable= not-callable + except APIError as error: + if (error.response_status_code == 400 and 'Dataspace not found' in str(error)): + raise ResourceNotFound(error) from error + raise + return wrapper + @classmethod def create(cls, dataspace_name: str) -> "Dataspace": """Create dataspace with given name. @@ -71,6 +89,7 @@ class Dataspace(CpsElement): ) return Dataspace(dataspace_name) + @exception_handler def create_anchor(self, schema_set: SchemaSet, anchor_name: str) -> Anchor: """Create anchor. @@ -90,6 +109,7 @@ class Dataspace(CpsElement): ) return Anchor(name=anchor_name, schema_set=schema_set) + @exception_handler def get_anchors(self) -> Iterable[Anchor]: """Get all dataspace's anchors. @@ -99,16 +119,23 @@ class Dataspace(CpsElement): Iterator[Anchor]: Anchor object """ - for anchor_data in self.send_message_json(\ - "GET",\ - "Get all CPS dataspace anchors",\ - f"{self.url}/anchors",\ - auth=self.auth\ - ): - yield Anchor(name=anchor_data["name"], - schema_set=SchemaSet(name=anchor_data["schemaSetName"], - dataspace=self)) - + try: + for anchor_data in self.send_message_json(\ + "GET",\ + "Get all CPS dataspace anchors",\ + f"{self.url}/anchors",\ + auth=self.auth\ + ): + yield Anchor(name=anchor_data["name"], + schema_set=SchemaSet(name=anchor_data["schemaSetName"], + dataspace=self)) + except APIError as error: + if (error.response_status_code == 400 and 'Dataspace not found' in str(error)): + raise ResourceNotFound(error) from error + raise + + + @exception_handler def get_anchor(self, anchor_name: str) -> Anchor: """Get dataspace anchor by name. @@ -131,6 +158,7 @@ class Dataspace(CpsElement): schema_set=SchemaSet(name=anchor_data["schemaSetName"], dataspace=self)) + @exception_handler def get_schema_set(self, schema_set_name: str) -> SchemaSet: """Get schema set by name. @@ -159,6 +187,7 @@ class Dataspace(CpsElement): ] ) + @exception_handler def create_schema_set(self, schema_set_name: str, schema_set: bytes) -> SchemaSet: """Create schema set. diff --git a/tests/test_cps.py b/tests/test_cps.py index 8f7bdc4..1c4a83a 100644 --- a/tests/test_cps.py +++ b/tests/test_cps.py @@ -12,10 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. +import requests from unittest import mock from typing import List -from onapsdk.cps import Anchor, Dataspace, SchemaSet, SchemaSetModuleReference, anchor +import pytest +from onapsdk.cps import Anchor, Dataspace, SchemaSet, SchemaSetModuleReference +from onapsdk.exceptions import (APIError, ResourceNotFound, SDKException) DATASPACE_ANCHOR = { "name": "anchor1", @@ -238,3 +241,65 @@ def test_anchor_delete_nodes(mock_send_message): mock_send_message.assert_called_once() url = mock_send_message.call_args[0][2] assert "xpath=test-xpath" in url + +@mock.patch("onapsdk.cps.Dataspace.send_message") +def test_dataspace_create_anchor_except(mock_send_message_json): + ds = Dataspace(name="test_creating_anchor") + mock_send_message_json.exceptions = requests.exceptions + mock_send_message_json.side_effect = APIError('Dataspace not found', 400) + with pytest.raises(ResourceNotFound): + ds.create_anchor(mock.MagicMock(), "test_creating_anchor") + mock_send_message_json.side_effect = APIError() + with pytest.raises(SDKException): + ds.create_anchor(mock.MagicMock(), "test_creating_anchor") + +@mock.patch("onapsdk.cps.Dataspace.send_message_json") +def test_dataspace_get_anchors_except(mock_send_message_json): + ds = Dataspace(name="test_ds") + mock_send_message_json.exceptions = requests.exceptions + mock_send_message_json.side_effect = APIError('Dataspace not found', 400) + with pytest.raises(ResourceNotFound): + list(ds.get_anchors()) + mock_send_message_json.side_effect = APIError() + with pytest.raises(SDKException): + list(ds.get_anchors()) + +@mock.patch("onapsdk.cps.Dataspace.send_message_json") +def test_dataspace_get_anchor_except(mock_send_message_json): + ds = Dataspace(name="test_ds") + mock_send_message_json.exceptions = requests.exceptions + mock_send_message_json.side_effect = APIError('Dataspace not found', 400) + with pytest.raises(ResourceNotFound): + ds.get_anchor(mock.MagicMock()) + mock_send_message_json.side_effect = APIError() + with pytest.raises(SDKException): + ds.get_anchor(mock.MagicMock()) + +@mock.patch("onapsdk.cps.Dataspace.send_message_json") +def test_dataspace_get_schema_set_except(mock_send_message_json): + mock_send_message_json.return_value = DATASPACE_SCHEMA_SET + ds = Dataspace(name="test_ds") + mock_send_message_json.exceptions = requests.exceptions + mock_send_message_json.side_effect = APIError('Dataspace not found', 400) + with pytest.raises(ResourceNotFound): + ds.get_schema_set(mock.MagicMock()) + mock_send_message_json.side_effect = APIError() + with pytest.raises(SDKException): + ds.get_schema_set(mock.MagicMock()) + + +@mock.patch("onapsdk.cps.Dataspace.send_message") +@mock.patch("onapsdk.cps.Dataspace.get_schema_set") +def test_dataspace_create_schema_set_except(mock_get_chema_set, mock_send_message): + ds = Dataspace(name="test_ds") + _ = ds.create_schema_set("test_schema_set_name", b"fake_file") + mock_send_message.exceptions = requests.exceptions + mock_send_message.side_effect = APIError('Dataspace not found', 400) + mock_get_chema_set.exceptions = requests.exceptions + mock_get_chema_set.side_effect = APIError('Dataspace not found', 400) + with pytest.raises(ResourceNotFound): + ds.create_schema_set(mock.MagicMock(), 8) + mock_send_message.side_effect = APIError() + mock_get_chema_set.side_effect = APIError() + with pytest.raises(SDKException): + ds.create_schema_set(mock.MagicMock(), 9) diff --git a/tests/test_onap_service.py b/tests/test_onap_service.py index 62a9880..685f906 100644 --- a/tests/test_onap_service.py +++ b/tests/test_onap_service.py @@ -378,3 +378,15 @@ def test_set_header(mock_session): assert headers["test-header-dict-key"] == "test-header-dict-value" assert "test-header-common-key" in headers assert headers["test-header-common-key"] == "test-header-common-value" + + mock_session.reset_mock() + cert = mock.MagicMock(name="test-cert") + OnapService.send_message("GET", 'test get', 'http://my.url/', headers={"test-header-common-key": "test-header-common-value"}, cert=cert) + _, _, kwargs = mock_session.return_value.request.mock_calls[0] + headers = kwargs["headers"] + assert "test-header-callable-key" in headers + assert headers["test-header-callable-key"] == "test-header-callable-value" + assert "test-header-dict-key" in headers + assert headers["test-header-dict-key"] == "test-header-dict-value" + assert "test-header-common-key" in headers + assert headers["test-header-common-key"] == "test-header-common-value" |