diff --git a/.github/workflows/integrationtest.yml b/.github/workflows/integrationtest.yml index a19e5b5..fa1aeb4 100644 --- a/.github/workflows/integrationtest.yml +++ b/.github/workflows/integrationtest.yml @@ -25,4 +25,4 @@ jobs: GRAFANA_DASHBOARD_PATH: ${{ secrets.GRAFANA_DASHBOARD_PATH }} GRAFANA_DASHBOARD_NAME: ${{ secrets.GRAFANA_DASHBOARD_NAME }} GRAFANA_USERNAME: ${{ secrets.GRAFANA_USERNAME }} - GRAFANA_PASSWORD: ${{ secrets.GRAFANA_PASSWORD }} + GRAFANA_PASSWORD: ${{ secrets.GRAFANA_PASSWORD }} \ No newline at end of file diff --git a/README.md b/README.md index 16c0c11..f6b216f 100644 --- a/README.md +++ b/README.md @@ -125,7 +125,7 @@ In general my focus inside this project is to implement and deliver old and new - Create a short url ### User -- Search Users +- Search users - Get user by id - Get user by username or email - Update the user @@ -150,6 +150,27 @@ In general my focus inside this project is to implement and deliver old and new - Delete snapshot by key - Delete snapshot by delete key +### Team +- Search team +- Get team by id +- Add team +- Update team +- Delete team by id +- Get team members +- Add team member +- Delete team member +- Get team preferences +- Update team preferences + +### Playlist +- Search playlist +- Get playlist +- Get playlist items +- Get playlist dashboards +- Create playlist +- Update playlist +- Delete playlist + ## 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. @@ -163,11 +184,10 @@ The following table describes the plan to implement the rest of the Grafana API | [Fine-grained access control HTTP API](https://grafana.com/docs/grafana/latest/http_api/access_control/) | | | | | | [HTTP Preferences API](https://grafana.com/docs/grafana/latest/http_api/preferences/) | | | | | | [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/) | | | | | -| [Other HTTP API](https://grafana.com/docs/grafana/latest/http_api/other/) | | | | | -| [Playlist HTTP API](https://grafana.com/docs/grafana/latest/http_api/playlist/) | 20 | [ZPascal](https://github.com/ZPascal) | | | -| [Reporting API](https://grafana.com/docs/grafana/latest/http_api/reporting/) | | | | | -| [Team HTTP API](https://grafana.com/docs/grafana/latest/http_api/team/) | 20 | [ZPascal](https://github.com/ZPascal) | | | +| [Licensing HTTP API](https://grafana.com/docs/grafana/latest/http_api/licensing/) | 22 | [ZPascal](https://github.com/ZPascal) | | | +| [Other HTTP API](https://grafana.com/docs/grafana/latest/http_api/other/) | 22 | [ZPascal](https://github.com/ZPascal) | | | +| [Query History API](https://grafana.com/docs/grafana/latest/http_api/query_history/) | 22 | [ZPascal](https://github.com/ZPascal) | | | +| [Reporting API](https://grafana.com/docs/grafana/latest/http_api/reporting/) | 22 | [ZPascal](https://github.com/ZPascal) | | | ## Installation diff --git a/docs/content/grafana_api/model.md b/docs/content/grafana_api/model.md index 0a50b35..061cb7b 100644 --- a/docs/content/grafana_api/model.md +++ b/docs/content/grafana_api/model.md @@ -12,6 +12,9 @@ * [AlertmanagerReceivers](#grafana_api.model.AlertmanagerReceivers) * [RulerRule](#grafana_api.model.RulerRule) * [UserObject](#grafana_api.model.UserObject) + * [PlaylistObject](#grafana_api.model.PlaylistObject) + * [PlaylistItemObject](#grafana_api.model.PlaylistItemObject) + * [TeamObject](#grafana_api.model.TeamObject) @@ -207,3 +210,52 @@ The class includes all necessary variables to generate a User object that is nec - `login` _str_ - Specify the expression of the rule - `theme` _str_ - Specify the Grafana alert of the rule + + +## PlaylistObject Objects + +```python +class PlaylistObject(NamedTuple) +``` + +The class includes all necessary variables to generate a playlist object + +**Arguments**: + +- `name` _str_ - Specify the name of the playlist +- `interval` _str_ - Specify the interval of the playlist +- `items` _list_ - Specify a list of PlaylistItemObjects + + + +## PlaylistItemObject Objects + +```python +class PlaylistItemObject(NamedTuple) +``` + +The class includes all necessary variables to generate a playlist item object that is necessary to update a playlist + +**Arguments**: + +- `type` _str_ - Specify the type of the playlist item +- `value` _str_ - Specify the value of the playlist item +- `order` _int_ - Specify the order of the playlist item +- `title` _str_ - Specify the title of the playlist item + + + +## TeamObject Objects + +```python +class TeamObject(NamedTuple) +``` + +The class includes all necessary variables to generate a team object that is necessary to create a team + +**Arguments**: + +- `name` _str_ - Specify the name of the team +- `email` _str_ - Specify the email of the team +- `org_id` _int_ - Specify the org_id of the team + diff --git a/docs/content/index.md b/docs/content/index.md index 648f7e0..b0cdbfe 100644 --- a/docs/content/index.md +++ b/docs/content/index.md @@ -19,16 +19,16 @@ The following list describes the currently supported features of the Grafana API - [ ] [External Group Sync HTTP API](https://grafana.com/docs/grafana/latest/http_api/external_group_sync/) - [ ] [Fine-grained access control HTTP API](https://grafana.com/docs/grafana/latest/http_api/access_control/) - [ ] [HTTP Preferences API](https://grafana.com/docs/grafana/latest/http_api/preferences/) -- [ ] [HTTP Snapshot API](https://grafana.com/docs/grafana/latest/http_api/snapshot/) +- [x] [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/) +- [x] [Organization HTTP API](https://grafana.com/docs/grafana/latest/http_api/org/) - [ ] [Other HTTP API](https://grafana.com/docs/grafana/latest/http_api/other/) -- [ ] [Playlist HTTP API](https://grafana.com/docs/grafana/latest/http_api/playlist/) +- [x] [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/) -- [ ] [Team HTTP API](https://grafana.com/docs/grafana/latest/http_api/team/) -- [ ] [User HTTP API](https://grafana.com/docs/grafana/latest/http_api/user/) +- [x] [Short URL HTTP API](https://grafana.com/docs/grafana/latest/http_api/short_url/) +- [x] [Team HTTP API](https://grafana.com/docs/grafana/latest/http_api/team/) +- [x] [User HTTP API](https://grafana.com/docs/grafana/latest/http_api/user/) ## Installation diff --git a/docs/grafana_api_sdk.md b/docs/grafana_api_sdk.md index 648f7e0..b0cdbfe 100644 --- a/docs/grafana_api_sdk.md +++ b/docs/grafana_api_sdk.md @@ -19,16 +19,16 @@ The following list describes the currently supported features of the Grafana API - [ ] [External Group Sync HTTP API](https://grafana.com/docs/grafana/latest/http_api/external_group_sync/) - [ ] [Fine-grained access control HTTP API](https://grafana.com/docs/grafana/latest/http_api/access_control/) - [ ] [HTTP Preferences API](https://grafana.com/docs/grafana/latest/http_api/preferences/) -- [ ] [HTTP Snapshot API](https://grafana.com/docs/grafana/latest/http_api/snapshot/) +- [x] [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/) +- [x] [Organization HTTP API](https://grafana.com/docs/grafana/latest/http_api/org/) - [ ] [Other HTTP API](https://grafana.com/docs/grafana/latest/http_api/other/) -- [ ] [Playlist HTTP API](https://grafana.com/docs/grafana/latest/http_api/playlist/) +- [x] [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/) -- [ ] [Team HTTP API](https://grafana.com/docs/grafana/latest/http_api/team/) -- [ ] [User HTTP API](https://grafana.com/docs/grafana/latest/http_api/user/) +- [x] [Short URL HTTP API](https://grafana.com/docs/grafana/latest/http_api/short_url/) +- [x] [Team HTTP API](https://grafana.com/docs/grafana/latest/http_api/team/) +- [x] [User HTTP API](https://grafana.com/docs/grafana/latest/http_api/user/) ## Installation diff --git a/setup.py b/setup.py index 0b487fa..46c4e82 100644 --- a/setup.py +++ b/setup.py @@ -8,7 +8,7 @@ setuptools.setup( name="grafana-api-sdk", - version="0.0.4", + version="0.0.5", author="Pascal Zimmermann", author_email="info@theiotstudio.com", description="A Grafana API SDK", diff --git a/src/grafana_api/api.py b/src/grafana_api/api.py index 6f574f4..179e9bd 100644 --- a/src/grafana_api/api.py +++ b/src/grafana_api/api.py @@ -1,4 +1,5 @@ import logging + import requests from .model import RequestsMethods, ERROR_MESSAGES, APIModel @@ -107,7 +108,7 @@ def __check_the_api_call_response(response: any = None) -> any: api_call (any): Returns the value of the api call """ - if type(response.json()) == dict: + if len(response.text) != 0 and type(response.json()) == dict: if ( "message" in response.json().keys() and response.json()["message"] in ERROR_MESSAGES diff --git a/src/grafana_api/model.py b/src/grafana_api/model.py index ba58a83..4ff72f4 100644 --- a/src/grafana_api/model.py +++ b/src/grafana_api/model.py @@ -29,6 +29,8 @@ class APIEndpoints(Enum): USERS = f"{api_prefix}/users" SNAPSHOTS = f"{api_prefix}/snapshots" DASHBOARD_SNAPSHOTS = f"{api_prefix}/dashboard/snapshots" + PLAYLISTS = f"{api_prefix}/playlists" + TEAMS = f"{api_prefix}/teams" class RequestsMethods(Enum): @@ -218,3 +220,47 @@ class UserObject(NamedTuple): name: str login: str theme: str + + +class PlaylistObject(NamedTuple): + """The class includes all necessary variables to generate a playlist object + + Args: + name (str): Specify the name of the playlist + interval (str): Specify the interval of the playlist + items (list): Specify a list of PlaylistItemObjects + """ + + name: str + interval: str + items: list + + +class PlaylistItemObject(NamedTuple): + """The class includes all necessary variables to generate a playlist item object that is necessary to update a playlist + + Args: + type (str): Specify the type of the playlist item + value (str): Specify the value of the playlist item + order (int): Specify the order of the playlist item + title (str): Specify the title of the playlist item + """ + + type: str + value: str + order: int + title: str + + +class TeamObject(NamedTuple): + """The class includes all necessary variables to generate a team object that is necessary to create a team + + Args: + name (str): Specify the name of the team + email (str): Specify the email of the team + org_id (int): Specify the org_id of the team + """ + + name: str + email: str + org_id: int diff --git a/src/grafana_api/playlist.py b/src/grafana_api/playlist.py new file mode 100644 index 0000000..caac160 --- /dev/null +++ b/src/grafana_api/playlist.py @@ -0,0 +1,299 @@ +import json +import logging + +from requests import Response + +from .model import ( + APIModel, + APIEndpoints, + RequestsMethods, + PlaylistObject, +) +from .api import Api + + +class Playlist: + """The class includes all necessary methods to access the Grafana playlist API endpoints + + 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 search_playlist(self, query: str = None, limit: int = None) -> list: + """The method includes a functionality to get the organization playlist's specified by the optional pagination functionality + + Args: + query (str): Specify the query to limit response to playlist having a name like this value(default None) + limit (int): Specify the limit as integer of the response (default None) + + Raises: + Exception: Unspecified error by executing the API call + + Returns: + api_call (list): Returns the organization playlist's + """ + + api_request_url: str = APIEndpoints.PLAYLISTS.value + + if query is not None: + api_request_url: str = f"{api_request_url}?query={query}" + + if limit is not None and limit != 0: + api_request_url: str = f"{api_request_url}?limit={limit}" + + if query is not None and (limit is not None or limit != 0): + api_request_url: str = f"{api_request_url}?query={query}&limit={limit}" + + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api( + api_request_url, + ) + .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 get_playlist(self, playlist_id: int) -> dict: + """The method includes a functionality to get the playlist specified by the playlist_id + + Args: + playlist_id (int): Specify the playlist_id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the corresponding playlist + """ + + if playlist_id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.PLAYLISTS.value}/{playlist_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 playlist_id defined.") + raise ValueError + + def get_playlist_items(self, playlist_id: int) -> list: + """The method includes a functionality to get the playlist items specified by the playlist_id + + Args: + playlist_id (int): Specify the playlist_id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the corresponding playlist items + """ + + if playlist_id != 0: + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.PLAYLISTS.value}/{playlist_id}/items", + ) + .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 + else: + logging.error("There is no playlist_id defined.") + raise ValueError + + def get_playlist_dashboards(self, playlist_id: int) -> list: + """The method includes a functionality to get the playlist dashboards specified by the playlist_id + + Args: + playlist_id (int): Specify the playlist_id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the corresponding playlist dashboards + """ + + if playlist_id != 0: + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.PLAYLISTS.value}/{playlist_id}/dashboards", + ) + .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 + else: + logging.error("There is no playlist_id defined.") + raise ValueError + + def create_playlist(self, playlist: PlaylistObject) -> dict: + """The method includes a functionality to create a playlist specified by the playlist object + + Args: + playlist (PlaylistObject): Specify the playlist object + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the corresponding playlist + """ + + if playlist is not None: + items: list = list() + + for item in playlist.items: + items.append( + { + "type": item.type, + "value": item.value, + "order": item.order, + "title": item.title, + } + ) + + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.PLAYLISTS.value}", + RequestsMethods.POST, + json.dumps( + dict( + { + "name": playlist.name, + "interval": playlist.interval, + "items": items, + } + ) + ), + ) + .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 playlist object defined.") + raise ValueError + + def update_playlist(self, playlist_id: int, playlist: PlaylistObject) -> dict: + """The method includes a functionality to update a playlist specified by the playlist object and playlist_id + + Args: + playlist_id (int): Specify the playlist_id + playlist (PlaylistObject): Specify the playlist object + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the corresponding playlist + """ + + if playlist_id != 0 and playlist is not None: + items: list = list() + + for item in playlist.items: + items.append( + { + "type": item.type, + "value": item.value, + "order": item.order, + "title": item.title, + } + ) + + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.PLAYLISTS.value}/{playlist_id}", + RequestsMethods.PUT, + json.dumps( + dict( + { + "name": playlist.name, + "interval": playlist.interval, + "items": items, + } + ) + ), + ) + .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 playlist_id or playlist object defined.") + raise ValueError + + def delete_playlist(self, playlist_id: int): + """The method includes a functionality to delete a playlist specified by the playlist_id + + Args: + playlist_id (int): Specify the playlist_id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if playlist_id != 0: + api_call: Response = Api(self.grafana_api_model).call_the_api( + f"{APIEndpoints.PLAYLISTS.value}/{playlist_id}", + RequestsMethods.DELETE, + ) + + if api_call.status_code != 200: + logging.error(f"Check the error: {api_call.text}.") + raise Exception + else: + logging.info("You successfully deleted the corresponding playlist.") + else: + logging.error("There is no playlist_id object defined.") + raise ValueError diff --git a/src/grafana_api/team.py b/src/grafana_api/team.py new file mode 100644 index 0000000..f36d226 --- /dev/null +++ b/src/grafana_api/team.py @@ -0,0 +1,426 @@ +import json +import logging + +from .model import ( + APIModel, + APIEndpoints, + RequestsMethods, + TeamObject, +) +from .api import Api + + +class Team: + """The class includes all necessary methods to access the Grafana team API endpoints. Be aware that all functionalities inside the class only working with the corresponding access rights, please check the following page for details: https://grafana.com/docs/grafana/latest/http_api/team/#team-api + + HINT: Note Grafana Enterprise API need required permissions if fine-grained access control is enabled + + 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 search_team( + self, results_per_page: int = 1000, pages: int = 1, query: str = None + ) -> dict: + """The method includes a functionality to get the organization teams specified by the optional pagination functionality + + Required Permissions: + Action: teams:read + Scope: teams:* + + Args: + results_per_page (int): Specify the results_per_page as integer (default 1000) + pages (int): Specify the pages as integer (default 1) + query (str): Specify the query (default None) + + Raises: + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the organization teams + """ + + api_request_url: str = ( + f"{APIEndpoints.TEAMS.value}/search?perpage={results_per_page}&page={pages}" + ) + + if query is not None and len(query) != 0: + api_request_url: str = f"{api_request_url}&query={query}" + + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + api_request_url, + ) + .json() + ) + + if api_call == dict() or api_call.get("totalCount") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + + def get_team_by_id(self, id: int) -> dict: + """The method includes a functionality to get the organization team specified by the id + + Required Permissions: + Action: teams:read + Scope: teams:* + + Args: + id (int): Specify the id of the team + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the organization team + """ + + if id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{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 id defined.") + raise ValueError + + def add_team(self, team: TeamObject) -> int: + """The method includes a functionality to add an organization team specified by the TeamObject + + Required Permissions: + Action: teams:create + Scope: N/A + + Args: + team (TeamObject): Specify the team + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + team_id (int): Returns the team id + """ + + if team is not None: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}", + RequestsMethods.POST, + json.dumps(dict({"name": team.name, "email": team.name})), + ) + .json() + ) + + if api_call.get("message") != "Team created": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return int(api_call.get("teamId")) + else: + logging.error("There is no team defined.") + raise ValueError + + def update_team(self, id: int, name: str, email: str): + """The method includes a functionality to update an organization team specified by the team_id, name and email + + Required Permissions: + Action: teams:write + Scope: teams:* + + Args: + id (int): Specify the team id + name (str): Specify the team name + email (str): Specify the team email + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if id != 0 and len(name) != 0 and len(email) != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{id}", + RequestsMethods.PUT, + json.dumps(dict({"name": name, "email": email})), + ) + .json() + ) + + if api_call.get("message") != "Team updated": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully updated the team.") + else: + logging.error("There is no id, name or email defined.") + raise ValueError + + def delete_team_by_id(self, id: int): + """The method includes a functionality to delete an organization team specified by the team_id + + Required Permissions: + Action: teams:delete + Scope: teams:* + + Args: + id (int): Specify the team id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{id}", + RequestsMethods.DELETE, + ) + .json() + ) + + if api_call.get("message") != "Team deleted": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully deleted the team.") + else: + logging.error("There is no id defined.") + raise ValueError + + def get_team_members(self, id: int) -> list: + """The method includes a functionality to get all organization team users specified by the team_id + + Required Permissions: + Action: teams.permissions:read + Scope: teams:* + + Args: + id (int): Specify the team id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (list): Returns the organization team members + """ + + if id != 0: + api_call: list = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{id}/members", + ) + .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 + else: + logging.error("There is no id defined.") + raise ValueError + + def add_team_member(self, id: int, user_id: int): + """The method includes a functionality to add an organization team user specified by the team_id and the user_id + + Required Permissions: + Action: teams.permissions:write + Scope: teams:* + + Args: + id (int): Specify the team id + user_id (int): Specify the user id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if id != 0 and user_id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{id}/members", + RequestsMethods.POST, + json.dumps(dict({"userId": user_id})), + ) + .json() + ) + + if api_call.get("message") != "Member added to Team": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully added a team member.") + else: + logging.error("There is no id or user_id defined.") + raise ValueError + + def delete_team_member(self, id: int, user_id: int): + """The method includes a functionality to delete an organization team user specified by the team_id and the user_id + + Required Permissions: + Action: teams.permissions:write + Scope: teams:* + + Args: + id (int): Specify the team id + user_id (int): Specify the user id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if id != 0 and user_id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{id}/members/{user_id}", + RequestsMethods.DELETE, + ) + .json() + ) + + if api_call.get("message") != "Team Member removed": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully removed a team member.") + else: + logging.error("There is no id or user_id defined.") + raise ValueError + + def get_team_preferences(self, id: int) -> dict: + """The method includes a functionality to get the organization team preferences specified by the team_id + + Required Permissions: + Action: teams:read + Scope: teams:* + + Args: + id (int): Specify the team id + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + api_call (dict): Returns the organization team preferences + """ + + if id != 0: + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{id}/preferences", + ) + .json() + ) + + if api_call == dict() or api_call.get("homeDashboardId") is None: + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.error("There is no id defined.") + raise ValueError + + def update_team_preferences( + self, id: int, theme: str = "", timezone: str = "", home_dashboard_id: int = 0 + ): + """The method includes a functionality to update the organization team preferences specified by the team_id, theme, home_dashboard_id and timezone + + Required Permissions: + Action: teams:write + Scope: teams:* + + Args: + id (int): Specify the team id + theme (str): Specify the team theme e.g. light or dark (default Grafana theme) + timezone (str): Specify the team timezone e.g. utc or browser (default Grafana timezone) + home_dashboard_id (int): Specify the home team dashboard by id (default 0) + + Raises: + ValueError: Missed specifying a necessary value + Exception: Unspecified error by executing the API call + + Returns: + None + """ + + if ( + id != 0 + and theme is not None + and home_dashboard_id is not None + and timezone is not None + ): + api_call: dict = ( + Api(self.grafana_api_model) + .call_the_api( + f"{APIEndpoints.TEAMS.value}/{id}/preferences", + RequestsMethods.PUT, + json.dumps( + dict( + { + "theme": theme, + "homeDashboardId": home_dashboard_id, + "timezone": timezone, + } + ) + ), + ) + .json() + ) + + if api_call.get("message") != "Preferences updated": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully updated the team preferences.") + else: + logging.error( + "There is no id, theme, home_dashboard_id or timezone defined." + ) + raise ValueError diff --git a/src/grafana_api/user.py b/src/grafana_api/user.py index 2531758..067be40 100644 --- a/src/grafana_api/user.py +++ b/src/grafana_api/user.py @@ -50,7 +50,7 @@ def search_users( """ api_request_url: str = ( - f"{APIEndpoints.USERS.value}?perpage={results_per_page}&page={pages}" + f"{APIEndpoints.USERS.value}/search?perpage={results_per_page}&page={pages}" ) if query is not None and len(query) != 0: diff --git a/tests/integrationtest/test_alerting.py b/tests/integrationtest/test_alerting.py index a5f047e..bd3cbdd 100644 --- a/tests/integrationtest/test_alerting.py +++ b/tests/integrationtest/test_alerting.py @@ -119,10 +119,10 @@ def test_get_prometheus_alerts(self): "data": { "alerts": [ { - "labels": {'alertname': 'Test'}, + "labels": {"alertname": "Test"}, "annotations": {}, "state": "Normal", - "activeAt": "0001-01-01T00:00:00Z", + "activeAt": "0001-01-01T00:53:28+00:53", "value": "", } ] diff --git a/tests/integrationtest/test_playlist.py b/tests/integrationtest/test_playlist.py new file mode 100644 index 0000000..6f759ac --- /dev/null +++ b/tests/integrationtest/test_playlist.py @@ -0,0 +1,61 @@ +import os +from unittest import TestCase + +from src.grafana_api.model import APIModel, PlaylistObject, PlaylistItemObject +from src.grafana_api.playlist import Playlist + + +class PlaylistTest(TestCase): + model: APIModel = APIModel( + host=os.environ["GRAFANA_HOST"], + token=os.environ["GRAFANA_TOKEN"], + ) + playlist: Playlist = Playlist(model) + + def test_search_playlist(self): + self.assertEqual("Test", self.playlist.search_playlist()[0].get("name")) + + def test_get_playlist(self): + self.assertEqual("Test", self.playlist.get_playlist(1).get("name")) + + def test_get_playlist_items(self): + print(self.playlist.get_playlist_items(1)[0]) + self.assertEqual("152", self.playlist.get_playlist_items(1)[0].get("value")) + + def test_get_playlist_dashboards(self): + self.assertEqual(152, self.playlist.get_playlist_dashboards(1)[0].get("id")) + + def test_a_create_playlist(self): + playlist_item: PlaylistItemObject = PlaylistItemObject( + type="dashboard_by_id", + value="152", + order=1, + title="Github Integrationtest/Test 1", + ) + playlist_object: PlaylistObject = PlaylistObject( + "Test1", "5m", list([playlist_item]) + ) + self.playlist.create_playlist(playlist_object) + + self.assertEqual("Test1", self.playlist.search_playlist()[1].get("name")) + + def test_b_update_playlist(self): + playlist_item: PlaylistItemObject = PlaylistItemObject( + type="dashboard_by_id", + value="152", + order=1, + title="Github Integrationtest/Test 1", + ) + playlist_object: PlaylistObject = PlaylistObject( + "Test2", "5m", list([playlist_item]) + ) + self.playlist.update_playlist( + self.playlist.search_playlist()[1].get("id"), playlist_object + ) + + self.assertEqual("Test2", self.playlist.search_playlist()[1].get("name")) + + def test_c_delete_playlist(self): + self.playlist.delete_playlist(self.playlist.search_playlist()[1].get("id")) + + self.assertEqual(1, len(self.playlist.search_playlist())) diff --git a/tests/integrationtest/test_team.py b/tests/integrationtest/test_team.py new file mode 100644 index 0000000..73ab2e7 --- /dev/null +++ b/tests/integrationtest/test_team.py @@ -0,0 +1,95 @@ +import os +from unittest import TestCase + +from src.grafana_api.model import APIModel, TeamObject +from src.grafana_api.team import Team +from src.grafana_api.organisation import Organisation + + +class TeamTest(TestCase): + model: APIModel = APIModel( + host=os.environ["GRAFANA_HOST"], + token=os.environ["GRAFANA_TOKEN"], + ) + team: Team = Team(model) + + def test_search_team(self): + self.assertEqual( + 4, self.team.search_team(query="Test").get("teams")[0].get("id") + ) + + def test_get_team_by_id(self): + self.assertEqual(4, self.team.get_team_by_id(4).get("id")) + + def test_a_add_team(self): + team_object: TeamObject = TeamObject( + name="Test1", + email="test@test.de", + org_id=Organisation(self.model).get_current_organization().get("id"), + ) + self.assertIsNotNone(self.team.add_team(team_object)) + + self.assertIsNotNone( + self.team.search_team(query="Test1").get("teams")[0].get("id") + ) + + def test_b_update_team(self): + self.assertIsNone( + self.team.update_team( + self.team.search_team(query="Test1").get("teams")[0].get("id"), + "Test2", + "test2@test.de", + ) + ) + + self.assertIsNotNone( + self.team.search_team(query="Test2").get("teams")[0].get("id") + ) + + def test_c_delete_team_by_id(self): + self.assertIsNone( + self.team.delete_team_by_id( + self.team.search_team(query="Test2").get("teams")[0].get("id") + ) + ) + + with self.assertRaises(Exception): + self.assertIsNotNone( + self.team.search_team(query="Test2").get("teams")[0].get("id") + ) + + def test_get_team_members(self): + self.assertEqual(1, len(self.team.get_team_members(4))) + + def test_d_add_team_member(self): + team_id: int = self.__team_creation_util() + + self.assertEqual(1, len(self.team.get_team_members(team_id))) + + self.team.delete_team_by_id(team_id) + + def test_e_delete_team_member(self): + team_id: int = self.__team_creation_util() + + self.team.delete_team_member(team_id, 8) + + with self.assertRaises(Exception): + self.team.get_team_members(team_id) + + self.team.delete_team_by_id(team_id) + + def test_get_team_preferences(self): + self.assertEqual(0, self.team.get_team_preferences(4).get("homeDashboardId")) + + def test_update_team_preferences(self): + self.assertIsNone(self.team.update_team_preferences(4, timezone="utc")) + + self.assertEqual("utc", self.team.get_team_preferences(4).get("timezone")) + + def __team_creation_util(self) -> int: + self.test_a_add_team() + team_id: int = self.team.search_team(query="Test1").get("teams")[0].get("id") + + self.team.add_team_member(team_id, 8) + + return team_id diff --git a/tests/integrationtest/test_user.py b/tests/integrationtest/test_user.py index 2c7ab0e..d24cbfc 100644 --- a/tests/integrationtest/test_user.py +++ b/tests/integrationtest/test_user.py @@ -22,8 +22,7 @@ def test_get_user_organizations(self): self.assertEqual(1, len(self.current_user.get_user_organizations())) def test_get_user_teams(self): - with self.assertRaises(Exception): - self.current_user.get_user_teams() + self.assertEqual(4, self.current_user.get_user_teams()[0].get("id")) def test_star_a_dashboard(self): self.current_user.star_a_dashboard( diff --git a/tests/unittests/test_api.py b/tests/unittests/test_api.py index ccd314a..a3d38ec 100644 --- a/tests/unittests/test_api.py +++ b/tests/unittests/test_api.py @@ -30,6 +30,7 @@ def test_call_the_api_basic_auth(self, get_mock): mock: Mock = Mock() mock.json = Mock(return_value={"status": "success"}) + mock.text = str({"status": "success"}) get_mock.return_value = mock @@ -42,6 +43,7 @@ def test_call_the_api_basic_auth(self, get_mock): def test_call_the_api_get_valid(self, get_mock): mock: Mock = Mock() mock.json = Mock(return_value={"status": "success"}) + mock.text = str({"status": "success"}) get_mock.return_value = mock @@ -58,6 +60,7 @@ def test_call_the_api_get_not_valid(self): def test_call_the_api_put_valid(self, put_mock): mock: Mock = Mock() mock.json = Mock(return_value={"status": "success"}) + mock.text = str({"status": "success"}) put_mock.return_value = mock @@ -78,6 +81,7 @@ def test_call_the_api_put_not_valid(self): def test_call_the_api_post_valid(self, post_mock): mock: Mock = Mock() mock.json = Mock(return_value={"status": "success"}) + mock.text = str({"status": "success"}) post_mock.return_value = mock @@ -106,6 +110,7 @@ def test_call_the_api_post_no_data(self): def test_call_the_api_patch_valid(self, post_mock): mock: Mock = Mock() mock.json = Mock(return_value={"status": "success"}) + mock.text = str({"status": "success"}) post_mock.return_value = mock @@ -134,6 +139,7 @@ def test_call_the_api_patch_no_data(self): def test_call_the_api_delete_valid(self, delete_mock): mock: Mock = Mock() mock.json = Mock(return_value={"message": "Deletion successful"}) + mock.text = str({"message": "Deletion successful"}) delete_mock.return_value = mock @@ -151,6 +157,7 @@ def test_call_the_api_delete_not_valid(self): def test_check_the_api_call_response(self): mock: Mock = Mock() mock.json = Mock(return_value=dict({"test": "test"})) + mock.text = str({"test": "test"}) self.assertEqual( dict({"test": "test"}), @@ -160,6 +167,7 @@ def test_check_the_api_call_response(self): def test_check_the_api_call_response_no_error_message(self): mock: Mock = Mock() mock.json = Mock(return_value=dict({"message": "test"})) + mock.text = str({"message": "test"}) self.assertEqual( dict({"message": "test"}), @@ -168,15 +176,16 @@ def test_check_the_api_call_response_no_error_message(self): def test_check_the_api_call_response_no_json_response_value(self): mock: Mock = Mock() - mock.text = Mock(return_value="test") + mock.text = "test" self.assertEqual( - "test", self.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): mock: Mock = Mock() mock.json = Mock(return_value=dict({"message": "invalid API key"})) + mock.text = str({"message": "invalid API key"}) with self.assertRaises(requests.exceptions.ConnectionError): self.api._Api__check_the_api_call_response(response=mock) diff --git a/tests/unittests/test_playlist.py b/tests/unittests/test_playlist.py new file mode 100644 index 0000000..357b85b --- /dev/null +++ b/tests/unittests/test_playlist.py @@ -0,0 +1,326 @@ +from unittest import TestCase +from unittest.mock import MagicMock, Mock, patch + +from src.grafana_api.model import APIModel, PlaylistObject, PlaylistItemObject +from src.grafana_api.playlist import Playlist + + +class PlaylistTestCase(TestCase): + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_playlist(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list([{"id": 1}])) + + call_the_api_mock.return_value = mock + + self.assertEqual(list([{"id": 1}]), playlist.search_playlist()) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_playlist_no_valid_result(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.search_playlist() + + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_playlist_query_defined(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.search_playlist(query="Test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_playlist_limit_defined(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.search_playlist(limit=1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_playlist_query_and_limit_defined(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.search_playlist(query="Test", limit=1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"id": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual(dict({"id": 1}), playlist.get_playlist(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_no_playlist_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + playlist.get_playlist(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_no_valid_result(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.get_playlist(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_items(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list([{"id": 1}])) + + call_the_api_mock.return_value = mock + + self.assertEqual(list([{"id": 1}]), playlist.get_playlist_items(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_items_no_playlist_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + playlist.get_playlist_items(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_items_no_valid_result(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.get_playlist_items(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_dashboards(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list([{"id": 1}])) + + call_the_api_mock.return_value = mock + + self.assertEqual(list([{"id": 1}]), playlist.get_playlist_dashboards(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_dashboards_no_playlist_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + playlist.get_playlist_dashboards(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_playlist_dashboards_no_valid_result(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.get_playlist_dashboards(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_playlist(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + playlist_items: PlaylistItemObject = PlaylistItemObject( + type="test", value="test", order=1, title="test" + ) + playlist_object: PlaylistObject = PlaylistObject( + "test", "5m", list([playlist_items]) + ) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"id": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual(dict({"id": 1}), playlist.create_playlist(playlist_object)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_playlist_no_playlist_object(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + playlist.create_playlist(None) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_create_playlist_no_valid_result(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + playlist_items: PlaylistItemObject = PlaylistItemObject( + type="test", value="test", order=1, title="test" + ) + playlist_object: PlaylistObject = PlaylistObject( + "test", "5m", list([playlist_items]) + ) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.create_playlist(playlist_object) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_playlist(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + playlist_items: PlaylistItemObject = PlaylistItemObject( + type="test", value="test", order=1, title="test" + ) + playlist_object: PlaylistObject = PlaylistObject( + "test", "5m", list([playlist_items]) + ) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"id": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual(dict({"id": 1}), playlist.update_playlist(1, playlist_object)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_playlist_no_playlist_object(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + playlist: Playlist = Playlist(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + playlist.update_playlist(1, None) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_playlist_no_valid_result(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + playlist_items: PlaylistItemObject = PlaylistItemObject( + type="test", value="test", order=1, title="test" + ) + playlist_object: PlaylistObject = PlaylistObject( + "test", "5m", list([playlist_items]) + ) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + playlist.update_playlist(1, playlist_object) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_playlist(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + call_the_api_mock.return_value.status_code = 200 + + self.assertEqual(None, playlist.delete_playlist(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_playlist_no_playlist_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + playlist: Playlist = Playlist(grafana_api_model=model) + + call_the_api_mock.return_value.status_code = 400 + + with self.assertRaises(ValueError): + playlist.delete_playlist(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_playlist_no_valid_result(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + playlist: Playlist = Playlist(grafana_api_model=model) + + call_the_api_mock.return_value.status_code = 400 + + with self.assertRaises(Exception): + playlist.delete_playlist(1) diff --git a/tests/unittests/test_team.py b/tests/unittests/test_team.py new file mode 100644 index 0000000..959e5d8 --- /dev/null +++ b/tests/unittests/test_team.py @@ -0,0 +1,458 @@ +from unittest import TestCase +from unittest.mock import MagicMock, Mock, patch + +from src.grafana_api.model import APIModel, TeamObject +from src.grafana_api.team import Team + + +class TeamTestCase(TestCase): + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_team(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"totalCount": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual(dict({"totalCount": 1}), team.search_team()) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_team_query(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"totalCount": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual( + dict({"totalCount": 1}), + team.search_team(query="Test"), + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_search_team_no_team(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.search_team() + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_by_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"id": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual(dict({"id": 1}), team.get_team_by_id(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_by_id_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.get_team_by_id(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_by_id_no_team(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.get_team_by_id(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_team(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + team_object: TeamObject = TeamObject(name="test", email="test", org_id=1) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Team created", "teamId": 1})) + + call_the_api_mock.return_value = mock + + self.assertEqual(1, team.add_team(team_object)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_team_no_team(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.add_team(None) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_team_add_not_possible(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + team_object: TeamObject = TeamObject(name="test", email="test", org_id=1) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Test"})) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.add_team(team_object) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_team(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Team updated"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, team.update_team(1, "test", "test")) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_team_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.update_team(0, None, None) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_team_update_not_possible(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Test"})) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.update_team(1, "test", "test") + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_team_by_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Team deleted"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, team.delete_team_by_id(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_team_by_id_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.delete_team_by_id(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_team_by_id_delete_not_possible(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Test"})) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.delete_team_by_id(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_members(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list([{"userId": 1}])) + + call_the_api_mock.return_value = mock + + self.assertEqual(list([{"userId": 1}]), team.get_team_members(1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_members_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.get_team_members(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_members_no_team_members(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=list()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.get_team_members(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_team_member(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Member added to Team"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, team.add_team_member(1, 1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_team_member_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.add_team_member(0, 0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_add_team_member_add_not_possible(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Test"})) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.add_team_member(1, 1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_team_member(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Team Member removed"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, team.delete_team_member(1, 1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_team_member_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.delete_team_member(0, 0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_delete_team_member_deletion_not_possible(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Test"})) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.delete_team_member(1, 1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_preferences(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock( + return_value=dict( + {"theme": "light", "homeDashboardId": "Test", "timezone": "utc"} + ) + ) + + call_the_api_mock.return_value = mock + + self.assertEqual( + dict({"theme": "light", "homeDashboardId": "Test", "timezone": "utc"}), + team.get_team_preferences(1), + ) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_preferences_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.get_team_preferences(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_get_team_preferences_no_preferences_available(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.get_team_preferences(1) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_team_preferences(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Preferences updated"})) + + call_the_api_mock.return_value = mock + + self.assertEqual(None, team.update_team_preferences(1, 1)) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_team_preferences_no_id(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict()) + + call_the_api_mock.return_value = mock + + with self.assertRaises(ValueError): + team.update_team_preferences(0) + + @patch("src.grafana_api.api.Api.call_the_api") + def test_update_team_preferences_update_not_possible(self, call_the_api_mock): + model: APIModel = APIModel( + host=MagicMock(), username=MagicMock(), password=MagicMock() + ) + team: Team = Team(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value=dict({"message": "Test"})) + + call_the_api_mock.return_value = mock + + with self.assertRaises(Exception): + team.update_team_preferences(1)