diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..954413c --- /dev/null +++ b/.coveragerc @@ -0,0 +1,22 @@ +[run] +branch = True +omit = + */setup.py +source = . + +[report] +omit = + */setup.py +fail_under = 80 +show_missing = True +skip_covered = False +exclude_lines = + # Have to re-enable the standard pragma + pragma: no cover + + # Don't complain if tests don't hit defensive assertion code: + raise AssertionError + raise NotImplementedError + + # Don't complain if non-runnable code isn't run: + if __name__ == .__main__.: diff --git a/.flake8 b/.flake8 new file mode 100644 index 0000000..e87e4f3 --- /dev/null +++ b/.flake8 @@ -0,0 +1,6 @@ +[flake8] +ignore = E203, E266, E501, W503 +max-line-length = 120 +max-complexity = 20 +select = B,C,D,E,F,W,T4,B902,B950 +exclude = venc,.git \ No newline at end of file diff --git a/.github/workflows/integrationtest.yml b/.github/workflows/integrationtest.yml new file mode 100644 index 0000000..11ff708 --- /dev/null +++ b/.github/workflows/integrationtest.yml @@ -0,0 +1,26 @@ +name: Integrationtest + +on: + push: + branches: [ main ] + pull_request: + branches: [ main ] + +jobs: + + test: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v2 + + - name: Install the requirements + run: pip install -r requirements.txt + + - name: Execute the integrationtests + run: python3 -m unittest discover tests/integrationtest + env: + GRAFANA_HOST: ${{ secrets.GRAFANA_HOST }} + GRAFANA_TOKEN: ${{ secrets.GRAFANA_TOKEN }} + GRAFANA_DASHBOARD_PATH: ${{ secrets.GRAFANA_DASHBOARD_PATH }} + GRAFANA_DASHBOARD_NAME: ${{ secrets.GRAFANA_DASHBOARD_NAME }} \ No newline at end of file diff --git a/.github/workflows/publish-to-pypi.yml b/.github/workflows/publish-to-pypi.yml new file mode 100644 index 0000000..aaba681 --- /dev/null +++ b/.github/workflows/publish-to-pypi.yml @@ -0,0 +1,48 @@ +name: Build and publish + +on: + release: + types: [ published ] + +jobs: + + build-and-publish: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ['3.x'] + + steps: + - name: Checkout the repository and the branch + uses: actions/checkout@v2 + + - name: Set up Python + uses: actions/setup-python@v2 + with: + python-version: ${{ matrix.python-version }} + architecture: x64 + cache: 'pip' + + - name: Install the requirements + run: pip install -r requirements.txt + + - name: Install pypa/build + run: >- + python -m + pip install + build + --user + + - name: Build a binary wheel and a source tarball + run: >- + python -m + build + --sdist + --wheel + --outdir dist/ + + - name: Publish distribution package to PyPI + if: startsWith(github.ref, 'refs/tags') + uses: pypa/gh-action-pypi-publish@master + with: + password: ${{ secrets.PYPI_API_TOKEN }} \ No newline at end of file diff --git a/.github/workflows/pull-request-checks.yml b/.github/workflows/pull-request-checks.yml new file mode 100644 index 0000000..8fd1726 --- /dev/null +++ b/.github/workflows/pull-request-checks.yml @@ -0,0 +1,114 @@ +name: PR checks + +on: + pull_request: + branches: [ main ] + +jobs: + + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: [ '3.x' ] + + steps: + - uses: actions/checkout@v2 + + - name: Set up Python + uses: actions/setup-python@v2 + with: + python-version: ${{ matrix.python-version }} + architecture: x64 + cache: 'pip' + + - name: Install the requirements + run: pip install -r requirements.txt + + - name: Execute the unittests + run: python3 -m unittest discover tests/unittests + + lint: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: [ '3.x' ] + + steps: + - uses: actions/checkout@v2 + + - name: Set up Python + uses: actions/setup-python@v2 + with: + python-version: ${{ matrix.python-version }} + architecture: x64 + cache: 'pip' + + - name: Install the requirements + run: pip install -r requirements.txt + + - name: Execute the linting checks + uses: reviewdog/action-flake8@v3.2.0 + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + flake8_args: --config=.flake8 + + coverage: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: [ '3.x' ] + + steps: + - uses: actions/checkout@v2 + with: + persist-credentials: false + fetch-depth: 0 + + - name: Set up Python + uses: actions/setup-python@v2 + with: + python-version: ${{ matrix.python-version }} + architecture: x64 + cache: 'pip' + + - name: Install the requirements + run: pip install -r requirements.txt && pip install pytest pytest-cov coverage-badge + + - name: Generate the coverage report + run: export PYTHONPATH=$PWD && pytest --junitxml=pytest.xml --cov=. tests/unittests | tee pytest-coverage.txt + + - name: Execute the coverage checks + uses: MishaKav/pytest-coverage-comment@v1.1.16 + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + pytest-coverage-path: ./pytest-coverage.txt + junitxml-path: ./pytest.xml + hide-badge: true + create-new-commit: true + + - name: "Check if coverage badge file existence" + id: check_files + uses: andstor/file-existence-action@v1 + with: + files: "docs/coverage.svg" + + - name: Generate coverage badge + if: steps.check_files.outputs.files_exists == 'false' + run: coverage-badge -o docs/coverage.svg -f + + - name: Commit files + if: steps.check_files.outputs.files_exists == 'false' + run: | + git config --local user.email "github-actions[bot]@users.noreply.github.com" + git config --local user.name "github-actions[bot]" + git add --force docs/coverage.svg + git commit -m "Add coverage badge" + + - name: Push changes + if: steps.check_files.outputs.files_exists == 'false' + uses: ad-m/github-push-action@master + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + branch: ${{ github.head_ref }} + force: true \ No newline at end of file diff --git a/README.md b/README.md index dbe3a41..2c52760 100644 --- a/README.md +++ b/README.md @@ -1,26 +1,84 @@ -# grafana_api_sdk -The repository includes an SDK for the Grafana API - -## TODO -- Documentation -- Docstrings -- PYPI support +# Grafana API SDK ![Coverage report](https://github.com/ZPascal/grafana_api_sdk/blob/main/docs/coverage.svg) +The repository includes an SDK for the Grafana API. It's possible to communicate with the Grafana API endpoints. Another feature of the SDK is the possibility to specify the used folder for the dashboard. ## Currently, supported features -- Get a Dashboard by uid -- Get folder id by dashboard path -- Get all folder ids and folder names -- Specify the grafana dashboard folder +### Dashboard - Create/ Update a dashboard - Delete a dashboard +- Get permissions of a dashboard +- Update the permissions of a dashboard +- Get all dashboard versions +- Get dashboard version of a specific dashboard +- Restore a dashboard version of a specific dashboard +- Compare two dashboard versions and extract the diff between booth dashboards + +### Folder +- Get folder id by dashboard path +- Get all folder ids and folder names +- Get all folders +- Get folder by uid +- Get folder by id +- Create a folder +- Update a folder +- Delete a folder +- Get permissions for a folder +- Update permissions for a folder + +### Search +- Execute a custom query against the Grafana search endpoint + +## 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/) | | | | | +| [Alerting HTTP API](https://grafana.com/docs/grafana/latest/http_api/alerting/) | 4 | [ZPascal](https://github.com/ZPascal) | | Planned | +| [Alerting Notification Channels HTTP API](https://grafana.com/docs/grafana/latest/http_api/alerting_notification_channels/) | 4 | [ZPascal](https://github.com/ZPascal) | | Planned | +| [Annotations HTTP API](https://grafana.com/docs/grafana/latest/http_api/annotations/) | | | | | +| [Authentication HTTP API](https://grafana.com/docs/grafana/latest/http_api/auth/) | | | | | +| [Data source HTTP API](https://grafana.com/docs/grafana/latest/http_api/data_source/) | 5 | [ZPascal](https://github.com/ZPascal) | | Planned | +| [Datasource Permissions HTTP API](https://grafana.com/docs/grafana/latest/http_api/datasource_permissions/) | | | | | +| [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/) | | | | | +| [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/) | | | | | +| [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/) | | | | | +| [Team HTTP API](https://grafana.com/docs/grafana/latest/http_api/team/) | | | | | +| [User HTTP API](https://grafana.com/docs/grafana/latest/http_api/user/) | | | | | + +## Installation + +`pip install grafana-api-sdk` + +## Example + +```python +import json + +from grafana_api.model import APIModel +from grafana_api.dashboard import Dashboard + +model: APIModel = APIModel(host="test", token="test") + +dashboard: Dashboard = Dashboard(model) -## Installation & Requirements +with open("/tmp/test/test.json") as file: + json_dashboard = json.load(file) -### Programs & tools to install +dashboard.create_or_update_dashboard(message="Create a new test dashboard", dashboard_json=json_dashboard, dashboard_path="test") +``` -- json-extensions -- requests +## Templating +If you want to template your JSON document based on a predefined folder structure you can check out one of my other [project](https://github.com/ZPascal/grafana_dashboard_templater) and integrate the functionality inside your code. ## Contribution If you would like to contribute something, have an improvement request, or want to make a change inside the code, please open a pull request. diff --git a/grafana_api/__init__.py b/docs/.placeholder similarity index 100% rename from grafana_api/__init__.py rename to docs/.placeholder diff --git a/docs/coverage.svg b/docs/coverage.svg new file mode 100644 index 0000000..e5db27c --- /dev/null +++ b/docs/coverage.svg @@ -0,0 +1,21 @@ + + + + + + + + + + + + + + + + coverage + coverage + 100% + 100% + + diff --git a/grafana_api/api.py b/grafana_api/api.py deleted file mode 100644 index b7dd7cc..0000000 --- a/grafana_api/api.py +++ /dev/null @@ -1,117 +0,0 @@ -import sys -import json -import logging -import requests - - -class GrafanaAPIModel: - - def __init__(self, host: str = None, token: str = None, message: str = None, dashboard_path: str = None, - dashboard_name: str = None): - self.host = host - self.token = token - self.message = message - self.dashboard_path = dashboard_path - self.dashboard_name = dashboard_name - - -# https://grafana.com/docs/grafana/latest/http_api/dashboard/ -class GrafanaAPI: - - def __init__(self, grafana_api_model: GrafanaAPIModel): - self.grafana_api_model = grafana_api_model - self.logging = logging.Logger - - def create_or_update_dashboard(self, dashboard_json: any, overwrite: bool = False): - folder_id: int = self.get_folder_id_by_dashboard_path() - - dashboard_json_complete: dict = { - "dashboard": dashboard_json, - "folderId": folder_id, - "message": self.grafana_api_model.message, - "overwrite": overwrite - } - - api_call = GrafanaAPI.__call_the_api(self, "/api/dashboards/db", "POST", json.dumps(dashboard_json_complete)) - - status: str = api_call["status"] - - if status != "success": - logging.error(f"Check the error: {api_call}.") - sys.exit(1) - else: - logging.info("You successfully deployed the dashboard.") - - def delete_dashboard_by_name_and_path(self): - dashboard_uids: list = self.get_dashboard_uid_by_name_and_folder() - - if dashboard_uids != list(): - for dashboard_uid in dashboard_uids: - api_call = GrafanaAPI.__call_the_api(self, f"/api/dashboards/uid/{dashboard_uid}", "DELETE") - - message: str = api_call["message"] - - if f"Dashboard {self.grafana_api_model.dashboard_name} deleted" != message: - logging.error(f"Please, check the error: {api_call}.") - sys.exit(1) - else: - logging.info("You successfully destroyed the dashboard.") - else: - logging.info("Nothing to delete. There is no dashboard available.") - - def get_dashboard_uid_by_name_and_folder(self) -> list: - folder_id: int = self.get_folder_id_by_dashboard_path() - - search_query: str = f"/api/search?folderIds={folder_id}&query={self.grafana_api_model.dashboard_name}" - dashboard_meta_list: list = GrafanaAPI.__call_the_api(self, search_query) - - dashboard_uids: list = list() - for dashboard_meta in dashboard_meta_list: - dashboard_uids.append(dashboard_meta["uid"]) - - return dashboard_uids - - def get_folder_id_by_dashboard_path(self) -> int: - folders: list = self.get_all_folder_ids_and_names() - folder_id: int = 0 - - for f in folders: - if self.grafana_api_model.dashboard_path == f["title"]: - folder_id = f["id"] - - if folder_id == 0: - logging.error("There's no folder_id available.") - sys.exit(1) - - return folder_id - - def get_all_folder_ids_and_names(self) -> list: - folders_raw: list = GrafanaAPI.__call_the_api(self, "/api/search/?folderIds=0") - folders_raw_len: int = len(folders_raw) - folders: list = list() - - for i in range(0, folders_raw_len): - folders.append({"title": folders_raw[i]["title"], "id": folders_raw[i]["id"]}) - - return folders - - def __call_the_api(self, api_call: str, method: str = "GET", dashboard_json_complete=None): - api_url: str = f"{self.grafana_api_model.host}{api_call}" - headers: dict = {"Authorization": f"Bearer {self.grafana_api_model.token}", "Content-Type": "application/json"} - try: - if method == "GET": - return requests.get(api_url, headers=headers).json() - elif method == "POST": - if dashboard_json_complete is not None: - return requests.post(api_url, data=dashboard_json_complete, headers=headers).json() - else: - logging.error("Please define the dashbboard_json_complete.") - sys.exit(1) - elif method == "DELETE": - return requests.delete(api_url, headers=headers).json() - else: - logging.error("Please a define valid method.") - sys.exit(1) - except Exception as e: - logging.error(f"Please, check the error: {e}.") - sys.exit(1) diff --git a/requirements.txt b/requirements.txt index 076a11e..663bd1f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1 @@ -json-extensions requests \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..5f672a8 --- /dev/null +++ b/setup.py @@ -0,0 +1,30 @@ +import setuptools + +with open("README.md", "r", encoding="utf-8") as fh: + coverage_string: str = "![Coverage report](https://github.com/ZPascal/grafana_api_sdk/blob/main/docs/coverage.svg)" + long_description: str = fh.read() + +long_description = long_description.replace(coverage_string, "") + +setuptools.setup( + name="grafana-api-sdk", + version="0.0.1", + author="Pascal Zimmermann", + author_email="info@theiotstudio.com", + description="A Grafana API SDK", + long_description=long_description, + long_description_content_type="text/markdown", + url="https://github.com/ZPascal/grafana_api_sdk", + project_urls={ + "Bug Tracker": "https://github.com/ZPascal/grafana_api_sdk/issues", + }, + classifiers=[ + "Programming Language :: Python :: 3", + "License :: OSI Approved", + "Operating System :: OS Independent", + ], + package_dir={"": "src"}, + packages=setuptools.find_packages(where="src"), + install_requires=["requests"], + python_requires=">=3.6", +) diff --git a/src/grafana_api/__init__.py b/src/grafana_api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/grafana_api/dashboard.py b/src/grafana_api/dashboard.py new file mode 100644 index 0000000..2890d15 --- /dev/null +++ b/src/grafana_api/dashboard.py @@ -0,0 +1,336 @@ +import json +import logging + +from .model import APIModel, APIEndpoints, RequestsMethods +from .folder import Folder +from .utils import Utils + + +class Dashboard: + """The class includes all necessary methods to access the Grafana dashboard API endpoints + + Keyword arguments: + grafana_api_model -> Inject a Grafana API model object that includes all necessary values and information + """ + + def __init__(self, grafana_api_model: APIModel): + self.grafana_api_model = grafana_api_model + + def create_or_update_dashboard( + self, + dashboard_path: str, + dashboard_json: dict, + message: str, + overwrite: bool = False, + ): + """The method includes a functionality to create the specified dashboard + + Keyword arguments: + dashboard_path -> Specify the dashboard path in which the dashboard is to be placed + dashboard_json -> Specify the inserted dashboard as dict + message -> Specify the message that should be injected as commit message inside the dashboard + overwrite -> Should the already existing dashboard be overwritten + """ + + if len(dashboard_path) != 0 and dashboard_json != dict() and len(message) != 0: + folder_id: int = Folder( + self.grafana_api_model + ).get_folder_id_by_dashboard_path(dashboard_path) + + dashboard_json_complete: dict = { + "dashboard": dashboard_json, + "folderId": folder_id, + "message": message, + "overwrite": overwrite, + } + + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/db", + RequestsMethods.POST, + json.dumps(dashboard_json_complete), + ) + + if api_call.get("status") != "success": + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully deployed the dashboard.") + else: + logging.info( + "There is no dashboard_path or dashboard_json or message defined." + ) + raise ValueError + + def delete_dashboard_by_name_and_path( + self, dashboard_name: str, dashboard_path: str + ): + """The method includes a functionality to delete the specified dashboard inside the model + + dashboard_name -> Specify the dashboard name of the deleted dashboard + dashboard_path -> Specify the dashboard path of the deleted dashboard + """ + + if len(dashboard_name) != 0 and len(dashboard_path) != 0: + dashboard_uid: dict = self.get_dashboard_uid_and_id_by_name_and_folder( + dashboard_name, dashboard_path + ) + + if len(dashboard_uid) != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/uid/{dashboard_uid.get('uid')}", + RequestsMethods.DELETE, + ) + + if f"Dashboard {dashboard_name} deleted" != api_call.get("message"): + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully destroyed the dashboard.") + else: + logging.info("Nothing to delete. There is no dashboard available.") + raise ValueError + else: + logging.info("There is no dashboard_name or dashboard_path defined.") + raise ValueError + + def get_dashboard_by_uid(self, uid: str) -> dict: + """The method includes a functionality to get the dashboard from the specified uid + + Keyword arguments: + uid -> Specify the uid of the dashboard + """ + + if len(uid) != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/uid/{uid}" + ) + + if api_call.get("dashboard") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no dashboard uid defined.") + raise ValueError + + def get_dashboard_home(self) -> dict: + """The method includes a functionality to get the home dashboard""" + + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/home" + ) + + if api_call.get("dashboard") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + + def get_dashboard_tags(self) -> list: + """The method includes a functionality to get the all tags of all dashboards""" + + api_call: list = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/tags" + ) + + if api_call == list() or api_call[0].get("term") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + + def get_dashboard_uid_and_id_by_name_and_folder( + self, dashboard_name: str, dashboard_path: str + ) -> dict: + """The method includes a functionality to extract the dashboard uid specified inside the model + + dashboard_name -> Specify the dashboard name of the dashboard + dashboard_path -> Specify the dashboard path of the dashboard + """ + + if len(dashboard_name) != 0 and len(dashboard_path) != 0: + folder_id: int = Folder( + self.grafana_api_model + ).get_folder_id_by_dashboard_path(dashboard_path) + + search_query: str = f"{APIEndpoints.SEARCH.value}?folderIds={folder_id}&query={dashboard_name}" + dashboard_meta: list = Utils(self.grafana_api_model).call_the_api( + search_query + ) + + return dict( + {"uid": dashboard_meta[0]["uid"], "id": dashboard_meta[0]["id"]} + ) + else: + logging.info("There is no dashboard_name or dashboard_path defined.") + raise ValueError + + def get_dashboard_permissions(self, id: int) -> list: + """The method includes a functionality to extract the dashboard permissions based on the specified id + + Keyword arguments: + id -> Specify the id of the dashboard + """ + + if id != 0: + api_call: list = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/id/{id}/permissions" + ) + + if api_call == list() or api_call[0].get("role") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no dashboard uid defined.") + raise ValueError + + def update_dashboard_permissions(self, id: int, permission_json: dict): + """The method includes a functionality to update the dashboard permissions based on the specified id + and the permission json document + + Keyword arguments: + id -> Specify the id of the dashboard + permission_json -> Specify the inserted permissions as dict + """ + + if id != 0 and len(permission_json) != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/id/{id}/permissions", + RequestsMethods.POST, + json.dumps(permission_json), + ) + + if api_call.get("message") != "Dashboard permissions updated": + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully modified the dashboard permissions.") + else: + logging.info("There is no dashboard uid or permission json defined.") + raise ValueError + + def get_dashboard_versions(self, id: int) -> list: + """The method includes a functionality to extract the versions of a dashboard based on the specified id + + Keyword arguments: + id -> Specify the id of the dashboard + """ + + if id != 0: + api_call: list = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/id/{id}/versions", + ) + + if api_call == list() or api_call[0].get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no dashboard uid defined.") + raise ValueError + + def get_dashboard_version(self, id: int, version_id: int) -> dict: + """The method includes a functionality to extract a specified version of a dashboard based on the specified \ + dashboard id and a version_id of the dashboard + + Keyword arguments: + id -> Specify the id of the dashboard + version_id -> Specify the version_id of a dashboard + """ + + if id != 0 and version_id != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/id/{id}/versions/{version_id}", + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no dashboard uid or version_id defined.") + raise ValueError + + def restore_dashboard_version(self, id: int, version: dict): + """The method includes a functionality to restore a specified version of a dashboard based on the specified \ + dashboard uid and a version as dict of the dashboard + + Keyword arguments: + uid -> Specify the id of the dashboard + version -> Specify the version_id of a dashboard + """ + + if id != 0 and version != dict(): + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.DASHBOARDS.value}/id/{id}/restore", + RequestsMethods.POST, + json.dumps(version), + ) + + if ( + api_call.get("status") != "success" + or api_call.get("message") is not None + ): + logging.error(f"Check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully restored the dashboard.") + else: + logging.info("There is no dashboard uid or version_id defined.") + raise ValueError + + def calculate_dashboard_diff( + self, + dashboard_id_and_version_base: dict, + dashboard_id_and_version_new: dict, + diff_type: str = "json", + ) -> str: + """The method includes a functionality to calculate the diff of specified versions of a dashboard based on the \ + specified dashboard uid and the selected version of the base dashboard and the new dashboard and the diff \ + type (basic or json) + + Keyword arguments: + dashboard_id_and_version_base -> Specify the version and id of the base dashboard + dashboard_id_and_version_new -> Specify the version and id of the new dashboard + diff_type -> Specify the diff type (basic or json the default is json) + """ + possible_diff_types: list = list(["basic", "json"]) + + if diff_type.lower() in possible_diff_types: + if ( + dashboard_id_and_version_base != dict() + and dashboard_id_and_version_new != 0 + ): + diff_object: dict = dict() + diff_object.update(dashboard_id_and_version_base) + diff_object.update(dashboard_id_and_version_new) + diff_object.update({"diffType": diff_type.lower()}) + + api_call: any = Utils( + self.grafana_api_model + ).call_the_api_non_json_output( + f"{APIEndpoints.DASHBOARDS.value}/calculate-diff", + RequestsMethods.POST, + json.dumps(diff_object), + ) + + if api_call.status_code != 200: + logging.error(f"Check the error: {api_call.text}.") + raise Exception + else: + return api_call.text + else: + logging.info( + "There is no dashboard_uid_and_version_base or dashboard_uid_and_version_new defined." + ) + raise ValueError + else: + logging.info( + f"The diff_type: {diff_type.lower()} is not valid. Please specify a valid value." + ) + raise ValueError diff --git a/src/grafana_api/folder.py b/src/grafana_api/folder.py new file mode 100644 index 0000000..904a264 --- /dev/null +++ b/src/grafana_api/folder.py @@ -0,0 +1,253 @@ +import logging +import json + +from .utils import Utils +from .model import APIModel, APIEndpoints, RequestsMethods + + +class Folder: + """The class includes all necessary methods to access the Grafana folder API endpoints + + Keyword arguments: + grafana_api_model -> Inject a Grafana API model object that includes all necessary values and information + """ + + def __init__(self, grafana_api_model: APIModel): + self.grafana_api_model = grafana_api_model + + def get_folders(self) -> list: + """The method includes a functionality to extract all folders inside the organization""" + + api_call: list = Utils(self.grafana_api_model).call_the_api( + APIEndpoints.FOLDERS.value + ) + + if api_call == list() or api_call[0].get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + + def get_folder_by_uid(self, uid: str) -> dict: + """The method includes a functionality to extract all folder information specified by the uid of the folder + + Keyword arguments: + uid -> Specify the uid of the folder + """ + + if len(uid) != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.FOLDERS.value}/{uid}" + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no dashboard uid defined.") + raise ValueError + + def get_folder_by_id(self, id: int) -> dict: + """The method includes a functionality to extract all folder information specified by the id of the folder + + Keyword arguments: + id -> Specify the id of the folder + """ + + if id != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.FOLDERS.value}/id/{id}", + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no folder id defined.") + raise ValueError + + def create_folder(self, title: str, uid: str = None) -> dict: + """The method includes a functionality to create a new folder inside the organization specified by the \ + defined title and the optional uid + + Keyword arguments: + title -> Specify the title of the folder + uid -> Specify the uid of the folder (the default value is None) + """ + + if len(title) != 0: + folder_information: dict = dict() + folder_information.update({"title": title}) + + if uid is not None and len(uid) != 0: + folder_information.update({"uid": uid}) + + api_call: dict = Utils(self.grafana_api_model).call_the_api( + APIEndpoints.FOLDERS.value, + RequestsMethods.POST, + json.dumps(folder_information), + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no folder uid or title defined.") + raise ValueError + + def update_folder( + self, title: str, uid: str = None, version: int = 0, overwrite: bool = False + ) -> dict: + """The method includes a functionality to update a folder information inside the organization specified \ + by the uid, the title, the version of the folder or if folder information be overwritten + + Keyword arguments: + uid -> Specify the uid of the folder + title -> Specify the title of the folder + version -> Specify the version of the folder + overwrite -> Should the already existing folder information be overwritten + """ + + if overwrite is True: + version = None + + if len(title) != 0 and version != 0: + folder_information: dict = dict() + folder_information.update({"title": title}) + folder_information.update({"overwrite": overwrite}) + + if uid is not None and len(uid) != 0: + folder_information.update({"uid": uid}) + + if version is not None: + folder_information.update({"version": version}) + + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.FOLDERS.value}/{uid}", + RequestsMethods.PUT, + json.dumps(folder_information), + ) + + if api_call == dict() or api_call.get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no folder title or version defined.") + raise ValueError + + def delete_folder(self, uid: str): + """The method includes a functionality to delete a folder inside the organization specified by the \ + defined uid + + Keyword arguments: + uid -> Specify the uid of the folder + """ + + if len(uid) != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.FOLDERS.value}/{uid}", + RequestsMethods.DELETE, + ) + + if "Folder deleted" != api_call.get("message"): + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully destroyed the folder.") + else: + logging.info("There is no folder uid defined.") + raise ValueError + + def get_folder_permissions(self, uid: str) -> list: + """The method includes a functionality to extract the folder permissions inside the organization specified by \ + the defined uid + + Keyword arguments: + uid -> Specify the uid of the folder + """ + + if len(uid) != 0: + api_call: list = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.FOLDERS.value}/{uid}/permissions", + RequestsMethods.GET, + ) + + if api_call == list() or api_call[0].get("id") is None: + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + return api_call + else: + logging.info("There is no folder uid defined.") + raise ValueError + + def update_folder_permissions(self, uid: str, permission_json: dict): + """The method includes a functionality to update the folder permissions based on the specified uid \ + and the permission json document + + Keyword arguments: + uid -> Specify the uid of the folder + permission_json -> Specify the inserted permissions as dict + """ + + if len(uid) != 0 and len(permission_json) != 0: + api_call: dict = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.FOLDERS.value}/{uid}/permissions", + RequestsMethods.POST, + json.dumps(permission_json), + ) + + if api_call.get("message") != "Folder permissions updated": + logging.error(f"Please, check the error: {api_call}.") + raise Exception + else: + logging.info("You successfully modified the folder permissions.") + else: + logging.info("There is no folder uid or permission json defined.") + raise ValueError + + def get_folder_id_by_dashboard_path(self, dashboard_path: str) -> int: + """The method includes a functionality to extract the folder id specified inside model dashboard path""" + + if len(dashboard_path) != 0: + folders: list = self.get_all_folder_ids_and_names() + folder_id: int = 0 + + for f in folders: + if dashboard_path == f.get("title"): + folder_id = f.get("id") + + if folder_id == 0: + logging.error( + f"There's no folder_id for the dashboard named {dashboard_path} available." + ) + raise Exception + + return folder_id + else: + logging.info("There is no dashboard_path defined.") + raise ValueError + + def get_all_folder_ids_and_names(self) -> list: + """The method extract all folder id and names inside the complete organisation""" + + folders_raw: list = Utils(self.grafana_api_model).call_the_api( + f"{APIEndpoints.SEARCH.value}?folderIds=0" + ) + folders_raw_len: int = len(folders_raw) + folders: list = list() + + for i in range(0, folders_raw_len): + folders.append( + {"title": folders_raw[i].get("title"), "id": folders_raw[i].get("id")} + ) + + return folders diff --git a/src/grafana_api/model.py b/src/grafana_api/model.py new file mode 100644 index 0000000..b6c9e35 --- /dev/null +++ b/src/grafana_api/model.py @@ -0,0 +1,46 @@ +from enum import Enum + +"""""" +ERROR_MESSAGES: list = ["invalid API key"] + + +class APIEndpoints(Enum): + """The class includes all necessary methods to template the selected dashboard and return it as a dict + + Keyword arguments: + dashboard_model -> Inject a dashboard object that includes all necessary values and information + """ + + SEARCH = "/api/search" + DASHBOARDS = "/api/dashboards" + FOLDERS = "/api/folders" + + +class RequestsMethods(Enum): + """The class includes all necessary methods to template the selected dashboard and return it as a dict + + Keyword arguments: + dashboard_model -> Inject a dashboard object that includes all necessary values and information + """ + + GET = "GET" + PUT = "PUT" + POST = "POST" + DELETE = "DELETE" + + +class APIModel: + """The class includes all necessary variables to establish a connection to the Grafana API endpoints + + Keyword arguments: + host -> Specify the host of the Grafana system + token -> Specify the access token of the Grafana system + """ + + def __init__( + self, + host: str = None, + token: str = None, + ): + self.host = host + self.token = token diff --git a/src/grafana_api/search.py b/src/grafana_api/search.py new file mode 100644 index 0000000..b77711b --- /dev/null +++ b/src/grafana_api/search.py @@ -0,0 +1,26 @@ +from .utils import Utils +from .model import APIModel + + +class Search: + """The class includes all necessary methods to access the Grafana search API endpoints + + Keyword arguments: + grafana_api_model -> Inject a Grafana API model object that includes all necessary values and information + """ + + def __init__(self, grafana_api_model: APIModel): + self.grafana_api_model = grafana_api_model + + def search(self, search_query: str) -> list: + """The method includes a functionality to execute a custom query + + Keyword arguments: + search_query -> Specify the inserted query as string + """ + + result: list = Utils(self.grafana_api_model).call_the_api(search_query) + if result == list(): + raise Exception + else: + return result diff --git a/src/grafana_api/utils.py b/src/grafana_api/utils.py new file mode 100644 index 0000000..b63d42b --- /dev/null +++ b/src/grafana_api/utils.py @@ -0,0 +1,128 @@ +import logging +import requests + +from .model import RequestsMethods, ERROR_MESSAGES, APIModel + + +class Utils: + """The class includes all necessary methods to make API calls to the Grafana API endpoints + + Keyword arguments: + grafana_api_model -> Inject a Grafana API model object that includes all necessary values and information + """ + + def __init__(self, grafana_api_model: APIModel): + self.grafana_api_model = grafana_api_model + + def call_the_api( + self, + api_call: str, + method: RequestsMethods = RequestsMethods.GET, + json_complete: str = None, + ) -> any: + """The method execute a defined API call against the Grafana endpoints + + Keyword arguments: + api_call -> Specify the API call endpoint + method -> Specify the used method + json_complete -> Specify the inserted JSON as string + """ + + api_url: str = f"{self.grafana_api_model.host}{api_call}" + + headers: dict = { + "Authorization": f"Bearer {self.grafana_api_model.token}", + "Content-Type": "application/json", + } + try: + if method.value == RequestsMethods.GET.value: + return Utils.__check_the_api_call_response( + requests.get(api_url, headers=headers).json() + ) + elif method.value == RequestsMethods.PUT.value: + if json_complete is not None: + return Utils.__check_the_api_call_response( + requests.put( + api_url, data=json_complete, headers=headers + ).json() + ) + else: + logging.error("Please define the json_complete.") + raise Exception + elif method.value == RequestsMethods.POST.value: + if json_complete is not None: + return Utils.__check_the_api_call_response( + requests.post( + api_url, data=json_complete, headers=headers + ).json() + ) + else: + logging.error("Please define the json_complete.") + raise Exception + elif method.value == RequestsMethods.DELETE.value: + return Utils.__check_the_api_call_response( + requests.delete(api_url, headers=headers).json() + ) + else: + logging.error("Please define a valid method.") + raise Exception + except Exception as e: + raise e + + def call_the_api_non_json_output( + self, + api_call: str, + method: RequestsMethods = RequestsMethods.GET, + json_complete: str = None, + ) -> any: + """The method execute a defined API call against the Grafana endpoints + + Keyword arguments: + api_call -> Specify the API call endpoint + method -> Specify the used method + json_complete -> Specify the inserted JSON as string + """ + + api_url: str = f"{self.grafana_api_model.host}{api_call}" + + headers: dict = { + "Authorization": f"Bearer {self.grafana_api_model.token}", + "Content-Type": "application/json", + } + try: + if method.value == RequestsMethods.GET.value: + return Utils.__check_the_api_call_response( + requests.get(api_url, headers=headers) + ) + elif method.value == RequestsMethods.POST.value: + if json_complete is not None: + return Utils.__check_the_api_call_response( + requests.post(api_url, data=json_complete, headers=headers) + ) + else: + logging.error("Please define the json_complete.") + raise Exception + elif method.value == RequestsMethods.DELETE.value: + return Utils.__check_the_api_call_response( + requests.delete(api_url, headers=headers) + ) + else: + logging.error("Please define a valid method.") + raise Exception + except Exception as e: + raise e + + @staticmethod + def __check_the_api_call_response(response: any = None) -> any: + """The method includes a functionality to check the output of API call method for errors + + Keyword arguments: + response -> Specify the inserted response + """ + + if type(response) == dict: + if "message" in response.keys() and response["message"] in ERROR_MESSAGES: + logging.error(response["message"]) + raise requests.exceptions.ConnectionError + + return response diff --git a/tests/integrationtest/resources/dashboard.json b/tests/integrationtest/resources/dashboard.json new file mode 100644 index 0000000..f20876e --- /dev/null +++ b/tests/integrationtest/resources/dashboard.json @@ -0,0 +1,30 @@ +{ + "id": null, + "uid": "tests", + "title": "Test", + "tags": [], + "style": "dark", + "timezone": "browser", + "editable": true, + "hideControls": false, + "graphTooltip": 1, + "panels": [], + "time": { + "from": "now-6h", + "to": "now" + }, + "timepicker": { + "time_options": [], + "refresh_intervals": [] + }, + "templating": { + "list": [] + }, + "annotations": { + "list": [] + }, + "refresh": "5s", + "schemaVersion": 17, + "version": 0, + "links": [] +} \ No newline at end of file diff --git a/tests/integrationtest/resources/dashboard_expected_result.json b/tests/integrationtest/resources/dashboard_expected_result.json new file mode 100644 index 0000000..007becc --- /dev/null +++ b/tests/integrationtest/resources/dashboard_expected_result.json @@ -0,0 +1,55 @@ +{ + "dashboard": { + "annotations": { + "list": [] + }, + "editable": true, + "graphTooltip": 1, + "hideControls": false, + "id": 104, + "links": [], + "panels": [], + "refresh": "5s", + "schemaVersion": 17, + "style": "dark", + "tags": [], + "templating": { + "list": [] + }, + "time": { + "from": "now-6h", + "to": "now" + }, + "timepicker": { + "refresh_intervals": [], + "time_options": [] + }, + "timezone": "browser", + "title": "Test 1", + "uid": "tests1", + "version": 1 + }, + "meta": { + "canAdmin": true, + "canEdit": true, + "canSave": true, + "canStar": true, + "created": "2022-01-15T20:46:44+01:00", + "createdBy": "Anonymous", + "expires": "0001-01-01T00:00:00Z", + "folderId": 72, + "folderTitle": "Github Integrationtest", + "folderUid": "6U_QdWJnz", + "folderUrl": "/dashboards/f/6U_QdWJnz/github-integrationtest", + "hasAcl": false, + "isFolder": false, + "provisioned": false, + "provisionedExternalId": "", + "slug": "test-1", + "type": "db", + "updated": "2022-01-15T20:46:44+01:00", + "updatedBy": "Anonymous", + "url": "/d/tests1/test-1", + "version": 1 + } +} \ No newline at end of file diff --git a/tests/integrationtest/test_dashboard.py b/tests/integrationtest/test_dashboard.py new file mode 100644 index 0000000..e59b991 --- /dev/null +++ b/tests/integrationtest/test_dashboard.py @@ -0,0 +1,62 @@ +import os +import json +from unittest import TestCase, main + +import requests.exceptions + +from src.grafana_api.model import APIModel +from src.grafana_api.dashboard import Dashboard +from src.grafana_api.folder import Folder + + +class DashboardTest(TestCase): + model: APIModel = APIModel( + host=os.environ["GRAFANA_HOST"], + token=os.environ["GRAFANA_TOKEN"], + ) + dashboard: Dashboard = Dashboard(model) + folder: Folder = Folder(model) + + def test_a_dashboard_creation(self): + with open( + f"{os.getcwd()}{os.sep}tests{os.sep}integrationtest{os.sep}resources{os.sep}dashboard.json" + ) as file: + json_dashboard = json.load(file) + + self.dashboard.create_or_update_dashboard( + message="Create a new test dashboard", + dashboard_json=json_dashboard, + dashboard_path=os.environ["GRAFANA_DASHBOARD_PATH"], + overwrite=True, + ) + + self.assertEqual( + "tests", self.dashboard.get_dashboard_uid_and_id_by_name_and_folder( + dashboard_path=os.environ["GRAFANA_DASHBOARD_PATH"], + dashboard_name=os.environ["GRAFANA_DASHBOARD_NAME"])["uid"] + ) + self.assertEqual(72, self.folder.get_folder_id_by_dashboard_path( + dashboard_path=os.environ["GRAFANA_DASHBOARD_PATH"])) + + def test_b_get_dashboard(self): + with open( + f"{os.getcwd()}{os.sep}tests{os.sep}integrationtest{os.sep}resources{os.sep}dashboard_expected_result.json" + ) as file: + json_dashboard = json.load(file) + + self.assertEqual(json_dashboard, self.dashboard.get_dashboard_by_uid("tests1")) + + def test_c_dashboard_deletion(self): + self.dashboard.delete_dashboard_by_name_and_path(dashboard_path=os.environ["GRAFANA_DASHBOARD_PATH"], + dashboard_name=os.environ["GRAFANA_DASHBOARD_NAME"]) + + def test_wrong_token(self): + self.model.token = "test" + + with self.assertRaises(requests.exceptions.ConnectionError): + self.dashboard.delete_dashboard_by_name_and_path(dashboard_path=os.environ["GRAFANA_DASHBOARD_PATH"], + dashboard_name=os.environ["GRAFANA_DASHBOARD_NAME"]) + + +if __name__ == "__main__": + main() diff --git a/tests/unittests/test_dashboard.py b/tests/unittests/test_dashboard.py new file mode 100644 index 0000000..a6731a3 --- /dev/null +++ b/tests/unittests/test_dashboard.py @@ -0,0 +1,382 @@ +from unittest import TestCase +from unittest.mock import MagicMock, patch + +from src.grafana_api.model import APIModel +from src.grafana_api.dashboard import Dashboard + + +class DashboardTestCase(TestCase): + @patch("src.grafana_api.utils.Utils.call_the_api") + @patch("src.grafana_api.folder.Folder.get_folder_id_by_dashboard_path") + def test_create_or_update_dashboard( + self, folder_id_by_dashboard_path_mock, call_the_api_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + folder_id_by_dashboard_path_mock.return_value = 1 + call_the_api_mock.return_value = dict({"status": "success"}) + self.assertEqual( + None, + dashboard.create_or_update_dashboard( + dashboard_path="test", + dashboard_json=dict({"test": "test"}), + message="test", + ), + ) + + def test_create_or_update_dashboard_no_dashboard_path_defined(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.create_or_update_dashboard( + dashboard_path="", dashboard_json=dict({"test": "test"}), message="test" + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + @patch("src.grafana_api.folder.Folder.get_folder_id_by_dashboard_path") + def test_create_or_update_dashboard_update_not_possible( + self, folder_id_by_dashboard_path_mock, call_the_api_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + folder_id_by_dashboard_path_mock.return_value = 1 + call_the_api_mock.return_value = dict({"status": "error"}) + with self.assertRaises(Exception): + dashboard.create_or_update_dashboard( + dashboard_path="test", + dashboard_json=dict({"test": "test"}), + message="test", + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + @patch( + "src.grafana_api.dashboard.Dashboard.get_dashboard_uid_and_id_by_name_and_folder" + ) + def test_delete_dashboard_by_name_and_path( + self, dashboard_uid_and_id_by_name_and_folder_mock, call_the_api_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + dashboard_uid_and_id_by_name_and_folder_mock.return_value = dict( + {"uid": "test", "id": 10} + ) + call_the_api_mock.return_value = dict({"message": "Dashboard test deleted"}) + self.assertEqual( + None, + dashboard.delete_dashboard_by_name_and_path( + dashboard_name="test", dashboard_path="test" + ), + ) + + def test_delete_dashboard_by_name_and_path_no_dashboard_name( + self, + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.delete_dashboard_by_name_and_path( + dashboard_name="", dashboard_path="test" + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + @patch( + "src.grafana_api.dashboard.Dashboard.get_dashboard_uid_and_id_by_name_and_folder" + ) + def test_delete_dashboard_by_name_and_path_deletion_list_empty( + self, dashboard_uid_and_id_by_name_and_folder_mock, call_the_api_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + dashboard_uid_and_id_by_name_and_folder_mock.return_value = dict() + call_the_api_mock.return_value = dict({"message": "error"}) + with self.assertRaises(ValueError): + dashboard.delete_dashboard_by_name_and_path( + dashboard_name="test", dashboard_path="test" + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + @patch( + "src.grafana_api.dashboard.Dashboard.get_dashboard_uid_and_id_by_name_and_folder" + ) + def test_delete_dashboard_by_name_and_path_deletion_not_possible( + self, dashboard_uid_and_id_by_name_and_folder_mock, call_the_api_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + dashboard_uid_and_id_by_name_and_folder_mock.return_value = dict( + {"uid": "test", "id": 10} + ) + call_the_api_mock.return_value = dict({"message": "error"}) + with self.assertRaises(Exception): + dashboard.delete_dashboard_by_name_and_path( + dashboard_name="test", dashboard_path="test" + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + @patch("src.grafana_api.folder.Folder.get_folder_id_by_dashboard_path") + def test_get_dashboard_uid_and_id_by_name_and_folder( + self, folder_id_by_dashboard_path_mock, call_the_api_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + folder_id_by_dashboard_path_mock.return_value = 1 + call_the_api_mock.return_value = [{"uid": "test", "id": 10}] + self.assertEqual( + dict({"uid": "test", "id": 10}), + dashboard.get_dashboard_uid_and_id_by_name_and_folder( + dashboard_name="test", dashboard_path="test" + ), + ) + + def test_get_dashboard_uid_and_id_by_name_and_folder_bo_dashboard_name_defined( + self, + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.get_dashboard_uid_and_id_by_name_and_folder( + dashboard_name="", dashboard_path="test" + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_by_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"dashboard": "test"}) + self.assertEqual( + dict({"dashboard": "test"}), dashboard.get_dashboard_by_uid(uid="test") + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_by_uid_no_dashboard(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(Exception): + dashboard.get_dashboard_by_uid(uid="test") + + def test_get_dashboard_by_uid_no_uid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.get_dashboard_by_uid(uid="") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_home(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"dashboard": "test"}) + self.assertEqual(dict({"dashboard": "test"}), dashboard.get_dashboard_home()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_home_no_dashboard(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(Exception): + dashboard.get_dashboard_home() + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_tags(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = list([{"term": "test", "count": 4}]) + self.assertEqual( + list([{"term": "test", "count": 4}]), dashboard.get_dashboard_tags() + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_tags_no_tags(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = list() + with self.assertRaises(Exception): + dashboard.get_dashboard_tags() + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_permissions(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = list([{"role": "test", "count": 4}]) + self.assertEqual( + list([{"role": "test", "count": 4}]), + dashboard.get_dashboard_permissions(1), + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_permissions_empty_list(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = list() + with self.assertRaises(Exception): + dashboard.get_dashboard_permissions(1) + + def test_get_dashboard_permissions_no_id(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.get_dashboard_permissions(0) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_dashboard_permissions(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict( + {"message": "Dashboard permissions updated"} + ) + self.assertEqual( + None, dashboard.update_dashboard_permissions(1, {"test": "test"}) + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_dashboard_permissions_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"message": "Error"}) + with self.assertRaises(Exception): + dashboard.update_dashboard_permissions(1, {"test": "test"}) + + def test_update_dashboard_permissions_no_uid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.update_dashboard_permissions(0, MagicMock()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_versions(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = list([{"id": "test"}]) + self.assertEqual(list([{"id": "test"}]), dashboard.get_dashboard_versions(1)) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_versions_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = list() + with self.assertRaises(Exception): + dashboard.get_dashboard_versions(1) + + def test_get_dashboard_versions_no_uid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.get_dashboard_versions(0) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_version(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"id": "test"}) + self.assertEqual(dict({"id": "test"}), dashboard.get_dashboard_version(1, 10)) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_dashboard_version_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(Exception): + dashboard.get_dashboard_version(1, MagicMock()) + + def test_get_dashboard_version_no_uid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.get_dashboard_version(0, MagicMock()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_restore_dashboard_version(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"status": "success"}) + self.assertEqual( + None, dashboard.restore_dashboard_version(1, dict({"version": 1})) + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_restore_dashboard_version_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"status": "error"}) + with self.assertRaises(Exception): + dashboard.restore_dashboard_version(1, dict({"version": 1})) + + def test_restore_dashboard_version_no_uid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.restore_dashboard_version(0, MagicMock()) + + @patch("src.grafana_api.utils.Utils.call_the_api_non_json_output") + def test_calculate_dashboard_diff(self, call_the_api_non_json_output_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_non_json_output_mock.return_value.status_code = 200 + call_the_api_non_json_output_mock.return_value.text = "test" + self.assertEqual( + "test", + dashboard.calculate_dashboard_diff( + dict({"dashboardId": 1, "version": 1}), + dict({"dashboardId": 2, "version": 1}), + ), + ) + + @patch("src.grafana_api.utils.Utils.call_the_api_non_json_output") + def test_calculate_dashboard_diff_error_response( + self, call_the_api_non_json_output_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + call_the_api_non_json_output_mock.status_code.return_value = 400 + with self.assertRaises(Exception): + dashboard.calculate_dashboard_diff( + dict({"dashboardId": 1, "version": 1}), + dict({"dashboardId": 2, "version": 1}), + ) + + def test_calculate_dashboard_diff_no_dashboard_id_and_version_base(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.calculate_dashboard_diff({}, MagicMock()) + + def test_calculate_dashboard_diff_no_valid_diff_type(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + dashboard: Dashboard = Dashboard(grafana_api_model=model) + + with self.assertRaises(ValueError): + dashboard.calculate_dashboard_diff({}, MagicMock(), "test") diff --git a/tests/unittests/test_folder.py b/tests/unittests/test_folder.py new file mode 100644 index 0000000..9f5efa5 --- /dev/null +++ b/tests/unittests/test_folder.py @@ -0,0 +1,288 @@ +from unittest import TestCase +from unittest.mock import MagicMock, patch + +from src.grafana_api.model import APIModel +from src.grafana_api.folder import Folder + + +class FolderTestCase(TestCase): + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folders(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = list([{"title": None, "id": 12}]) + self.assertEqual(list([{"title": None, "id": 12}]), folder.get_folders()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folders_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = list() + with self.assertRaises(Exception): + folder.get_folders() + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_by_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"title": None, "id": 12}) + self.assertEqual( + dict({"title": None, "id": 12}), folder.get_folder_by_uid("xty13y") + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_by_uid_no_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(ValueError): + folder.get_folder_by_uid("") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_by_uid_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(Exception): + folder.get_folder_by_uid("xty13y") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_by_id(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"title": None, "id": 12}) + self.assertEqual(dict({"title": None, "id": 12}), folder.get_folder_by_id(12)) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_by_id_no_id(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(ValueError): + folder.get_folder_by_id(0) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_by_id_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(Exception): + folder.get_folder_by_id(10) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_create_folder(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"title": None, "id": 12}) + self.assertEqual( + dict({"title": None, "id": 12}), folder.create_folder("test") + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_create_folder_specified_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"title": None, "id": 12, "uid": "test"}) + self.assertEqual( + dict({"title": None, "id": 12, "uid": "test"}), folder.create_folder("test", "test") + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_create_folder_no_title(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(ValueError): + folder.create_folder(MagicMock()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_create_folder_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(Exception): + folder.create_folder("test") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"title": "test1", "id": 12}) + self.assertEqual( + dict({"title": "test1", "id": 12}), folder.update_folder("test", "test1", 10) + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder_no_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"title": "test", "id": 12}) + self.assertEqual( + dict({"title": "test", "id": 12}), + folder.update_folder("test", overwrite=True), + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder_overwrite_true(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"title": "test", "id": 12}) + self.assertEqual( + dict({"title": "test", "id": 12}), + folder.update_folder("test", "test", overwrite=True), + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder_no_title(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(ValueError): + folder.update_folder(MagicMock(), MagicMock()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(Exception): + folder.update_folder("test", "test", 10) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_delete_folder(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"message": "Folder deleted"}) + self.assertEqual(None, folder.delete_folder("test")) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_delete_folder_no_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(ValueError): + folder.delete_folder("") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_delete_folder_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"message": "error"}) + with self.assertRaises(Exception): + folder.delete_folder("test") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_permissions(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = list([{"id": "test"}]) + self.assertEqual(list([{"id": "test"}]), folder.get_folder_permissions("test")) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_permissions_no_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = list() + with self.assertRaises(ValueError): + folder.get_folder_permissions("") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_folder_permissions_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = list([{"test": "test"}]) + with self.assertRaises(Exception): + folder.get_folder_permissions("test") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder_permissions(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"message": "Folder permissions updated"}) + self.assertEqual( + None, folder.update_folder_permissions("test", dict({"test": "test"})) + ) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder_permissions_no_uid(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict() + with self.assertRaises(ValueError): + folder.update_folder_permissions("", dict()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_update_folder_permissions_error_response(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = dict({"message": "test"}) + with self.assertRaises(Exception): + folder.update_folder_permissions("test", dict({"test": "test"})) + + @patch("src.grafana_api.folder.Folder.get_all_folder_ids_and_names") + def test_get_folder_id_by_dashboard_path(self, all_folder_ids_and_names_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + all_folder_ids_and_names_mock.return_value = list([{"title": "test", "id": 12}]) + self.assertEqual( + 12, folder.get_folder_id_by_dashboard_path(dashboard_path="test") + ) + + def test_get_folder_id_by_dashboard_path_no_dashboard_path_defined(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + with self.assertRaises(ValueError): + folder.get_folder_id_by_dashboard_path(dashboard_path="") + + @patch("src.grafana_api.folder.Folder.get_all_folder_ids_and_names") + def test_get_folder_id_by_dashboard_path_no_title_match( + self, all_folder_ids_and_names_mock + ): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + all_folder_ids_and_names_mock.return_value = list( + [{"title": None, "id": "xty13y"}] + ) + with self.assertRaises(Exception): + folder.get_folder_id_by_dashboard_path(dashboard_path="test") + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_get_all_folder_ids_and_names(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + folder: Folder = Folder(grafana_api_model=model) + + call_the_api_mock.return_value = list( + [{"title": "test", "id": 12, "test": "test"}] + ) + self.assertEqual( + list([{"title": "test", "id": 12}]), folder.get_all_folder_ids_and_names() + ) diff --git a/tests/unittests/test_model.py b/tests/unittests/test_model.py new file mode 100644 index 0000000..3807ab2 --- /dev/null +++ b/tests/unittests/test_model.py @@ -0,0 +1,24 @@ +from unittest import TestCase + +from src.grafana_api.model import APIModel, RequestsMethods, APIEndpoints + + +class APIEndpointsTestCase(TestCase): + def test_api_endpoints_init(self): + self.assertEqual("APIEndpoints.DASHBOARDS", str(APIEndpoints.DASHBOARDS)) + self.assertEqual("APIEndpoints.SEARCH", str(APIEndpoints.SEARCH)) + + +class RequestsMethodsTestCase(TestCase): + def test_requests_methods_init(self): + self.assertEqual("RequestsMethods.GET", str(RequestsMethods.GET)) + self.assertEqual("RequestsMethods.POST", str(RequestsMethods.POST)) + self.assertEqual("RequestsMethods.DELETE", str(RequestsMethods.DELETE)) + + +class APIModelTestCase(TestCase): + def test_api_model_init(self): + model = APIModel(host="test", token="test") + + self.assertEqual("test", model.host) + self.assertEqual("test", model.token) diff --git a/tests/unittests/test_search.py b/tests/unittests/test_search.py new file mode 100644 index 0000000..b9bf11e --- /dev/null +++ b/tests/unittests/test_search.py @@ -0,0 +1,36 @@ +from unittest import TestCase +from unittest.mock import MagicMock, patch + +from src.grafana_api.model import APIModel +from src.grafana_api.search import Search + + +class SearchTestCase(TestCase): + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_search(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + search: Search = Search(grafana_api_model=model) + + call_the_api_mock.return_value = ["test"] + + self.assertEqual(["test"], search.search(search_query=MagicMock())) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_search_invalid_empty_list(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + search: Search = Search(grafana_api_model=model) + + call_the_api_mock.return_value = list() + + with self.assertRaises(Exception): + search.search(search_query=MagicMock()) + + @patch("src.grafana_api.utils.Utils.call_the_api") + def test_search_invalid_output(self, call_the_api_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + search: Search = Search(grafana_api_model=model) + + call_the_api_mock.side_effect = Exception + + with self.assertRaises(Exception): + search.search(search_query=MagicMock()) diff --git a/tests/unittests/test_utils.py b/tests/unittests/test_utils.py new file mode 100644 index 0000000..28a9b25 --- /dev/null +++ b/tests/unittests/test_utils.py @@ -0,0 +1,237 @@ +import requests + +from unittest import TestCase +from unittest.mock import MagicMock, patch, Mock + +from requests.exceptions import MissingSchema + +from src.grafana_api.model import APIModel, RequestsMethods +from src.grafana_api.utils import Utils + + +class UtilsTestCase(TestCase): + def test_call_the_api_non_method(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.call_the_api(api_call=MagicMock(), method=None) + + def test_call_the_api_non_valid_method(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.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()) + utils: Utils = Utils(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value={"status": "success"}) + + get_mock.return_value = mock + + self.assertEqual( + "success", + utils.call_the_api(api_call=MagicMock())["status"], + ) + + def test_call_the_api_get_not_valid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(MissingSchema): + utils.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()) + utils: Utils = Utils(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value={"status": "success"}) + + put_mock.return_value = mock + + self.assertEqual( + "success", + utils.call_the_api( + api_call=MagicMock(), + method=RequestsMethods.PUT, + json_complete=MagicMock(), + )["status"], + ) + + def test_call_the_api_put_not_valid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.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()) + utils: Utils = Utils(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value={"status": "success"}) + + post_mock.return_value = mock + + self.assertEqual( + "success", + utils.call_the_api( + api_call=MagicMock(), + method=RequestsMethods.POST, + json_complete=MagicMock(), + )["status"], + ) + + def test_call_the_api_post_not_valid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(MissingSchema): + utils.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()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.call_the_api(api_call=MagicMock(), method=RequestsMethods.POST) + + @patch("requests.delete") + def test_call_the_api_delete_valid(self, delete_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + mock: Mock = Mock() + mock.json = Mock(return_value={"message": "Deletion successful"}) + + delete_mock.return_value = mock + + self.assertEqual( + "Deletion successful", + utils.call_the_api(api_call=MagicMock(), method=RequestsMethods.DELETE)[ + "message" + ], + ) + + def test_call_the_api_delete_not_valid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.call_the_api(api_call=MagicMock(), method=RequestsMethods.DELETE) + + def test_call_the_api_non_json_output_non_method(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.call_the_api_non_json_output(api_call=MagicMock(), method=None) + + def test_call_the_api_non_json_output_non__valid_method(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.call_the_api_non_json_output(api_call=MagicMock(), method=MagicMock()) + + @patch("requests.get") + def test_call_the_api_non_json_output_get_valid(self, get_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + get_mock.return_value.text = "success" + + self.assertEqual( + "success", + utils.call_the_api_non_json_output(api_call=MagicMock()).text, + ) + + def test_call_the_api_non_json_output_get_not_valid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(MissingSchema): + utils.call_the_api_non_json_output( + api_call=MagicMock(), method=RequestsMethods.GET + ) + + @patch("requests.post") + def test_call_the_api_non_json_output_post_valid(self, post_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + post_mock.return_value.text = "success" + + self.assertEqual( + "success", + utils.call_the_api_non_json_output( + api_call=MagicMock(), + method=RequestsMethods.POST, + json_complete=MagicMock(), + ).text, + ) + + def test_call_the_api_non_json_output_post_not_valid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(MissingSchema): + utils.call_the_api_non_json_output( + api_call=MagicMock(), + method=RequestsMethods.POST, + json_complete=MagicMock(), + ) + + def test_call_the_api_non_json_output_post_no_data(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.call_the_api_non_json_output( + api_call=MagicMock(), method=RequestsMethods.POST + ) + + @patch("requests.delete") + def test_call_the_api_non_json_output_delete_valid(self, delete_mock): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + delete_mock.return_value.text = "Deletion successful" + + self.assertEqual( + "Deletion successful", + utils.call_the_api_non_json_output( + api_call=MagicMock(), method=RequestsMethods.DELETE + ).text, + ) + + def test_call_the_api_non_json_output_delete_not_valid(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(Exception): + utils.call_the_api_non_json_output( + api_call=MagicMock(), method=RequestsMethods.DELETE + ) + + def test_check_the_api_call_response(self): + model: APIModel = APIModel(host=MagicMock(), token=MagicMock()) + utils: Utils = Utils(grafana_api_model=model) + + with self.assertRaises(requests.exceptions.ConnectionError): + utils._Utils__check_the_api_call_response( + response=dict({"message": "invalid API key"}) + )