diff --git a/README.md b/README.md index fa5524c..62daa6b 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,6 @@ The repository includes an SDK for the Grafana API. It's possible to communicate - Get all Alertmanager alerts - Create or update Alertmanager alerts - Get Alertmanager group alerts - - Get all Alertmanager silences - Get Alertmanager silence by id - Create or update Alertmanager silence @@ -94,13 +93,35 @@ The repository includes an SDK for the Grafana API. It's possible to communicate - Delete a notification channel by uid - Test a notification channel +### Organization +- Get current organisation +- Update the current organisation name +- Add a new user and the role to the current organisation +- Get all users from current organisation +- Get all users from current organisation (lookup) +- Update the role of an organisation user by the user id +- Delete an organisation user by the user id +- Get an organisation by the id +- Get an organisation by the name +- Get all organisations +- Create an organisation +- Update an organisation +- Delete an organisation +- Get organisation users +- Add a new organisation user +- Update an organisation user +- Delete an organisation user + +### Short URL +- Create a short url + ## Feature timeline The following table describes the plan to implement the rest of the Grafana API functionality. Please, open an issue and vote them up, if you prefer a faster implementation of an API functionality. | API endpoint group | Implementation week | Maintainer | PR | State | |:------------------:|:-------------------:|:----------:|:--:|:-----:| -| [Admin HTTP API](https://grafana.com/docs/grafana/latest/http_api/admin/) | | | | | +| [Admin HTTP API](https://grafana.com/docs/grafana/latest/http_api/admin/) | 16 | [ZPascal](https://github.com/ZPascal) | | | | [Annotations HTTP API](https://grafana.com/docs/grafana/latest/http_api/annotations/) | | | | | | [Authentication HTTP API](https://grafana.com/docs/grafana/latest/http_api/auth/) | | | | | | [External Group Sync HTTP API](https://grafana.com/docs/grafana/latest/http_api/external_group_sync/) | | | | | @@ -109,13 +130,11 @@ The following table describes the plan to implement the rest of the Grafana API | [HTTP Snapshot API](https://grafana.com/docs/grafana/latest/http_api/snapshot/) | | | | | | [Library Element HTTP API](https://grafana.com/docs/grafana/latest/http_api/library_element/) | | | | | | [Licensing HTTP API](https://grafana.com/docs/grafana/latest/http_api/licensing/) | | | | | -| [Organization HTTP API](https://grafana.com/docs/grafana/latest/http_api/org/) | 13 | | | In process | | [Other HTTP API](https://grafana.com/docs/grafana/latest/http_api/other/) | | | | | | [Playlist HTTP API](https://grafana.com/docs/grafana/latest/http_api/playlist/) | | | | | | [Reporting API](https://grafana.com/docs/grafana/latest/http_api/reporting/) | | | | | -| [Short URL HTTP API](https://grafana.com/docs/grafana/latest/http_api/short_url/) | 13 | | | In process | | [Team HTTP API](https://grafana.com/docs/grafana/latest/http_api/team/) | | | | | -| [User HTTP API](https://grafana.com/docs/grafana/latest/http_api/user/) | | | | | +| [User HTTP API](https://grafana.com/docs/grafana/latest/http_api/user/) | 16 | [ZPascal](https://github.com/ZPascal) | | | ## Installation diff --git a/docs/content/grafana_api/model.md b/docs/content/grafana_api/model.md index 3b81c75..2af0fbf 100644 --- a/docs/content/grafana_api/model.md +++ b/docs/content/grafana_api/model.md @@ -50,6 +50,8 @@ The class includes all necessary variables to establish a connection to the Graf - `host` _str_ - Specify the host of the Grafana system - `token` _str_ - Specify the access token of the Grafana system +- `username` _str_ - Specify the username of the Grafana system +- `password` _str_ - Specify the password of the Grafana system diff --git a/docs/coverage.svg b/docs/coverage.svg index 6bfc8fa..e5db27c 100644 --- a/docs/coverage.svg +++ b/docs/coverage.svg @@ -15,7 +15,7 @@ coverage coverage - 99% - 99% + 100% + 100% diff --git a/requirements.txt b/requirements.txt index b08fa6b..f8e7df0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ requests -pydoc-markdown +pydoc-markdown==4.6.2 mkdocs mkdocs-material \ No newline at end of file diff --git a/src/grafana_api/api.py b/src/grafana_api/api.py index a948207..6f574f4 100644 --- a/src/grafana_api/api.py +++ b/src/grafana_api/api.py @@ -38,11 +38,22 @@ def call_the_api( """ api_url: str = f"{self.grafana_api_model.host}{api_call}" + headers: dict = dict( + {"Authorization": f"Bearer {self.grafana_api_model.token}"} + ) + + if ( + self.grafana_api_model.username is not None + and self.grafana_api_model.password is not None + ): + url: str = ( + f"{self.grafana_api_model.username}:{self.grafana_api_model.password}@" + ) + api_url = api_url.replace("https://", f"https://{url}") + api_url = api_url.replace("http://", f"http://{url}") + else: + headers["Content-Type"] = "application/json" - headers: dict = { - "Authorization": f"Bearer {self.grafana_api_model.token}", - "Content-Type": "application/json", - } try: if method.value == RequestsMethods.GET.value: return Api.__check_the_api_call_response( @@ -64,6 +75,14 @@ def call_the_api( else: logging.error("Please define the json_complete.") raise Exception + elif method.value == RequestsMethods.PATCH.value: + if json_complete is not None: + return Api.__check_the_api_call_response( + requests.patch(api_url, data=json_complete, headers=headers) + ) + else: + logging.error("Please define the json_complete.") + raise Exception elif method.value == RequestsMethods.DELETE.value: return Api.__check_the_api_call_response( requests.delete(api_url, headers=headers) diff --git a/src/grafana_api/model.py b/src/grafana_api/model.py index 39d9f1d..2ea04c8 100644 --- a/src/grafana_api/model.py +++ b/src/grafana_api/model.py @@ -20,6 +20,9 @@ class APIEndpoints(Enum): ALERTS_NGALERT = "/api/v1/ngalert" DATASOURCES = "/api/datasources" DATASOURCE_QUERY = "/api/tsdb/query" + SHORT_URLS = "/api/short-urls" + ORGANISATION = "/api/org" + ORGANISATIONS = "/api/orgs" class RequestsMethods(Enum): @@ -28,6 +31,7 @@ class RequestsMethods(Enum): GET = "GET" PUT = "PUT" POST = "POST" + PATCH = "PATCH" DELETE = "DELETE" @@ -37,10 +41,14 @@ class APIModel(NamedTuple): Args: host (str): Specify the host of the Grafana system token (str): Specify the access token of the Grafana system + username (str): Specify the username of the Grafana system + password (str): Specify the password of the Grafana system """ host: str - token: str + token: str = None + username: str = None + password: str = None class DatasourceQuery(NamedTuple): diff --git a/src/grafana_api/organisation.py b/src/grafana_api/organisation.py new file mode 100644 index 0000000..f71d12a --- /dev/null +++ b/src/grafana_api/organisation.py @@ -0,0 +1,629 @@ +import json +import logging + +from .model import APIModel, APIEndpoints, RequestsMethods +from .api import Api + + +class Organisation: + """The class includes all necessary methods to access the Grafana organisation API endpoint + + Args: + grafana_api_model (APIModel): Inject a Grafana API model object that includes all necessary values and information + + Attributes: + grafana_api_model (APIModel): This is where we store the grafana_api_model + """ + + def __init__(self, grafana_api_model: APIModel): + self.grafana_api_model = grafana_api_model + + def get_current_organization(self) -> dict: + """The method includes a functionality to get the current organization. + + Required Permissions: + Action: orgs:read + Scope: N/A + + Raises: + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the current organization + """ + + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api(APIEndpoints.ORGANISATION.value) + .json() + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + + def get_all_users_by_the_current_organization(self) -> list: + """The method includes a functionality to get all users from the current organization. + + Required Permissions: + Action: org.users:read + Scope: users:* + + Raises: + Exception: Unspecified error by executing the API call + + Returns: + api_call (list): Returns the users of the current organization + """ + + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api(f"{APIEndpoints.ORGANISATION.value}/users") + .json() + ) + + if api_call == list() or api_call[0].get("orgId") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + + def get_all_users_by_the_current_organization_lookup(self) -> list: + """The method includes a functionality to get the lookup information of all users from the current organization. + + Required Permissions: + Action: org.users:read + Scope: users:* + + Raises: + Exception: Unspecified error by executing the API call + + Returns: + api_call (list): Returns the users of the current organization + """ + + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api(f"{APIEndpoints.ORGANISATION.value}/users/lookup") + .json() + ) + + if api_call == list() or api_call[0].get("userId") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + + def update_organization_user_role_by_user_id(self, user_id: int, role: str): + """The method includes a functionality to update the current organization user by the user id. + + Args: + user_id (int): Specify the id of the user + role (str): Specify the role of the user + + Required Permissions: + Action: org.users.role:update + Scope: users:* + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if user_id != 0 and len(role) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATION.value}/users/{user_id}", + RequestsMethods.PATCH, + json.dumps(dict({"role": role})), + ) + .json() + ) + + if api_call.get("message") != "Organization user updated": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully updated the organization user.") + else: + logging.error("There is no user_id or dict defined.") + raise ValueError + + def delete_organization_user_by_user_id(self, user_id: int): + """The method includes a functionality to delete the current organization user by the user id. + + Args: + user_id (int): Specify the id of the user + + Required Permissions: + Action: org.users:remove + Scope: users:* + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if user_id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATION.value}/users/{user_id}", + RequestsMethods.DELETE, + ) + .json() + ) + + if api_call.get("message") != "User removed from organization": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully removed the organization user.") + else: + logging.error("There is no user_id defined.") + raise ValueError + + def update_current_organization(self, name: str): + """The method includes a functionality to update the current organization. + + Args: + name (str): Specify the new name of the current organization + + Required Permissions: + Action: orgs:write + Scope: N/A + + Raises + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if len(name) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + APIEndpoints.ORGANISATION.value, + RequestsMethods.PUT, + json.dumps(dict({"name": name})), + ) + .json() + ) + + if api_call.get("message") != "Organization updated": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully updated the organization.") + else: + logging.error("There is no role_name defined.") + raise ValueError + + def add_new_user_to_current_organization( + self, login_or_email: str, role: str + ) -> int: + """The method includes a functionality to add a new user to the current organization. + + Args: + login_or_email (str): Specify the added user + role (str): Specify the added role for the user + + Required Permissions: + Action: org.users:add + Scope: users:* + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + user_id (int): Returns the id of the created user + """ + + if len(login_or_email) != 0 and len(role) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATION.value}/users", + RequestsMethods.POST, + json.dumps(dict({"loginOrEmail": login_or_email, "role": role})), + ) + .json() + ) + + if api_call.get("message") != "User added to organization": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call.get("userId") + else: + logging.error("There is no user_role defined.") + raise ValueError + + +class OrganisationAdmin: + """The class includes all necessary methods to access the Grafana organisation Admin API endpoint. Be aware that all functionalities inside the class only working with basic authentication (username and password). + + Args: + grafana_api_model (APIModel): Inject a Grafana API model object that includes all necessary values and information + + Attributes: + grafana_api_model (APIModel): This is where we store the grafana_api_model + """ + + def __init__(self, grafana_api_model: APIModel): + self.grafana_api_model = grafana_api_model + + def get_organization_by_id(self, org_id: int) -> dict: + """The method includes a functionality to get an organization by the id. + + Args: + org_id (int): Specify the organization id + + Required Permissions: + Action: orgs:read + Scope: N/A + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the organization as dict + """ + + if org_id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api(f"{APIEndpoints.ORGANISATIONS.value}/{org_id}") + .json() + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.error("There is no org_id defined.") + raise ValueError + + def get_organization_by_name(self, name: str) -> dict: + """The method includes a functionality to get an organization by the name. + + Args: + name (str): Specify the organization name + + Required Permissions: + Action: orgs:read + Scope: N/A + Note: Needs to be assigned globally. + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the organization as dict + """ + + if len(name) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api(f"{APIEndpoints.ORGANISATIONS.value}/name/{name}") + .json() + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.error("There is no name defined.") + raise ValueError + + def get_organizations(self) -> list: + """The method includes a functionality to get all organizations. + + Required Permissions: + Action: orgs:read + Scope: N/A + Note: Needs to be assigned globally. + + Raises: + Exception: Unspecified error by executing the API call + + Returns: + api_call (list): Returns all organizations as list + """ + + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api(APIEndpoints.ORGANISATIONS.value) + .json() + ) + + if api_call == list() or api_call[0].get("id") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + + def create_organization(self, name: str) -> int: + """The method includes a functionality to create an organization. + + Args: + name (str): Specify the organization name + + Required Permissions: + Action: orgs:create + Scope: N/A + Note: Needs to be assigned globally. + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + org_id (int): Returns the id of the created organization + """ + + if len(name) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + APIEndpoints.ORGANISATIONS.value, + RequestsMethods.POST, + json.dumps(dict({"name": name})), + ) + .json() + ) + + if api_call.get("message") != "Organization created": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return int(api_call.get("orgId")) + else: + logging.error("There is no name defined.") + raise ValueError + + def update_organization(self, org_id: int, name: str): + """The method includes a functionality to update an organization. + + Args: + org_id (int): Specify the organization id + name (str): Specify the organization name + + Required Permissions: + Action: orgs:write + Scope: N/A + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if org_id != 0 and len(name) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATIONS.value}/{org_id}", + RequestsMethods.PUT, + json.dumps(dict({"name": name})), + ) + .json() + ) + + if api_call.get("message") != "Organization updated": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully updated the organization.") + else: + logging.error("There is no org_id or name defined.") + raise ValueError + + def delete_organization(self, org_id: int): + """The method includes a functionality to delete an organization. + + Args: + org_id (int): Specify the organization id + + Required Permissions: + Action: orgs:delete + Scope: N/A + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if org_id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATIONS.value}/{org_id}", + RequestsMethods.DELETE, + ) + .json() + ) + + if api_call.get("message") != "Organization deleted": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully deleted the organization.") + else: + logging.error("There is no org_id defined.") + raise ValueError + + def get_organization_users(self, org_id: int) -> list: + """The method includes a functionality to get all organization users specified by the organization id. + + Args: + org_id (int): Specify the organization id + + Required Permissions: + Action: org.users:read + Scope: users:* + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (list): Returns all organization users as list + """ + + if org_id != 0: + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api(f"{APIEndpoints.ORGANISATIONS.value}/{org_id}/users") + .json() + ) + + if api_call == list() or api_call[0].get("orgId") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.error("There is no org_id defined.") + raise ValueError + + def add_organization_user(self, org_id: int, login_or_email: str, role: str) -> int: + """The method includes a functionality to add a user to an organization. + + Args: + org_id (int): Specify the organization id + login_or_email (str): Specify the added user + role (str): Specify the added role for the user + + Required Permissions: + Action: org.users:add + Scope: users:* + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + user_id (int): Returns the added user id + """ + + if org_id != 0 and len(login_or_email) != 0 and len(role) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATIONS.value}/{org_id}/users", + RequestsMethods.POST, + json.dumps(dict({"loginOrEmail": login_or_email, "role": role})), + ) + .json() + ) + + if api_call.get("message") != "User added to organization": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call.get("userId") + else: + logging.error("There is no org_id, login_or_email or role defined.") + raise ValueError + + def update_organization_user(self, org_id: int, user_id: int, role: str): + """The method includes a functionality to update organization user specified by the organization id, the user_id and the role. + + Args: + org_id (int): Specify the organization id + user_id (int): Specify the user id + role (str): Specify the added role for the user + + Required Permissions: + Action: org.users.role:update + Scope: users:* + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if org_id != 0 and user_id != 0 and len(role) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATIONS.value}/{org_id}/users/{user_id}", + RequestsMethods.PATCH, + json.dumps(dict({"role": role})), + ) + .json() + ) + + if api_call.get("message") != "Organization user updated": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully updated user inside the organization.") + else: + logging.error("There is no org_id, user_id or role defined.") + raise ValueError + + def delete_organization_user(self, org_id: int, user_id: int): + """The method includes a functionality to remove an organization users specified by the organization id and the user id. + + Args: + org_id (int): Specify the organization id + user_id (int): Specify the user id + + Required Permissions: + Action: org.users:remove + Scope: users:* + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if org_id != 0 and user_id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.ORGANISATIONS.value}/{org_id}/users/{user_id}", + RequestsMethods.DELETE, + ) + .json() + ) + + if api_call.get("message") != "User removed from organization": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully removed user from the organization.") + else: + logging.error("There is no org_id or user_id defined.") + raise ValueError diff --git a/src/grafana_api/short_url.py b/src/grafana_api/short_url.py new file mode 100644 index 0000000..0b62355 --- /dev/null +++ b/src/grafana_api/short_url.py @@ -0,0 +1,53 @@ +import json +import logging + +from .model import APIModel, APIEndpoints, RequestsMethods +from .api import Api + + +class ShortUrl: + """The class includes all necessary methods to access the Grafana short url API endpoint + + Args: + grafana_api_model (APIModel): Inject a Grafana API model object that includes all necessary values and information + + Attributes: + grafana_api_model (APIModel): This is where we store the grafana_api_model + """ + + def __init__(self, grafana_api_model: APIModel): + self.grafana_api_model = grafana_api_model + + def create_short_url(self, path: str): + """The method includes a functionality to create a short link for a specific dashboard + + Args: + path (str): Specify the corresponding dashboard path + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the uid and the url of the newly generated link + """ + + if len(path) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + APIEndpoints.SHORT_URLS.value, + RequestsMethods.POST, + json.dumps(dict({"path": path})), + ) + .json() + ) + + if api_call == dict() or api_call.get("url") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.error("There is no path defined.") + raise ValueError diff --git a/tests/integrationtest/test_organisation.py b/tests/integrationtest/test_organisation.py new file mode 100644 index 0000000..56534ec --- /dev/null +++ b/tests/integrationtest/test_organisation.py @@ -0,0 +1,73 @@ +import os +from unittest import TestCase + +from src.grafana_api.model import ( + APIModel, +) +from src.grafana_api.organisation import Organisation + + +class OrganisationTest(TestCase): + model: APIModel = APIModel( + host=os.environ["GRAFANA_HOST"], + token=os.environ["GRAFANA_TOKEN"], + ) + organisation: Organisation = Organisation(model) + + def test_get_current_organization(self): + organisation: dict = self.organisation.get_current_organization() + + self.assertEqual(4, organisation.get("id")) + + def test_get_all_users_by_the_current_organization(self): + organisation_users: list = ( + self.organisation.get_all_users_by_the_current_organization() + ) + + self.assertEqual(4, organisation_users[0].get("userId")) + + def test_get_all_users_by_the_current_organization_lookup(self): + organisation_users: list = ( + self.organisation.get_all_users_by_the_current_organization_lookup() + ) + + self.assertEqual(4, organisation_users[0].get("userId")) + + def test_a_update_current_organization(self): + self.organisation.update_current_organization("Test") + self.assertEqual( + "Test", self.organisation.get_current_organization().get("name") + ) + + def test_b_update_current_organization(self): + self.organisation.update_current_organization("Github") + self.assertEqual( + "Github", self.organisation.get_current_organization().get("name") + ) + + def test_a_add_new_user_to_current_organization(self): + self.organisation.add_new_user_to_current_organization( + "info@theiotstudio.com", "Viewer" + ) + + organisation_users: list = ( + self.organisation.get_all_users_by_the_current_organization_lookup() + ) + self.assertEqual("Test", organisation_users[0].get("login")) + + def test_b_update_organization_user_role_by_user_id(self): + self.organisation.update_organization_user_role_by_user_id(7, "Editor") + + organisation_users: list = ( + self.organisation.get_all_users_by_the_current_organization() + ) + self.assertEqual(7, organisation_users[0].get("userId")) + self.assertEqual("Editor", organisation_users[0].get("role")) + + def test_c_delete_organization_user_by_user_id(self): + self.organisation.delete_organization_user_by_user_id(7) + + organisation_users: list = ( + self.organisation.get_all_users_by_the_current_organization() + ) + self.assertEqual(2, len(organisation_users)) diff --git a/tests/integrationtest/test_short_url.py b/tests/integrationtest/test_short_url.py new file mode 100644 index 0000000..0996ae7 --- /dev/null +++ b/tests/integrationtest/test_short_url.py @@ -0,0 +1,18 @@ +import os +from unittest import TestCase + +from src.grafana_api.model import APIModel +from src.grafana_api.short_url import ShortUrl + + +class ShortUrlTest(TestCase): + model: APIModel = APIModel( + host=os.environ["GRAFANA_HOST"], + token=os.environ["GRAFANA_TOKEN"], + ) + short_url: ShortUrl = ShortUrl(model) + + def test_create_short_url(self): + self.assertIsNotNone( + self.short_url.create_short_url("d/test1/test-1?orgId=4&from=now-1h&to=now") + ) diff --git a/tests/unittests/test_alerting.py b/tests/unittests/test_alerting.py index d4805de..c0364c1 100644 --- a/tests/unittests/test_alerting.py +++ b/tests/unittests/test_alerting.py @@ -470,7 +470,7 @@ def test_test_alertmanager_receivers_test_not_possible(self, call_the_api_mock): with self.assertRaises(Exception): alerting.test_alertmanager_receivers( - {"test": "test"}, alertmanager_receivers + {"test": "test"}, list([alertmanager_receivers]) ) @patch("src.grafana_api.api.Api.call_the_api") @@ -495,7 +495,7 @@ def test_test_alertmanager_receivers_test_not_found(self, call_the_api_mock): with self.assertRaises(Exception): alerting.test_alertmanager_receivers( - {"test": "test"}, alertmanager_receivers + {"test": "test"}, list([alertmanager_receivers]) ) @patch("src.grafana_api.api.Api.call_the_api") @@ -787,7 +787,9 @@ def test_test_rule(self, call_the_api_mock): call_the_api_mock.return_value = mock - self.assertEqual(dict({"test": "test"}), alerting.test_rule([datasource_rule_query])) + self.assertEqual( + dict({"test": "test"}), alerting.test_rule([datasource_rule_query]) + ) def test_test_rule_no_data_query(self): model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) diff --git a/tests/unittests/test_api.py b/tests/unittests/test_api.py index e9497ab..ccd314a 100644 --- a/tests/unittests/test_api.py +++ b/tests/unittests/test_api.py @@ -10,23 +10,22 @@ class ApiTestCase(TestCase): - def test_call_the_api_non_method(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + api: Api = Api(grafana_api_model=model) + def test_call_the_api_non_method(self): with self.assertRaises(Exception): - api.call_the_api(api_call=MagicMock(), method=None) + self.api.call_the_api(api_call=MagicMock(), method=None) def test_call_the_api_non_valid_method(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - with self.assertRaises(Exception): - api.call_the_api(api_call=MagicMock(), method=MagicMock()) + self.api.call_the_api(api_call=MagicMock(), method=MagicMock()) @patch("requests.get") - def test_call_the_api_get_valid(self, get_mock): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + def test_call_the_api_basic_auth(self, get_mock): + model: APIModel = APIModel( + host="https://test.test.de", username="test", password="test" + ) api: Api = Api(grafana_api_model=model) mock: Mock = Mock() @@ -39,18 +38,24 @@ def test_call_the_api_get_valid(self, get_mock): api.call_the_api(api_call=MagicMock()).json()["status"], ) - def test_call_the_api_get_not_valid(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) + @patch("requests.get") + def test_call_the_api_get_valid(self, get_mock): + mock: Mock = Mock() + mock.json = Mock(return_value={"status": "success"}) + + get_mock.return_value = mock + self.assertEqual( + "success", + self.api.call_the_api(api_call=MagicMock()).json()["status"], + ) + + def test_call_the_api_get_not_valid(self): with self.assertRaises(MissingSchema): - api.call_the_api(api_call=MagicMock(), method=RequestsMethods.GET) + self.api.call_the_api(api_call=MagicMock(), method=RequestsMethods.GET) @patch("requests.put") def test_call_the_api_put_valid(self, put_mock): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - mock: Mock = Mock() mock.json = Mock(return_value={"status": "success"}) @@ -58,7 +63,7 @@ def test_call_the_api_put_valid(self, put_mock): self.assertEqual( "success", - api.call_the_api( + self.api.call_the_api( api_call=MagicMock(), method=RequestsMethods.PUT, json_complete=MagicMock(), @@ -66,17 +71,11 @@ def test_call_the_api_put_valid(self, put_mock): ) def test_call_the_api_put_not_valid(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - with self.assertRaises(Exception): - api.call_the_api(api_call=MagicMock(), method=RequestsMethods.PUT) + self.api.call_the_api(api_call=MagicMock(), method=RequestsMethods.PUT) @patch("requests.post") def test_call_the_api_post_valid(self, post_mock): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - mock: Mock = Mock() mock.json = Mock(return_value={"status": "success"}) @@ -84,7 +83,7 @@ def test_call_the_api_post_valid(self, post_mock): self.assertEqual( "success", - api.call_the_api( + self.api.call_the_api( api_call=MagicMock(), method=RequestsMethods.POST, json_complete=MagicMock(), @@ -92,28 +91,47 @@ def test_call_the_api_post_valid(self, post_mock): ) def test_call_the_api_post_not_valid(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - with self.assertRaises(MissingSchema): - api.call_the_api( + self.api.call_the_api( api_call=MagicMock(), method=RequestsMethods.POST, json_complete=MagicMock(), ) def test_call_the_api_post_no_data(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) + with self.assertRaises(Exception): + self.api.call_the_api(api_call=MagicMock(), method=RequestsMethods.POST) + @patch("requests.patch") + def test_call_the_api_patch_valid(self, post_mock): + mock: Mock = Mock() + mock.json = Mock(return_value={"status": "success"}) + + post_mock.return_value = mock + + self.assertEqual( + "success", + self.api.call_the_api( + api_call=MagicMock(), + method=RequestsMethods.PATCH, + json_complete=MagicMock(), + ).json()["status"], + ) + + def test_call_the_api_patch_not_valid(self): + with self.assertRaises(MissingSchema): + self.api.call_the_api( + api_call=MagicMock(), + method=RequestsMethods.PATCH, + json_complete=MagicMock(), + ) + + def test_call_the_api_patch_no_data(self): with self.assertRaises(Exception): - api.call_the_api(api_call=MagicMock(), method=RequestsMethods.POST) + self.api.call_the_api(api_call=MagicMock(), method=RequestsMethods.PATCH) @patch("requests.delete") def test_call_the_api_delete_valid(self, delete_mock): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - mock: Mock = Mock() mock.json = Mock(return_value={"message": "Deletion successful"}) @@ -121,59 +139,44 @@ def test_call_the_api_delete_valid(self, delete_mock): self.assertEqual( "Deletion successful", - api.call_the_api( + self.api.call_the_api( api_call=MagicMock(), method=RequestsMethods.DELETE ).json()["message"], ) def test_call_the_api_delete_not_valid(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - with self.assertRaises(Exception): - api.call_the_api(api_call=MagicMock(), method=RequestsMethods.DELETE) + self.api.call_the_api(api_call=MagicMock(), method=RequestsMethods.DELETE) def test_check_the_api_call_response(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - mock: Mock = Mock() mock.json = Mock(return_value=dict({"test": "test"})) self.assertEqual( dict({"test": "test"}), - api._Api__check_the_api_call_response(response=mock).json(), + self.api._Api__check_the_api_call_response(response=mock).json(), ) def test_check_the_api_call_response_no_error_message(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - mock: Mock = Mock() mock.json = Mock(return_value=dict({"message": "test"})) self.assertEqual( dict({"message": "test"}), - api._Api__check_the_api_call_response(response=mock).json(), + self.api._Api__check_the_api_call_response(response=mock).json(), ) def test_check_the_api_call_response_no_json_response_value(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - mock: Mock = Mock() mock.text = Mock(return_value="test") self.assertEqual( - "test", api._Api__check_the_api_call_response(response=mock).text() + "test", self.api._Api__check_the_api_call_response(response=mock).text() ) def test_check_the_api_call_response_exception(self): - model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) - api: Api = Api(grafana_api_model=model) - mock: Mock = Mock() mock.json = Mock(return_value=dict({"message": "invalid API key"})) with self.assertRaises(requests.exceptions.ConnectionError): - api._Api__check_the_api_call_response(response=mock) + self.api._Api__check_the_api_call_response(response=mock) diff --git a/tests/unittests/test_model.py b/tests/unittests/test_model.py index ebe38df..e2cd93b 100644 --- a/tests/unittests/test_model.py +++ b/tests/unittests/test_model.py @@ -17,7 +17,9 @@ def test_api_endpoints_init(self): class RequestsMethodsTestCase(TestCase): def test_requests_methods_init(self): self.assertEqual("RequestsMethods.GET", str(RequestsMethods.GET)) + self.assertEqual("RequestsMethods.PUT", str(RequestsMethods.PUT)) self.assertEqual("RequestsMethods.POST", str(RequestsMethods.POST)) + self.assertEqual("RequestsMethods.PATCH", str(RequestsMethods.PATCH)) self.assertEqual("RequestsMethods.DELETE", str(RequestsMethods.DELETE)) @@ -28,6 +30,13 @@ def test_api_model_init(self): self.assertEqual("test", model.host) self.assertEqual("test", model.token) + def test_api_model_init_basic_auth(self): + model = APIModel(host="test", username="test", password="test") + + self.assertEqual("test", model.host) + self.assertEqual("test", model.username) + self.assertEqual("test", model.password) + class DatasourceQueryTestCase(TestCase): def test_datasource_query_init(self): diff --git a/tests/unittests/test_organisation.py b/tests/unittests/test_organisation.py new file mode 100644 index 0000000..71c262f --- /dev/null +++ b/tests/unittests/test_organisation.py @@ -0,0 +1,510 @@ +from unittest import TestCase +from unittest.mock import MagicMock, Mock, patch + +from src.grafana_api.model import APIModel +from src.grafana_api.organisation import Organisation, OrganisationAdmin + + +class OrganisationTestCase(TestCase): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + organisation: Organisation = Organisation(grafana_api_model=model) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_current_organization(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"id": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual(dict({"id": 1}), self.organisation.get_current_organization()) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_current_organization_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.get_current_organization() + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_all_users_by_the_current_organization(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=list([dict({"orgId": 1})])) + + call_the_api_mock.return_value = mock + + self.assertEqual( + list([dict({"orgId": 1})]), + self.organisation.get_all_users_by_the_current_organization(), + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_all_users_by_the_current_organization_error_response( + self, call_the_api_mock + ): + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.get_all_users_by_the_current_organization() + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_all_users_by_the_current_organization_lookup(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=list([dict({"userId": 1})])) + + call_the_api_mock.return_value = mock + + self.assertEqual( + list([dict({"userId": 1})]), + self.organisation.get_all_users_by_the_current_organization_lookup(), + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_all_users_by_the_current_organization_lookup_error_response( + self, call_the_api_mock + ): + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.get_all_users_by_the_current_organization_lookup() + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_user_role_by_user_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Organization user updated"})) + + call_the_api_mock.return_value = mock + + self.assertEqual( + None, + self.organisation.update_organization_user_role_by_user_id(1, "Viewer"), + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_user_role_by_user_id_no_user_id( + self, call_the_api_mock + ): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.update_organization_user_role_by_user_id(0, "") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_user_role_by_user_id_error_response( + self, call_the_api_mock + ): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.update_organization_user_role_by_user_id(1, "Viewer") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_user_by_user_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock( + return_value=dict({"message": "User removed from organization"}) + ) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, self.organisation.delete_organization_user_by_user_id(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_user_by_user_id_no_user_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.delete_organization_user_by_user_id(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_user_by_user_id_error_response( + self, call_the_api_mock + ): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.delete_organization_user_by_user_id(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_current_organization(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Organization updated"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, self.organisation.update_current_organization("test")) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_current_organization_no_role_name(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.update_current_organization("") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_current_organization_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.update_current_organization("test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_new_user_to_current_organization(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock( + return_value=dict({"message": "User added to organization", "userId": 1}) + ) + + call_the_api_mock.return_value = mock + + self.assertEqual( + 1, self.organisation.add_new_user_to_current_organization("test", "test") + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_new_user_to_current_organization_no_role(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.add_new_user_to_current_organization("", "") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_new_user_to_current_organization_error_response( + self, call_the_api_mock + ): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.add_new_user_to_current_organization("test", "test") + + +class OrganisationAdminTestCase(TestCase): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + organisation: OrganisationAdmin = OrganisationAdmin(grafana_api_model=model) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_by_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"id": 10})) + + call_the_api_mock.return_value = mock + + self.assertEqual(dict({"id": 10}), self.organisation.get_organization_by_id(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_by_id_no_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.get_organization_by_id(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_by_id_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.get_organization_by_id(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_by_name(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"id": 10})) + + call_the_api_mock.return_value = mock + + self.assertEqual( + dict({"id": 10}), self.organisation.get_organization_by_name("test") + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_by_name_no_name(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.get_organization_by_name("") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_by_name_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.get_organization_by_name("test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organizations(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=list([{"id": 1}])) + + call_the_api_mock.return_value = mock + + self.assertEqual(list([{"id": 1}]), self.organisation.get_organizations()) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organizations_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.get_organizations() + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_organization(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock( + return_value=dict({"message": "Organization created", "orgId": 10}) + ) + + call_the_api_mock.return_value = mock + + self.assertEqual(10, self.organisation.create_organization("test")) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_organization_no_name(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.create_organization("") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_organization_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.create_organization("test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Organization updated"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, self.organisation.update_organization(1, "test")) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_no_org_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.update_organization(0, "") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.update_organization(1, "test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Organization deleted"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, self.organisation.delete_organization(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_no_org_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.delete_organization(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.delete_organization(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_users(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=list([dict({"orgId": 1, "userId": 10})])) + + call_the_api_mock.return_value = mock + + self.assertEqual( + list([dict({"orgId": 1, "userId": 10})]), + self.organisation.get_organization_users(1), + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_users_no_org_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.get_organization_users(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_organization_users_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.get_organization_users(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_organization_user(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock( + return_value=dict({"message": "User added to organization", "userId": 10}) + ) + + call_the_api_mock.return_value = mock + + self.assertEqual(10, self.organisation.add_organization_user(1, "test", "test")) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_organization_user_no_org_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.add_organization_user(0, "", "") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_organization_user_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.add_organization_user(1, "test", "test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_user(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Organization user updated"})) + + call_the_api_mock.return_value = mock + + self.assertEqual( + None, self.organisation.update_organization_user(1, 10, "test") + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_user_no_org_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.update_organization_user(0, 0, "") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_organization_user_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.update_organization_user(1, 10, "test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_user(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock( + return_value=dict({"message": "User removed from organization"}) + ) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, self.organisation.delete_organization_user(1, 10)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_user_no_org_id(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + self.organisation.delete_organization_user(0, 0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_organization_user_error_response(self, call_the_api_mock): + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + self.organisation.delete_organization_user(1, 10) diff --git a/tests/unittests/test_search.py b/tests/unittests/test_search.py index bbbcebc..20fb32b 100644 --- a/tests/unittests/test_search.py +++ b/tests/unittests/test_search.py @@ -29,7 +29,7 @@ def test_search_invalid_empty_list(self, call_the_api_mock): call_the_api_mock.return_value = mock with self.assertRaises(Exception): - search.search(search_query=MagicMock()) + search.search(search_query="test") @patch("src.grafana_api.api.Api.call_the_api") def test_search_invalid_output(self, call_the_api_mock): diff --git a/tests/unittests/test_short_url.py b/tests/unittests/test_short_url.py new file mode 100644 index 0000000..7429ec6 --- /dev/null +++ b/tests/unittests/test_short_url.py @@ -0,0 +1,45 @@ +from unittest import TestCase +from unittest.mock import MagicMock, Mock, patch + +from src.grafana_api.model import APIModel +from src.grafana_api.short_url import ShortUrl + + +class ShortUrlTestCase(TestCase): + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_short_url(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + short_url: ShortUrl = ShortUrl(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"url": "test"})) + + call_the_api_mock.return_value = mock + + self.assertEqual("test", short_url.create_short_url(path="Test").get("url")) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_short_url_invalid_path(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + short_url: ShortUrl = ShortUrl(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + short_url.create_short_url(path="") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_short_url_invalid_output(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + short_url: ShortUrl = ShortUrl(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"url": None})) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + short_url.create_short_url(path="test")