From a46e67ed27072f3597e95a5ed0d029d278e5e071 Mon Sep 17 00:00:00 2001 From: Nazar F Date: Thu, 13 Jun 2024 10:22:05 +0200 Subject: [PATCH] feat(IASOClient): implement IASO client --- README.md | 4 +- docs/iaso.md | 51 +++++ openhexa/toolbox/iaso/__init__.py | 4 + openhexa/toolbox/iaso/api_client.py | 135 +++++++++++++ openhexa/toolbox/iaso/iaso.py | 126 ++++++++++++ pyproject.toml | 2 + tests/iaso/fixtures/iaso_api_fixtures.py | 243 +++++++++++++++++++++++ tests/iaso/test_iaso.py | 111 +++++++++++ tests/test_lib.py | 6 - 9 files changed, 674 insertions(+), 8 deletions(-) create mode 100644 docs/iaso.md create mode 100644 openhexa/toolbox/iaso/__init__.py create mode 100644 openhexa/toolbox/iaso/api_client.py create mode 100644 openhexa/toolbox/iaso/iaso.py create mode 100644 tests/iaso/fixtures/iaso_api_fixtures.py create mode 100644 tests/iaso/test_iaso.py delete mode 100644 tests/test_lib.py diff --git a/README.md b/README.md index 5306ac5..c6f21a7 100644 --- a/README.md +++ b/README.md @@ -29,5 +29,5 @@ pip install openhexa.toolbox ## Modules -[**openhexa.toolbox.dhis2**](docs/dhis2.md)
-Acquire and process data from DHIS2 instances \ No newline at end of file +[**openhexa.toolbox.dhis2**](docs/dhis2.md) - Acquire and process data from DHIS2 instances
+[**openhexa.toolbox.iaso**](docs/iaso.md) - Acquire and process data from IASO instances
\ No newline at end of file diff --git a/docs/iaso.md b/docs/iaso.md new file mode 100644 index 0000000..fe68e88 --- /dev/null +++ b/docs/iaso.md @@ -0,0 +1,51 @@ + +# OpenHEXA Toolbox IASO + +Module to fetch data from IASO + +* [Installation](#installation) +* [Usage](#usage) + * [Connect to an instance](#connect-to-an-instance) + * [Read data](#read-data) + +## [Installation](#) + +``` sh +pip install openhexa.toolbox +``` + +## [Usage](#) + +### [Connect to an instance](#) +Credentials are required to initialize a connection to IASO instance. Credentials should contain the username and +password to connect to an instance of IASO. You have as well to provide the host name to for the api to connect to: +* Staging environment https://iaso-staging.bluesquare.org/api +* Production environment https://iaso.bluesquare.org/api + +Import IASO module as: +``` +from openhexa.toolbox.iaso import IASO + +iaso = IASO("https://iaso-staging.bluesquare.org","username", "password") +``` + +### [Read data](#) +After importing IASO module, you can use provided method to fetch Projects, Organisation Units and Forms that you have +permissions for. +``` +# Fetch projects +iaso.get_projects() +# Fetch organisation units +iaso.get_org_units() +# Fetch submitted forms filtered by form_ids passed in url parameters and with choice to fetch them as dataframe +iaso.get_form_instances(page=1, limit=1, as_dataframe=True, + dataframe_columns=["Date de création","Date de modification","Org unit"], ids=276) +# Fetch forms filtered by organisaiton units and projects that you have permissions to +iaso.get_forms(org_units=[781], projects=[149]) +``` + +You can as well provide additional parameters to the method to filter on desired values as key value arguments. +You can have an overview on the arguments you can filter on API documentation of IASO. + + + diff --git a/openhexa/toolbox/iaso/__init__.py b/openhexa/toolbox/iaso/__init__.py new file mode 100644 index 0000000..9c3c186 --- /dev/null +++ b/openhexa/toolbox/iaso/__init__.py @@ -0,0 +1,4 @@ +from .api_client import ApiClient +from .iaso import IASO + +__all__ = ["IASO", "ApiClient"] diff --git a/openhexa/toolbox/iaso/api_client.py b/openhexa/toolbox/iaso/api_client.py new file mode 100644 index 0000000..17ecfbb --- /dev/null +++ b/openhexa/toolbox/iaso/api_client.py @@ -0,0 +1,135 @@ +import logging +from datetime import datetime, timezone +from typing import Union + +import requests +import jwt +from requests.adapters import HTTPAdapter +from urllib3 import Retry + + +class IASOError(Exception): + """ + Base exception for IASO API errors. + """ + + def __init__(self, message: str): + self.message = message + super().__init__(self.message) + self.log_error() + + def __str__(self): + return self.message + + def log_error(self): + logging.error(f"IASO Error : {self.message}") + + +class ApiClient(requests.Session): + """ + Client to manage HTTP session with IASO API on behalf of OpenHexa toolbox + + """ + + def __init__(self, server_url: str, username: str, password: str): + """ + Initialize the IASO API client. + + :param server_url: IASO server URL + :param username: IASO instance username + :param password: IASO instance password + + Examples: + >>> client = ApiClient(server_url="http://localhost:8080", username="admin", password="") + """ + super().__init__() + self.server_url = server_url.rstrip("/") + self.username = username + self.password = password + self.headers.update( + { + "User-Agent": "Openhexa-Toolbox", + } + ) + self.token = None + self.token_expiry = None + self._refresh_token = None + self.authenticate() + + def request(self, method: str, url: str, *args, **kwargs) -> requests.Response: + """ + Sends HTTP request to IASO API, handles exceptions raised during request + """ + full_url = f"{self.server_url}/{url.strip('/')}/" + try: + resp = super().request(method, full_url, *args, **kwargs) + self.raise_if_error(resp) + return resp + except requests.RequestException as exc: + logging.exception(exc) + raise + + def authenticate(self) -> None: + """ + Authenticates with OpenHexa API with username and password. + Calling the endpoints to fetch authorization and refresh token. + Ensures that failures are handles with status management, both with or without SSL communication + """ + credentials = {"username": self.username, "password": self.password} + response = self.request("POST", "/api/token/", json=credentials) + json_data = response.json() + self.token = json_data["access"] + self.token_expiry = self.decode_token_expiry(self.token) + self._refresh_token = json_data["refresh"] + self.headers.update({"Authorization": f"Bearer {self.token}"}) + adapter = HTTPAdapter( + max_retries=Retry( + total=3, + backoff_factor=5, + allowed_methods=["HEAD", "GET"], + status_forcelist=[429, 500, 502, 503, 504], + ) + ) + self.mount("https://", adapter) + self.mount("http://", adapter) + + def refresh_session(self) -> None: + """ + Refreshes the session token by calling the refresh endpoint and updates the authentication token + """ + response = self.request("POST", "/api/token/refresh/", json={"refresh": self._refresh_token}) + self.token = response.json()["access"] + self.headers.update({"Authorization": f"Bearer {self.token}"}) + + def raise_if_error(self, response: requests.Response) -> None: + """ + Method to raise an exception if an error occurs during the request + We raise a custom error if a JSON message is provided with an error + + :param response: the response object returned by the request + """ + if response.status_code == 401 and self._refresh_token: + self.refresh_session() + return + if response.status_code >= 300 and "json" in response.headers.get("content-type", ""): + raise IASOError(f"{response.json()}") + response.raise_for_status() + + @staticmethod + def decode_token_expiry(token: str) -> Union[datetime, None]: + """ + Decodes base64 encoded JWT token and returns expiry time from 'exp' field of the JWT token + + :param token: JWT token + + :return: Expiry datetime or None + + Examples: + >>> decode_token_expiry(token = "eyJhbGciOiJIUzI1NiJ9.eyJSb2xlIjoiQWRtaW4iLCJJc3N1ZXIiOiJJc3N1ZXIiLCJVc2VybmFt\\ + ZSI6IkphdmFJblVzZSIsImV4cCI6MTcxNzY5MDEwNCwiaWF0IjoxNzE3NzYwMTA0fQ._pXcqDw0QgvznvNuhVPwYyIms3H5imH-q6A7lIQJjYQ") + """ + decoded_token = jwt.decode(token, options={"verify_signature": False}) + exp_timestamp = decoded_token.get("exp") + if exp_timestamp: + return datetime.fromtimestamp(exp_timestamp, timezone.utc) + return None diff --git a/openhexa/toolbox/iaso/iaso.py b/openhexa/toolbox/iaso/iaso.py new file mode 100644 index 0000000..b902f47 --- /dev/null +++ b/openhexa/toolbox/iaso/iaso.py @@ -0,0 +1,126 @@ +import io +import typing + +from openhexa.toolbox.iaso.api_client import ApiClient +import polars as pl + + +class IASO: + """ + The IASO toolbox provides an interface to interact with the IASO. + """ + + def __init__(self, server_url: str, username: str, password: str) -> None: + """ + Initializes the IASO toolbox. + :param server_url: IASO server URL + :param username: IASO instance username + :param password: IASO instance password + + Examples: + >>> from openhexa.toolbox.iaso import IASO + >>> iaso = IASO(server_url="http://iaso-staging.bluesquare.org", + >>> username="user", + >>> password="pass") + """ + self.api_client = ApiClient(server_url, username, password) + + def get_projects(self, page: int = 0, limit: int = 10, **kwargs) -> dict: + """ + Fetches projects list from IASO. Method is paginated by default. Pagination can be modified and additional + arguments can be passed as key value parameters. + + Examples: + >>> from openhexa.toolbox.iaso import IASO + >>> iaso = IASO(client=ApiClient(url="http://iaso-staging.bluesquare.org", + >>> username="user", + >>> password="pass")) + >>> iaso.get_projects(page=1, limit=1, id=1) + """ + + params = kwargs + params.update({"page": page, "limit": limit}) + response = self.api_client.get("/api/projects", params=params) + return response.json().get("projects") + + def get_org_units(self, page: int = 0, limit: int = 10, **kwargs) -> dict: + """ + Fetches org units from IASO. Method is paginated by default. Pagination can be modified and additional + arguments can be passed as key value parameters. + + Examples: + >>> from openhexa.toolbox.iaso import IASO + >>> iaso = IASO(client=ApiClient(url="http://iaso-staging.bluesquare.org", + >>> username="user", + >>> password="pass")) + >>> projects = iaso.get_org_units(page=1, limit=1, id=1) + """ + params = kwargs + params.update({"page": page, "limit": limit}) + response = self.api_client.get("/api/orgunits", params=params) + return response.json().get("orgUnits") + + def get_form_instances( + self, + page: int = 0, + limit: int = 10, + as_dataframe: bool = False, + dataframe_columns: typing.List[str] = None, + **kwargs, + ) -> typing.Union[dict, pl.DataFrame]: + """ + Fetches form instances from IASO filtered by form id. Method is paginated by default. + Pagination can be modified and additional arguments can be passed as key value parameters. + There is a possiblity to fetch forms as DataFrames. + + Params: + :param page: The page number of the form instance. + :param limit: The maximum number of form instances. + :param as_dataframe: If true, will return a DataFrame containing form instances. + :param dataframe_columns: The column names of the form instances. + :param kwargs: additonal arguments passed to the /forms endpoint as URL parameters. + + Examples: + >>> from openhexa.toolbox.iaso import IASO + >>> iaso = IASO(url="http://iaso-staging.bluesquare.org", username="user", password="pass") + >>> form_dataframes = iaso.get_form_instances(page=1, limit=1, as_dataframe=True, + >>> dataframe_columns=["Date de création","Date de modification","Org unit"], ids=276) + """ + + params = kwargs + params.update({"page": page, "limit": limit}) + if as_dataframe: + params.update({"csv": "true"}) + response = self.api_client.get("/api/instances", params=params) + forms = pl.read_csv(io.StringIO(response.content.decode("utf-8")))[dataframe_columns] + return forms + response = self.api_client.get("/api/instances/", params=kwargs) + forms = response.json().get("instances") + return forms + + def get_forms( + self, org_units: typing.List[int], projects: typing.List[int], page: int = 0, limit: int = 10, **kwargs + ) -> dict: + """ + Fetches forms from IASO. Method is paginated by default. + Pagination can be modified and additional arguments can be passed as key value parameters. + + Params: + :param org_units: A required list of organization units IDs. + :param projects: A required list of project IDs. + :param page: The page number of the form. + :param limit: The maximum number of form. + :param kwargs: additonal arguments passed to the /forms endpoint as URL parameters. + + Examples: + >>> from openhexa.toolbox.iaso import IASO + >>> iaso = IASO(url="http://iaso-staging.bluesquare.org",username="user",password="pass") + >>> forms_by_orgunits_and_projects = iaso.get_forms(page=1, limit=1, org_units=[300], projects=[23]) + """ + + if org_units is [] or projects is []: + raise ValueError("Values for org_units and projects cannot be empty lists") + params = kwargs + params.update({"page": page, "limit": limit, "org_units": org_units, "projects": projects}) + response = self.api_client.post("/api/forms", data=params) + return response.json().get("forms") diff --git a/pyproject.toml b/pyproject.toml index e4cac15..1a342ea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,8 @@ dependencies = [ "geopandas", "polars", "diskcache", + "stringcase", + "pyjwt" ] [project.optional-dependencies] diff --git a/tests/iaso/fixtures/iaso_api_fixtures.py b/tests/iaso/fixtures/iaso_api_fixtures.py new file mode 100644 index 0000000..e668a7e --- /dev/null +++ b/tests/iaso/fixtures/iaso_api_fixtures.py @@ -0,0 +1,243 @@ +iaso_mocked_auth_token = { + "access": "eyJhbGciOiJIUzI1NiJ9.eyJSb2xlIjoiQWRtaW4iLCJJc3N1ZXIiOiJJc3N1ZXIiLCJVc2VybmFtZSI6IkphdmFJblVzZSIsImV4cCI6MTcxNzY5MDEwNCwiaWF0IjoxNzE3NjkwMTA0fQ.WsmnKvyKFR2eWNL4wD4yrnd6F9CDBV2dCaMx9lE6V84", # noqa: E501 + "refresh": "eyJhbGciOiJIUzI1NiJ9.eyJSb2xlIjoiQWRtaW4iLCJJc3N1ZXIiOiJJc3N1ZXIiLCJVc2VybmFtZSI6IkphdmFJblVzZSIsImV4cCI6MTcxNzY5MDEwNCwiaWF0IjoxNzE3NjkwMTA0fQ.WsmnKvyKFR2eWNL4wD4yrnd6F9CDBV2dCaMx9lE6V84", # noqa: E501 +} +iaso_mocked_refreshed_auth_token = { + "access": "eyJhbGciOiJIUzI1NiJ9.eyJSb2xlIjoiQWRtaW4iLCJJc3N1ZXIiOiJJc3N1ZXIiLCJVc2VybmFtZSI6IkphdmFJblVzZSIsImV4cCI6MTcxNzY5MDEwNCwiaWF0IjoxNzE3NzYwMTA0fQ._pXcqDw0QgvznvNuhVPwYyIms3H5imH-q6A7lIQJjYQ", # noqa: E501 + "refresh": "eyJhbGciOiJIUzI1NiJ9.eyJSb2xlIjoiQWRtaW4iLCJJc3N1ZXIiOiJJc3N1ZXIiLCJVc2VybmFtZSI6IkphdmFJblVzZSIsImV4cCI6MTcxNzY5MDEwNCwiaWF0IjoxNzE3NjkwMTA0fQ.WsmnKvyKFR2eWNL4wD4yrnd6F9CDBV2dCaMx9lE6V84", # noqa: E501 +} + + +iaso_mocked_forms = { + "forms": [ + { + "id": 278, + "name": "Test (form styling)", + "form_id": "pathways_indonesia_survey_1", + "device_field": "deviceid", + "location_field": "", + "org_unit_types": [{"id": 781, "name": "Province", "short_name": "Prov", "created_at": 1712825023.047433}], + "projects": [{"id": 149, "name": "Pathways"}], + "created_at": 1713171086.141424, + } + ] +} + +iaso_mocked_projects = { + "projects": [ + { + "id": 149, + "name": "Pathways", + "app_id": "pathways", + "feature_flags": [ + {"id": 3, "name": "GPS point for each form", "code": "TAKE_GPS_ON_FORM"}, + {"id": 7, "name": "Mobile: Show data collection screen", "code": "DATA_COLLECTION"}, + {"id": 12, "name": "Mobile: Finalized forms are read only", "code": "MOBILE_FINALIZED_FORM_ARE_READ"}, + {"id": 4, "name": "Authentication", "code": "REQUIRE_AUTHENTICATION"}, + ], + "created_at": 1710153966.532745, + "updated_at": 1717664805.185712, + "needs_authentication": True, + } + ] +} + +iaso_mocked_orgunits = { + "orgUnits": [ + { + "name": "ACEH", + "id": 1978297, + "parent_id": 1978331, + "org_unit_type_id": 781, + "org_unit_type_name": "Province", + "validation_status": "VALID", + "created_at": 1712825023.085615, + "updated_at": 1712828860.665764, + } + ] +} + +iaso_mocked_instances = { + "count": 28, + "instances": [ + { + "uuid": "4dc8c051-5b86-4b8e-b767-137e017d3c07", + "export_id": "ArtNWdBsfAj", + "file_name": "276_513ae6cf-5efa-4754-b7fb-2d2124ef3efd_2024-04-11_14-17-05.xml", + "file_content": { + "age": "14", + "name": "qfdsf", + "U1_u1": "U1", + "address": "qsdfqsdf", + "endtime": "2024-04-11T14:18:46.435+02:00", + "village": "dddddd", + "_version": "6004241215", + "deviceid": "853e64340cfa95b8", + "district": "badung", + "location": "urban", + "religion": "protestant", + "username": "", + "ethnicity": "sasak", + "pregnancy": "yes", + "starttime": "2024-04-11T14:17:05.394+02:00", + "time_slot": "14_to_16", + "instanceID": "uuid:676558f6-17f2-4a4f-a8f0-08b43cc572da", + "occupation": "ffffffff", + "subdistrict": "dddd", + "children_age": "12,12", + "partner_name": "", + "phone_number": "111", + "type_of_work": "formal", + "have_children": "yes", + "type_of_phone": "feature_phone", + "devicephonenum": "", + "final_comments": "thanks", + "jkn_mobile_app": "no", + "marital_status": "single", + "preferred_date": "2024-04-11", + "access_to_phone": "yes", + "number_of_children": "1", + "type_jnk_insurance": "pbi", + "living_with_partner": "no", + "enrolled_jnk_insurance": "yes", + "person_with_disability": "yes", + "education_participant_1": "no_education", + "screening_questions_intro": "", + "preferred_language_interview": "bahasa_daerah", + }, + "file_url": "https://iaso-stg.s3.amazonaws.com/instances/276_513ae6cf-5efa-4754-b7fb-2d2124ef3efd_2024-04-11_14-17-05.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIAS4KZU3S6DZBAV5GW%2F20240611%2Feu-central-1%2Fs3%2Faws4_request&X-Amz-Date=20240611T075045Z&X-Amz-Expires=3600&X-Amz-SignedHeaders=host&X-Amz-Signature=7ebdbe2c44d8251589b31596aa61cf488fae5970a64a76084804b7c7cb9356c7", + "id": 36627, + "form_id": 276, + "form_name": "Pathways Indonesia Recruitment (test)", + "created_at": 1712837926.0, + "updated_at": 1712837933.562898, + "org_unit": { + "name": "ACEH", + "short_name": "ACEH", + "id": 1978297, + "source": "Indonesia", + "source_id": 192, + "source_ref": "ID001001000000000000", + "parent_id": 1978331, + "org_unit_type_id": 781, + "org_unit_type_name": "Province", + "org_unit_type_depth": 1, + "created_at": 1712825023.085615, + "updated_at": 1712828860.665764, + "aliases": None, + "validation_status": "VALID", + "latitude": None, + "longitude": None, + "altitude": None, + "has_geo_json": True, + "version": 0, + "opening_date": None, + "closed_date": None, + }, + "latitude": 50.8267504, + "longitude": 4.3505693, + "altitude": 113.30000305175781, + "period": None, + "status": "READY", + "correlation_id": 36627503, + "created_by": {"username": "mdewulf-pathways", "first_name": "Martin", "last_name": "De Wulf"}, + "last_modified_by": "mdewulf-pathways", + "can_user_modify": True, + "is_locked": False, + "is_instance_of_reference_form": False, + "is_reference_instance": False, + }, + { + "uuid": "d2fc1f2f-ed83-47db-b4db-be2679a26aff", + "export_id": "pNI8aE6Am2Z", + "file_name": "276_1770f9ff-254f-43af-9c13-21679a12a3d6_2024-04-12_11-47-29.xml", + "file_content": { + "age": "30", + "name": "Test 1", + "R1_r1": "R1", + "address": "Test 1", + "endtime": "2024-04-12T11:51:29.661+03:00", + "village": "Test 1", + "_version": "6004241215", + "deviceid": "cb5f71d5c9b0f248", + "district": "pidie", + "location": "rural", + "religion": "protestant", + "username": "", + "ethnicity": "aceh", + "pregnancy": "no", + "starttime": "2024-04-12T11:47:29.176+03:00", + "time_slot": "12_to_14", + "instanceID": "uuid:3b07bced-1155-4a3a-9324-189653cac25e", + "occupation": "Test 1", + "subdistrict": "Test 1", + "partner_name": "", + "phone_number": "235866358", + "type_of_work": "informal", + "have_children": "no", + "type_of_phone": "smartphone", + "devicephonenum": "", + "final_comments": "Test 1", + "jkn_mobile_app": "yes", + "marital_status": "married_partner", + "preferred_date": "2024-04-12", + "access_to_phone": "yes", + "jkn_mobile_app_use": "sometimes_use_it", + "type_jnk_insurance": "pbi", + "living_with_partner": "no", + "enrolled_jnk_insurance": "yes", + "person_with_disability": "no", + "education_participant_1": "some_education", + "hh_member_with_disability": "no", + "screening_questions_intro": "", + "preferred_language_interview": "bahasa_indonesia", + }, + "file_url": "https://iaso-stg.s3.amazonaws.com/instances/276_1770f9ff-254f-43af-9c13-21679a12a3d6_2024-04-12_11-47-29.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIAS4KZU3S6DZBAV5GW%2F20240611%2Feu-central-1%2Fs3%2Faws4_request&X-Amz-Date=20240611T075045Z&X-Amz-Expires=3600&X-Amz-SignedHeaders=host&X-Amz-Signature=05eb2a925b34dd620cf35b27a0b9850f0dc4f0351e96a4c41ad4d82da58f11e6", + "id": 36628, + "form_id": 276, + "form_name": "Pathways Indonesia Recruitment (test)", + "created_at": 1712911889.0, + "updated_at": 1712911890.646247, + "org_unit": { + "name": "ACEH", + "short_name": "ACEH", + "id": 1978297, + "source": "Indonesia", + "source_id": 192, + "source_ref": "ID001001000000000000", + "parent_id": 1978331, + "org_unit_type_id": 781, + "org_unit_type_name": "Province", + "org_unit_type_depth": 1, + "created_at": 1712825023.085615, + "updated_at": 1712828860.665764, + "aliases": None, + "validation_status": "VALID", + "latitude": None, + "longitude": None, + "altitude": None, + "has_geo_json": True, + "version": 0, + "opening_date": None, + "closed_date": None, + }, + "latitude": 60.1916042, + "longitude": 24.9458922, + "altitude": 46.978047416201505, + "period": None, + "status": "READY", + "correlation_id": 36628816, + "created_by": None, + "last_modified_by": None, + "can_user_modify": True, + "is_locked": False, + "is_instance_of_reference_form": False, + "is_reference_instance": False, + }, + ], + "has_next": True, + "has_previous": False, + "page": 1, + "pages": 14, + "limit": 2, +} diff --git a/tests/iaso/test_iaso.py b/tests/iaso/test_iaso.py new file mode 100644 index 0000000..931b1a4 --- /dev/null +++ b/tests/iaso/test_iaso.py @@ -0,0 +1,111 @@ +import pytest +import responses + +from openhexa.toolbox.iaso import IASO +from openhexa.toolbox.iaso.api_client import ApiClient +from tests.iaso.fixtures.iaso_api_fixtures import ( + iaso_mocked_auth_token, + iaso_mocked_forms, + iaso_mocked_orgunits, + iaso_mocked_refreshed_auth_token, + iaso_mocked_projects, + iaso_mocked_instances, +) + + +class TestIasoAPI: + @pytest.fixture + def mock_responses(self): + with responses.RequestsMock() as rsps: + yield rsps + + def test_authenticate(self, mock_responses): + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/token/", json=iaso_mocked_auth_token, status=200 + ) + + iaso_api_client = ApiClient("https://iaso-staging.bluesquare.org", "username", "password") + iaso_api_client.authenticate() + assert iaso_api_client.token == iaso_mocked_auth_token["access"] + + def test_get_projects(self, mock_responses): + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/token/", json=iaso_mocked_auth_token, status=200 + ) + mock_responses.add( + responses.GET, "https://iaso-staging.bluesquare.org/api/projects/", json=iaso_mocked_projects, status=200 + ) + + iaso = IASO("https://iaso-staging.bluesquare.org", "username", "password") + r = iaso.get_projects() + assert len(r) > 0 + + def test_get_org_units(self, mock_responses): + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/token/", json=iaso_mocked_auth_token, status=200 + ) + mock_responses.add( + responses.GET, "https://iaso-staging.bluesquare.org/api/orgunits/", json=iaso_mocked_orgunits, status=200 + ) + iaso = IASO("https://iaso-staging.bluesquare.org", "username", "password") + r = iaso.get_org_units() + assert len(r) > 0 + + def test_get_forms(self, mock_responses): + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/token/", json=iaso_mocked_auth_token, status=200 + ) + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/forms/", json=iaso_mocked_forms, status=200 + ) + iaso = IASO("https://iaso-staging.bluesquare.org", "username", "password") + r = iaso.get_forms(org_units=[781], projects=[149]) + assert len(r) > 0 + + def test_get_form_instances(self, mock_responses): + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/token/", json=iaso_mocked_auth_token, status=200 + ) + mock_responses.add( + responses.GET, "https://iaso-staging.bluesquare.org/api/instances/", json=iaso_mocked_instances, status=200 + ) + iaso = IASO("https://iaso-staging.bluesquare.org", "user", "test") + form_instances = iaso.get_form_instances(form_ids=276) + assert len(form_instances) > 0 + + def test_failing_forms(self, mock_responses): + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/token/", json=iaso_mocked_auth_token, status=200 + ) + mock_responses.add( + responses.POST, + "https://iaso-staging.bluesquare.org/api/forms/", + json={"message": "Form submission failed"}, + status=500, + ) + iaso = IASO("https://iaso-staging.bluesquare.org", "user", "test") + try: + iaso.get_forms([781], [149]) + except Exception as e: + assert str(e) == "{'message': 'Form submission failed'}" + + def test_verify_expired_token(self, mock_responses): + mock_responses.add( + responses.POST, "https://iaso-staging.bluesquare.org/api/token/", json=iaso_mocked_auth_token, status=200 + ) + mock_responses.add( + responses.POST, + "https://iaso-staging.bluesquare.org/api/forms/", + json={"message": "No authorized"}, + status=401, + ) + mock_responses.add( + responses.POST, + "https://iaso-staging.bluesquare.org/api/token/refresh/", + json=iaso_mocked_refreshed_auth_token, + status=200, + ) + iaso = IASO("https://iaso-staging.bluesquare.org", "user", "test") + iaso.get_forms([781], [149]) + assert mock_responses.calls[2].request.url == "https://iaso-staging.bluesquare.org/api/token/refresh/" + assert iaso.api_client.token == iaso_mocked_refreshed_auth_token["access"] diff --git a/tests/test_lib.py b/tests/test_lib.py deleted file mode 100644 index 2de7843..0000000 --- a/tests/test_lib.py +++ /dev/null @@ -1,6 +0,0 @@ -import unittest - - -class ToolboxTest(unittest.TestCase): - def test_true(self): - self.assertTrue(True)