From 401b2c7bde08aff49f57c97603a8ddf765421471 Mon Sep 17 00:00:00 2001 From: Vladimir Prusakov Date: Wed, 24 Jan 2024 19:50:06 +0200 Subject: [PATCH] 2.0.1 (2024-01-24) ------------------ **Fixed:** FortigateAPI `get`, `update` methods for objects without UID --- CHANGELOG.rst | 5 + docs/conf.py | 2 +- docs/fortigateapi/cmdb/system/global_.rst | 39 ++++ examples/cmdb/system/global_.py | 31 +++ fortigate_api/cmdb/system/global_.py | 1 + fortigate_api/connector.py | 53 +++-- fortigate_api/schema/custom/settings.py | 3 + pyproject.toml | 4 +- tests/helpers__tst.py | 139 ++++++----- tests/test__connector.py | 220 ++++++++++++++++++ ...l_policy.py => test__connector__policy.py} | 0 tests/test__connector_create.py | 29 --- tests/test__connector_delete.py | 52 ----- tests/test__connector_get.py | 43 ---- tests/test__connector_is_exist.py | 35 --- tests/test__connector_update.py | 37 --- 16 files changed, 421 insertions(+), 272 deletions(-) create mode 100644 examples/cmdb/system/global_.py create mode 100644 tests/test__connector.py rename tests/{test__connector__cmdb_firewall_policy.py => test__connector__policy.py} (100%) delete mode 100644 tests/test__connector_create.py delete mode 100644 tests/test__connector_delete.py delete mode 100644 tests/test__connector_get.py delete mode 100644 tests/test__connector_is_exist.py delete mode 100644 tests/test__connector_update.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 3a04e4da..f5656ad6 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -12,6 +12,11 @@ Unreleased **New:** `FortiGateAPI.monitor` connectors, to work with all `Monitor API` endpoints. +2.0.1 (2024-01-24) +------------------ +**Fixed:** FortigateAPI `get`, `update` methods for objects without UID + + 2.0.0 (2024-01-16) ------------------ diff --git a/docs/conf.py b/docs/conf.py index e4dcc45d..f6fa9ff6 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -8,7 +8,7 @@ project = "fortigate-api" copyright = "2021, Vladimirs Prusakovs" author = "Vladimirs Prusakovs" -release = "2.0.0" +release = "2.0.1" extensions = [ "sphinx.ext.autodoc", diff --git a/docs/fortigateapi/cmdb/system/global_.rst b/docs/fortigateapi/cmdb/system/global_.rst index e8418fb6..68adab89 100644 --- a/docs/fortigateapi/cmdb/system/global_.rst +++ b/docs/fortigateapi/cmdb/system/global_.rst @@ -7,3 +7,42 @@ FortiGateAPI.cmdb.system.global_ :inherited-members: :class-doc-from: class + +Usage +----- + +.. code:: python + + """api/v2/cmdb/system/global + + - Update data in the Fortigate + - Get data from the Fortigate + """ + + from pprint import pprint + + from fortigate_api import FortiGateAPI + + HOST = "host" + USERNAME = "username" + PASSWORD = "password" + + api = FortiGateAPI(host=HOST, username=USERNAME, password=PASSWORD) + + # Update data in the Fortigate + data = {"timezone": 80} + response = api.cmdb.system.global_.update(data) + print(f"update {response}") # update + + # Get data from the Fortigate + result = api.cmdb.system.global_.get() + pprint(result) + # [{"admin-concurrent": "enable", + # "admin-console-timeout": 300, + # "admin-hsts-max-age": 15552000, + # "timezone": "80", + # ... + + api.logout() + + diff --git a/examples/cmdb/system/global_.py b/examples/cmdb/system/global_.py new file mode 100644 index 00000000..d36fa9bf --- /dev/null +++ b/examples/cmdb/system/global_.py @@ -0,0 +1,31 @@ +"""api/v2/cmdb/system/global + +- Update data in the Fortigate +- Get data from the Fortigate +""" + +from pprint import pprint + +from fortigate_api import FortiGateAPI + +HOST = "host" +USERNAME = "username" +PASSWORD = "password" + +api = FortiGateAPI(host=HOST, username=USERNAME, password=PASSWORD) + +# Update data in the Fortigate +data = {"timezone": 80} +response = api.cmdb.system.global_.update(data) +print(f"update {response}") # update + +# Get data from the Fortigate +result = api.cmdb.system.global_.get() +pprint(result) +# [{"admin-concurrent": "enable", +# "admin-console-timeout": 300, +# "admin-hsts-max-age": 15552000, +# "timezone": "80", +# ... + +api.logout() diff --git a/fortigate_api/cmdb/system/global_.py b/fortigate_api/cmdb/system/global_.py index 9013ad6b..a75c0dfe 100644 --- a/fortigate_api/cmdb/system/global_.py +++ b/fortigate_api/cmdb/system/global_.py @@ -8,3 +8,4 @@ class GlobalSC(Connector): uid = "" _path = "api/v2/cmdb/system/global" + _path_ui = "/ng/system/settings" diff --git a/fortigate_api/connector.py b/fortigate_api/connector.py index 65766cb1..f78bbc7a 100644 --- a/fortigate_api/connector.py +++ b/fortigate_api/connector.py @@ -73,7 +73,8 @@ def delete( - `` Object not found in the Fortigate. :rtype: Response """ - _ = kwargs # noqa todo extract uid key from kwargs (if uid not specified) + if not uid: + uid = str(kwargs.get(self.uid) or "") h.check_uid_filter(uid, filter) if filter: return self._delete_by_filter(filter) @@ -101,25 +102,6 @@ def get(self, **kwargs) -> LDAny: items = [item] return items - def update(self, data: DAny) -> Response: - """Update fortigate-object on the Fortigate. - - :param dict data: Data of the fortigate-object to update. - More details can be found at https://fndn.fortinet.net for related ``PUT`` method. - - :return: Session response. - - - `` Object successfully updated, - - `` Object has not been updated. - :rtype: Response - """ - uid = str(data[self.uid]) - uid = h.quote(uid) - if not uid: - raise ValueError(f"{self.uid} value is required.") - url = f"{self.url}/{uid}" - return self.fortigate.put(url=url, data=data) - def is_exist(self, uid: StrInt) -> bool: """Check if a fortigate-object exists in the Fortigate. @@ -136,7 +118,21 @@ def is_exist(self, uid: StrInt) -> bool: response = self.fortigate.exist(url) return response.ok - # =========================== helpers =========================== + def update(self, data: DAny) -> Response: + """Update fortigate-object on the Fortigate. + + :param dict data: Data of the fortigate-object to update. + More details can be found at https://fndn.fortinet.net for related ``PUT`` method. + + :return: Session response. + + - `` Object successfully updated, + - `` Object has not been updated. + :rtype: Response + """ + uid: str = self._get_uid(data) + url = f"{self.url}/{uid}".rstrip("/") + return self.fortigate.put(url=url, data=data) # noinspection PyShadowingBuiltins def _delete_by_filter(self, filter: UStr) -> Response: # pylint: disable=redefined-builtin @@ -161,3 +157,18 @@ def _delete_by_filter(self, filter: UStr) -> Response: # pylint: disable=redefi response = self.fortigate.delete(url=url) responses.append(response) return h.highest_response(responses) + + def _get_uid(self, data) -> str: + """Get UID value based on the UID key + + :param data: A dictionary containing the UID value. + :return: The UID value extracted from the data. + :raises ValueError: If the UID is required but not present in the data. + """ + if not self.uid: + return "" + if self.uid not in data: + raise ValueError(f"{self.uid} value is required.") + uid = str(data[self.uid]) + uid = h.quote(uid) + return uid diff --git a/fortigate_api/schema/custom/settings.py b/fortigate_api/schema/custom/settings.py index 8610a0b0..c78b3af6 100644 --- a/fortigate_api/schema/custom/settings.py +++ b/fortigate_api/schema/custom/settings.py @@ -53,6 +53,9 @@ "external-resource": { "path_ui": "ng/external-connector", }, + "global": { + "path_ui": "/ng/system/settings", + }, "interface": { "is_custom_model": True, }, diff --git a/pyproject.toml b/pyproject.toml index 334db4bf..a5435013 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "fortigate_api" -version = "2.0.0" +version = "2.0.1" description = "Python package to configure Fortigate (Fortios) devices using REST API and SSH" authors = ["Vladimirs Prusakovs "] readme = "README.rst" @@ -55,7 +55,7 @@ test = ["pytest"] [tool.poetry.urls] "Bug Tracker" = "https://github.com/vladimirs-git/fortigate-api/issues" -"Download URL" = "https://github.com/vladimirs-git/fortigate-api/archive/refs/tags/2.0.0.tar.gz" +"Download URL" = "https://github.com/vladimirs-git/fortigate-api/archive/refs/tags/2.0.1.tar.gz" [tool.pylint] max-line-length = 100 diff --git a/tests/helpers__tst.py b/tests/helpers__tst.py index fa776da3..df80086f 100644 --- a/tests/helpers__tst.py +++ b/tests/helpers__tst.py @@ -55,6 +55,9 @@ "firewall/policy", "system.snmp/community", ] +WO_UID = [ + "system/global", +] @pytest.fixture @@ -87,12 +90,23 @@ def connectors_id(): api = FortiGateAPI(host="host") api.fortigate._session = Session() items = [ - # api.cmdb.firewall.internet_service, + api.cmdb.firewall.internet_service, api.cmdb.system_snmp.community, ] return items +@pytest.fixture +def connectors_wo_uid(): + """Init Connector without uid key.""" + api = FortiGateAPI(host="host") + api.fortigate._session = Session() + items = [ + api.cmdb.system.global_, + ] + return items + + def create_cookie(name: str, value: str) -> Cookie: """Return Cookie object.""" return Cookie( @@ -163,11 +177,11 @@ def session_delete(mocker: MockerFixture, url, *args, **kwargs) -> Response: _ = args, kwargs app_model = h.url_to_app_model(url) if app_model in UID_NAME: - resp = connector_delete(url) + resp = connector_delete__name(url) elif app_model == "firewall/policy": - resp = firewall_policy_delete(url) + resp = connector_delete__policy(url) elif app_model in UID_ID: - resp = firewall_policy_delete(url) + resp = connector_delete__policy(url) # invalid url else: resp = create_response("delete", url, 400) @@ -179,13 +193,15 @@ def session_get(mocker: MockerFixture, url, *args, **kwargs) -> Response: _ = args, kwargs app_model = h.url_to_app_model(url) if app_model in UID_NAME: - resp = connector_get_name(url) + resp = connector_get__name(url) elif app_model == "firewall/policy": - resp = firewall_policy_get(url) + resp = connector_get__policy(url) elif app_model == "firewall.ipmacbinding/setting": resp = firewall_ipmacbinding_setting_get(url) elif app_model in UID_ID: - resp = connector_get_id(url) + resp = connector_get__id(url) + elif app_model in WO_UID: + resp = connector_get__wo_uid(url) # invalid url else: resp = create_response("get", url, 400) @@ -197,11 +213,11 @@ def session_post(mocker: MockerFixture, url, *args, **kwargs) -> Response: _ = args, kwargs app_model = h.url_to_app_model(url) if app_model in UID_NAME: - resp = connector_post(url, data=kwargs) + resp = connector_post__name(url, data=kwargs) elif app_model == "firewall/policy": - resp = firewall_policy_post(url, data=kwargs) + resp = connector_post__policy(url, data=kwargs) elif app_model in UID_ID: - resp = connector_post(url, data=kwargs) + resp = connector_post__name(url, data=kwargs) # invalid url else: resp = create_response("post", url, 400) @@ -210,14 +226,16 @@ def session_post(mocker: MockerFixture, url, *args, **kwargs) -> Response: def session_put(mocker: MockerFixture, url, *args, **kwargs) -> Response: """Mock Session.put(), update""" - _ = args, kwargs + _ = args app_model = h.url_to_app_model(url) if app_model in UID_NAME: - resp = connector_put(url) + resp = connector_put__name(url) elif app_model == "firewall/policy": - resp = firewall_policy_put(url) + resp = connector_put__policy(url) elif app_model in UID_ID: - resp = firewall_policy_put(url) + resp = connector_put__policy(url) + elif app_model in WO_UID: + resp = connector_put__wo_uid(url, **kwargs) # invalid url else: resp = create_response("put", url, 400) @@ -226,7 +244,7 @@ def session_put(mocker: MockerFixture, url, *args, **kwargs) -> Response: # ============================ connector ============================= -def connector_delete(url: str) -> Response: +def connector_delete__name(url: str) -> Response: """Delete""" uid = h.url_to_uid(url) query = urlparse(url).query @@ -238,7 +256,16 @@ def connector_delete(url: str) -> Response: return create_response("delete", url, status_code) -def connector_get_name(url: str) -> Response: +def connector_delete__policy(url: str) -> Response: + """Delete""" + uid = h.url_to_uid(url) + status_code = { + "1": 200, # exist + }.get(uid, 404) + return create_response("delete", url, status_code) + + +def connector_get__name(url: str) -> Response: """Get""" uid = h.url_to_uid(url) query = urlparse(url).query @@ -255,7 +282,7 @@ def connector_get_name(url: str) -> Response: return create_response("get", url, status_code, data=items) -def connector_get_id(url: str) -> Response: +def connector_get__id(url: str) -> Response: """Get""" uid = h.url_to_uid(url) query = urlparse(url).query @@ -268,38 +295,7 @@ def connector_get_id(url: str) -> Response: return create_response("get", url, status_code, data=items) -def connector_post(url: str, data: dict) -> Response: - """Create""" - data = json.loads(data["data"]) - uid = data["name"] - status_code = { - "NAME1": 500, # exist - "A/B": 500, - }.get(uid, 200) - return create_response("post", url, status_code) - - -def connector_put(url: str) -> Response: - """Update""" - uid = h.url_to_uid(url) - status_code = { - "NAME1": 200, # exist - }.get(uid, 404) - return create_response("get", url, status_code) - - -# ======================= cmdb/firewall/policy ======================= - -def firewall_policy_delete(url: str) -> Response: - """Delete""" - uid = h.url_to_uid(url) - status_code = { - "1": 200, # exist - }.get(uid, 404) - return create_response("delete", url, status_code) - - -def firewall_policy_get(url: str) -> Response: +def connector_get__policy(url: str) -> Response: """Get""" uid = h.url_to_uid(url) query = urlparse(url).query @@ -316,7 +312,30 @@ def firewall_policy_get(url: str) -> Response: return create_response("get", url, status_code, data=items) -def firewall_policy_post(url: str, data: dict) -> Response: +def connector_get__wo_uid(url: str) -> Response: + """Get""" + uid = h.url_to_uid(url) + query = urlparse(url).query + key = (uid, query) + items = { + ("", "key=value"): {"key": "value"}, + }.get(key) + status_code = 404 if items is None else 200 + return create_response("get", url, status_code, data=items) + + +def connector_post__name(url: str, data: dict) -> Response: + """Create""" + data = json.loads(data["data"]) + uid = data["name"] + status_code = { + "NAME1": 500, # exist + "A/B": 500, + }.get(uid, 200) + return create_response("post", url, status_code) + + +def connector_post__policy(url: str, data: dict) -> Response: """Create""" data = json.loads(data["data"]) name = data["name"] @@ -326,7 +345,16 @@ def firewall_policy_post(url: str, data: dict) -> Response: return create_response("post", url, status_code) -def firewall_policy_put(url: str) -> Response: +def connector_put__name(url: str) -> Response: + """Update""" + uid = h.url_to_uid(url) + status_code = { + "NAME1": 200, # exist + }.get(uid, 404) + return create_response("get", url, status_code) + + +def connector_put__policy(url: str) -> Response: """Update""" uid = h.url_to_uid(url) status_code = { @@ -336,7 +364,14 @@ def firewall_policy_put(url: str) -> Response: return create_response("put", url, status_code) -# ======================= cmdb/firewall/policy ======================= +def connector_put__wo_uid(url: str, **kwargs) -> Response: + """Update""" + data = kwargs["data"] + status_code = 404 + if "key" in data: + status_code = 200 + return create_response("put", url, status_code, data=data) + def firewall_ipmacbinding_setting_get(url: str) -> Response: """Get""" diff --git a/tests/test__connector.py b/tests/test__connector.py new file mode 100644 index 00000000..fa5bf1c0 --- /dev/null +++ b/tests/test__connector.py @@ -0,0 +1,220 @@ +"""Test connector create.""" +from typing import Any + +import pytest +from pytest_mock import MockerFixture + +from tests import helpers__tst as tst +from tests.helpers__tst import connectors_name, connectors_id, connectors_wo_uid + + +@pytest.mark.parametrize("data, expected", [ + ({"name": "NAME1"}, 500), # object exists + ({"name": "A/B"}, 500), # object exists + ({"name": "NAME9"}, 200), # object does not exist + ({"typo": "NAME1"}, KeyError), # invalid uid +]) +def test__create(connectors_name: list, connectors_id: list, mocker: MockerFixture, data, expected): + """Connector create() uid is str.""" + mocker.patch(target="requests.Session.post", + side_effect=lambda *args, **kw: tst.session_post(mocker, *args, **kw)) + + connectors = [*connectors_name, *connectors_id] + for connector in connectors: + if isinstance(expected, int): + response = connector.create(data) + actual = response.status_code + assert actual == expected + else: + with pytest.raises(expected): + connector.create(data) + + +@pytest.mark.parametrize("kwargs, expected", [ + ({"uid": "NAME1"}, 200), # object exists + ({"uid": "NAME9"}, 404), # object does not exist + ({"filter": "name==NAME1"}, 200), # object has been found + ({"filter": "name==NAME9"}, 404), # object has not been found + ({"uid": ""}, ValueError), # invalid uid + ({"uid": "NAME1", "filter": "name==NAME1"}, ValueError), # invalid uid +]) +def test__delete__name(connectors_name: list, mocker: MockerFixture, kwargs, expected): + """Connector.delete() uid is str.""" + mocker.patch(target="requests.Session.get", + side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) + mocker.patch(target="requests.Session.delete", + side_effect=lambda *args, **kw: tst.session_delete(mocker, *args, **kw)) + + for connector in connectors_name: + if isinstance(expected, int): + response = connector.delete(**kwargs) + actual = response.status_code + assert actual == expected + + +@pytest.mark.parametrize("kwargs, expected", [ + ({"uid": 1}, 200), # object exists + ({"uid": "1"}, 200), # object exists + ({"uid": 9}, 404), # object does not exist + ({"uid": 0}, ValueError), # invalid uid +]) +def test__delete__id(connectors_id: list, mocker: MockerFixture, kwargs, expected): + """Connector.delete() uid is int.""" + mocker.patch(target="requests.Session.get", + side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) + mocker.patch(target="requests.Session.delete", + side_effect=lambda *args, **kw: tst.session_delete(mocker, *args, **kw)) + + for connector in connectors_id: + if isinstance(expected, int): + response = connector.delete(**kwargs) + actual = response.status_code + assert actual == expected + else: + with pytest.raises(expected): + connector.delete(**kwargs) + + +@pytest.mark.parametrize("kwargs, expected", [ + (dict(name="NAME1"), ["NAME1"]), # object exists + (dict(name="A/B"), ["A/B"]), # object exists + (dict(name="NAME9"), []), # object does not exist + (dict(name="NAME1", filter="name==NAME1"), ["NAME1"]), # object has been found + (dict(filter="name==NAME1"), ["NAME1"]), # object has been found + (dict(filter="name==NAME9"), []), # object has not been found + (dict(typo="NAME1"), []), # invalid param +]) +def test__get__name(connectors_name: list, mocker: MockerFixture, kwargs, expected): + """Connector.get() uid is str.""" + mocker.patch(target="requests.Session.get", + side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) + + for connector in connectors_name: + items = connector.get(**kwargs) + actual = [d["name"] for d in items] + assert actual == expected + + +@pytest.mark.parametrize("kwargs, expected", [ + (dict(id=1), ["NAME1"]), # object exists + (dict(name="NAME9"), []), # object does not exist + (dict(typo="NAME1"), []), # invalid param +]) +def test__get__id(connectors_id: list, mocker: MockerFixture, kwargs, expected): + """Connector.get() uid is int.""" + mocker.patch(target="requests.Session.get", + side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) + + for connector in connectors_id: + items = connector.get(**kwargs) + actual = [d["name"] for d in items] + assert actual == expected + + +@pytest.mark.parametrize("kwargs, expected", [ + (dict(key="value"), ["value"]), # object exists + (dict(typo="value"), [None]), # object does not exist +]) +def test__get__wo_uid(connectors_wo_uid: list, mocker: MockerFixture, kwargs, expected): + """Connector.get() without uid.""" + mocker.patch(target="requests.Session.get", + side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) + + for connector in connectors_wo_uid: + items = connector.get(**kwargs) + actual = [d.get("key") for d in items] + assert actual == expected + + +@pytest.mark.parametrize("uid, expected", [ + ("NAME1", True), # object exists + ("NAME9", False), # object does not exist +]) +def test__is_exist__name(connectors_name: list, mocker: MockerFixture, uid, expected): + """Connector.is_exist() uid is str.""" + mocker.patch(target="requests.Session.get", + side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) + + for connector in connectors_name: + actual = connector.is_exist(uid=uid) + assert actual == expected, connector + + +@pytest.mark.parametrize("uid, expected", [ + (1, True), # object exists + (9, False), # object does not exist +]) +def test__is_exist__id(connectors_id: list, mocker: MockerFixture, uid, expected): + """Connector.is_exist() uid is int.""" + mocker.patch(target="requests.Session.get", + side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) + + for connector in connectors_id: + actual = connector.is_exist(uid=uid) + assert actual == expected, connector + + +@pytest.mark.parametrize("data, expected", [ + (dict(name="NAME1"), 200), # object exists + (dict(name="NAME9"), 404), # object does not exist +]) +def test__update__name(connectors_name: list, mocker: MockerFixture, data, expected): + """Connector.update() uid is str.""" + mocker.patch(target="requests.Session.put", + side_effect=lambda *args, **kw: tst.session_put(mocker, *args, **kw)) + + for connector in connectors_name: + response = connector.update(data=data) + actual = response.status_code + assert actual == expected + + +@pytest.mark.parametrize("data, expected", [ + ({"id": 1}, 200), # object exists + ({"id": 9}, 404), # object does not exist +]) +def test__update__id(connectors_id: list, mocker: MockerFixture, data, expected): + """Connector.update() uid is int.""" + mocker.patch(target="requests.Session.put", + side_effect=lambda *args, **kw: tst.session_put(mocker, *args, **kw)) + + for connector in connectors_id: + response = connector.update(data=data) + actual = response.status_code + assert actual == expected + + +@pytest.mark.parametrize("data, expected", [ + ({"key": "value"}, 200), # object exists + ({"typo": "value"}, 404), # object does not exist +]) +def test__update__wo_uid(connectors_wo_uid: list, mocker: MockerFixture, data, expected): + """Connector.update() without uid.""" + mocker.patch(target="requests.Session.put", + side_effect=lambda *args, **kw: tst.session_put(mocker, *args, **kw)) + + for connector in connectors_wo_uid: + response = connector.update(data=data) + actual = response.status_code + assert actual == expected + + +@pytest.mark.parametrize("uid, data, expected", [ + ("", {}, ""), + ("", {"name": "NAME"}, ""), + ("policyid", {"policyid": 0}, "0"), + ("policyid", {"policyid": 1}, "1"), + ("name", {"name": "NAME"}, "NAME"), + ("name", {"typo": "NAME"}, ValueError), + ("name", {"name": "A/B"}, "A%2FB"), +]) +def test__get_uid(connectors_name: list, uid, data, expected: Any): + """Connector._get_uid()""" + connector = connectors_name[0] + connector.uid = uid + if isinstance(expected, str): + actual = connector._get_uid(data=data) + assert actual == expected + else: + with pytest.raises(expected): + connector._get_uid(data=data) diff --git a/tests/test__connector__cmdb_firewall_policy.py b/tests/test__connector__policy.py similarity index 100% rename from tests/test__connector__cmdb_firewall_policy.py rename to tests/test__connector__policy.py diff --git a/tests/test__connector_create.py b/tests/test__connector_create.py deleted file mode 100644 index a43fe1f3..00000000 --- a/tests/test__connector_create.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Test connector create.""" - -import pytest -from pytest_mock import MockerFixture - -from tests import helpers__tst as tst -from tests.helpers__tst import connectors_name, connectors_id - - -@pytest.mark.parametrize("data, expected", [ - ({"name": "NAME1"}, 500), # object exists - ({"name": "A/B"}, 500), # object exists - ({"name": "NAME9"}, 200), # object does not exist - ({"typo": "NAME1"}, KeyError), # invalid uid -]) -def test__create(connectors_name: list, connectors_id: list, mocker: MockerFixture, data, expected): - """Connector create()""" - mocker.patch(target="requests.Session.post", - side_effect=lambda *args, **kw: tst.session_post(mocker, *args, **kw)) - - connectors = [*connectors_name, *connectors_id] - for connector in connectors: - if isinstance(expected, int): - response = connector.create(data) - actual = response.status_code - assert actual == expected - else: - with pytest.raises(expected): - connector.create(data) diff --git a/tests/test__connector_delete.py b/tests/test__connector_delete.py deleted file mode 100644 index 1737f6db..00000000 --- a/tests/test__connector_delete.py +++ /dev/null @@ -1,52 +0,0 @@ -"""Test connector delete.""" - -import pytest -from pytest_mock import MockerFixture - -from tests import helpers__tst as tst -from tests.helpers__tst import connectors_name, connectors_id - - -@pytest.mark.parametrize("kwargs, expected", [ - ({"uid": "NAME1"}, 200), # object exists - ({"uid": "NAME9"}, 404), # object does not exist - ({"filter": "name==NAME1"}, 200), # object has been found - ({"filter": "name==NAME9"}, 404), # object has not been found - ({"uid": ""}, ValueError), # invalid uid - ({"uid": "NAME1", "filter": "name==NAME1"}, ValueError), # invalid uid -]) -def test__delete__name(connectors_name: list, mocker: MockerFixture, kwargs, expected): - """Connector delete()""" - mocker.patch(target="requests.Session.get", - side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) - mocker.patch(target="requests.Session.delete", - side_effect=lambda *args, **kw: tst.session_delete(mocker, *args, **kw)) - - for connector in connectors_name: - if isinstance(expected, int): - response = connector.delete(**kwargs) - actual = response.status_code - assert actual == expected - - -@pytest.mark.parametrize("kwargs, expected", [ - ({"uid": 1}, 200), # object exists - ({"uid": "1"}, 200), # object exists - ({"uid": 9}, 404), # object does not exist - ({"uid": 0}, ValueError), # invalid uid -]) -def test__delete__id(connectors_id: list, mocker: MockerFixture, kwargs, expected): - """Connector delete()""" - mocker.patch(target="requests.Session.get", - side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) - mocker.patch(target="requests.Session.delete", - side_effect=lambda *args, **kw: tst.session_delete(mocker, *args, **kw)) - - for connector in connectors_id: - if isinstance(expected, int): - response = connector.delete(**kwargs) - actual = response.status_code - assert actual == expected - else: - with pytest.raises(expected): - connector.delete(**kwargs) diff --git a/tests/test__connector_get.py b/tests/test__connector_get.py deleted file mode 100644 index 9ab8f293..00000000 --- a/tests/test__connector_get.py +++ /dev/null @@ -1,43 +0,0 @@ -"""Test connector get.""" - -import pytest -from pytest_mock import MockerFixture - -from tests import helpers__tst as tst -from tests.helpers__tst import connectors_name, connectors_id - - -@pytest.mark.parametrize("kwargs, expected", [ - (dict(name="NAME1"), ["NAME1"]), # object exists - (dict(name="A/B"), ["A/B"]), # object exists - (dict(name="NAME9"), []), # object does not exist - (dict(name="NAME1", filter="name==NAME1"), ["NAME1"]), # object has been found - (dict(filter="name==NAME1"), ["NAME1"]), # object has been found - (dict(filter="name==NAME9"), []), # object has not been found - (dict(typo="NAME1"), []), # invalid param -]) -def test__get__name(connectors_name: list, mocker: MockerFixture, kwargs, expected): - """Connector get()""" - mocker.patch(target="requests.Session.get", - side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) - - for connector in connectors_name: - items = connector.get(**kwargs) - actual = [d["name"] for d in items] - assert actual == expected - - -@pytest.mark.parametrize("kwargs, expected", [ - (dict(id=1), ["NAME1"]), # object exists - (dict(name="NAME9"), []), # object does not exist - (dict(typo="NAME1"), []), # invalid param -]) -def test__get__id(connectors_id: list, mocker: MockerFixture, kwargs, expected): - """Connector get()""" - mocker.patch(target="requests.Session.get", - side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) - - for connector in connectors_id: - items = connector.get(**kwargs) - actual = [d["name"] for d in items] - assert actual == expected diff --git a/tests/test__connector_is_exist.py b/tests/test__connector_is_exist.py deleted file mode 100644 index a97f3617..00000000 --- a/tests/test__connector_is_exist.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Test connector is_exist.""" - -import pytest -from pytest_mock import MockerFixture - -from tests import helpers__tst as tst -from tests.helpers__tst import connectors_name, connectors_id - - -@pytest.mark.parametrize("uid, expected", [ - ("NAME1", True), # object exists - ("NAME9", False), # object does not exist -]) -def test__is_exist__name(connectors_name: list, mocker: MockerFixture, uid, expected): - """Connector is_exist()""" - mocker.patch(target="requests.Session.get", - side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) - - for connector in connectors_name: - actual = connector.is_exist(uid=uid) - assert actual == expected, connector - - -@pytest.mark.parametrize("uid, expected", [ - (1, True), # object exists - (9, False), # object does not exist -]) -def test__is_exist__id(connectors_id: list, mocker: MockerFixture, uid, expected): - """Connector is_exist()""" - mocker.patch(target="requests.Session.get", - side_effect=lambda *args, **kw: tst.session_get(mocker, *args, **kw)) - - for connector in connectors_id: - actual = connector.is_exist(uid=uid) - assert actual == expected, connector diff --git a/tests/test__connector_update.py b/tests/test__connector_update.py deleted file mode 100644 index 4af6a5fc..00000000 --- a/tests/test__connector_update.py +++ /dev/null @@ -1,37 +0,0 @@ -"""Test connector update.""" - -import pytest -from pytest_mock import MockerFixture - -from tests import helpers__tst as tst -from tests.helpers__tst import connectors_name, connectors_id - - -@pytest.mark.parametrize("data, expected", [ - ({"name": "NAME1"}, 200), # object exists - ({"name": "NAME9"}, 404), # object does not exist -]) -def test__update__name(connectors_name: list, mocker: MockerFixture, data, expected): - """Connector update()""" - mocker.patch(target="requests.Session.put", - side_effect=lambda *args, **kw: tst.session_put(mocker, *args, **kw)) - - for connector in connectors_name: - response = connector.update(data=data) - actual = response.status_code - assert actual == expected - - -@pytest.mark.parametrize("data, expected", [ - ({"id": 1}, 200), # object exists - ({"id": 9}, 404), # object does not exist -]) -def test__update__id(connectors_id: list, mocker: MockerFixture, data, expected): - """Connector update()""" - mocker.patch(target="requests.Session.put", - side_effect=lambda *args, **kw: tst.session_put(mocker, *args, **kw)) - - for connector in connectors_id: - response = connector.update(data=data) - actual = response.status_code - assert actual == expected