Add more endpoints and checks

- add new organization and user endpoints
  - get a user's profile
  - get an NGO user's organization's profile
  - get the applications of an organization's user
  - get the list of apps
  - get organization information
  - get user(s) information
- checks for users making use of these endpoints
  - check if a user has an application (self and other users)
  - check if an organization has an application

pyproject.toml

@@ -65,7 +65,6 @@ addopts = "-ra -q"
 testpaths = [
   "tests",
 ]
-env_files = [
-  ".env",
-  ".env.test",
+markers = [
+  "investigation_needed: mark test as needing investigation (deselect with '-m \"not investigation_needed\"')",
 ]

src/ngohub/core.py

@@ -1,6 +1,8 @@
 from abc import ABC, abstractmethod
-from typing import Any, Dict, List
+from copy import deepcopy
+from typing import Any, Dict, List, Union
 
+from ngohub.exceptions import HubHTTPException, MissingUserException
 from ngohub.network import HTTPClient, HTTPClientResponse
 
 
@@ -11,12 +13,13 @@ class BaseHub(ABC):
 
     @abstractmethod
     def __init__(self, api_base_url: str) -> None:
-        pass
+        self.api_base_url: str = api_base_url or ""
 
 
 class NGOHub(BaseHub):
     def __init__(self, api_base_url: str) -> None:
-        self.api_base_url: str = api_base_url or ""
+        super().__init__(api_base_url)
+
         self.client: HTTPClient = HTTPClient(self.api_base_url)
 
     def is_healthy(self) -> bool:
@@ -96,3 +99,181 @@ def get_beneficiaries_nomenclatures(self):
 
     def get_issuers_nomenclatures(self):
         return self._get_nomenclature("issuers")
+
+    # User related methods
+    def get_profile(self, user_token: str) -> Dict[str, Any]:
+        response: HTTPClientResponse = self.client.api_get("/profile/", token=user_token)
+
+        return response.to_dict()
+
+    # Organization related methods
+    def get_organization_profile(self, ngo_token: str) -> Dict[str, Any]:
+        response: HTTPClientResponse = self.client.api_get("/organization-profile/", token=ngo_token)
+
+        return response.to_dict()
+
+    def get_user_organization_applications(self, ngo_token: str) -> List[Dict[str, Any]]:
+        response: HTTPClientResponse = self.client.api_get("/organization/applications/", token=ngo_token)
+
+        return list(response.to_dict())
+
+    def check_user_organization_has_application(self, ngo_token: str, login_link: str) -> Dict[str, Any]:
+        organization_applications: List[Dict[str, Any]] = self.get_user_organization_applications(ngo_token)
+
+        for app in organization_applications:
+            if app["loginLink"].startswith(login_link) and app["status"] == "active" and app["ongStatus"] == "active":
+                return app
+
+        return {}
+
+    # Admin related methods
+    def get_application_list(self, admin_token: str) -> List[Dict[str, Any]]:
+        response: HTTPClientResponse = self.client.api_get("/application/list/", token=admin_token)
+
+        return list(response.to_dict())
+
+    def get_organization(self, admin_token: str, organization_id: int) -> Dict[str, Any]:
+        response: HTTPClientResponse = self.client.api_get(f"/organization/{organization_id}/", token=admin_token)
+
+        return response.to_dict()
+
+    def get_organization_applications(self, admin_token: str, organization_id: int) -> List[Dict[str, Any]]:
+        response: HTTPClientResponse = self.client.api_get(
+            f"/application/organization/{organization_id}/", token=admin_token
+        )
+
+        return list(response.to_dict())
+
+    def get_user(self, admin_token: str, user_id: int) -> Dict[str, Any]:
+        try:
+            response: HTTPClientResponse = self.client.api_get(f"/user/{user_id}/", token=admin_token)
+        except HubHTTPException as e:
+            if e.status_code == 404:
+                raise MissingUserException(f"User with ID {user_id} not found")
+
+            raise e
+
+        return response.to_dict()
+
+    def get_users(
+        self,
+        admin_token: str,
+        organization_id: int,
+        limit: int = 1000,
+        page: int = 1,
+        search: str = None,
+        order_by: str = None,
+        order_direction: str = None,
+        start: str = None,
+        end: str = None,
+        status: str = None,
+        available_apps_ids: List[int] = None,
+    ) -> Dict[str, Any]:
+        request_url: str = f"/user?organization_id={organization_id}&limit={limit}&page={page}"
+        if search:
+            request_url += f"&search={search}"
+        if order_by:
+            request_url += f"&orderBy={order_by}"
+        if order_direction and order_direction.upper() in ["ASC", "DESC"]:
+            request_url += f"&orderDirection={order_direction.upper()}"
+        if start:
+            request_url += f"&start={start}"
+        if end:
+            request_url += f"&end={end}"
+        if status and status.lower() in ["active", "pending", "restricted"]:
+            request_url += f"&status={status}"
+        if available_apps_ids:
+            for app_id in available_apps_ids:
+                request_url += f"&availableAppsIds={app_id}"
+
+        response: HTTPClientResponse = self.client.api_get(request_url, token=admin_token)
+
+        return response.to_dict()
+
+    def check_organization_has_application(
+        self, admin_token: str, organization_id: int, login_link: str
+    ) -> Dict[str, Any]:
+        organization_applications: List[Dict[str, Any]] = self.get_organization_applications(
+            admin_token, organization_id
+        )
+
+        for application in organization_applications:
+            if (
+                application["loginLink"].startswith(login_link)
+                and application["status"] == "active"
+                and application["ongStatus"] == "active"
+            ):
+                return application
+
+        return {}
+
+    def _check_user_has_application(self, admin_token, organization_id, user_id, response) -> Dict[str, Any]:
+        continue_searching: bool = True
+        page: int = 1
+
+        response: Dict[str, Any] = deepcopy(response)
+        searched_application_id = response["application"]["id"]
+
+        while continue_searching:
+            organization_users: Dict[str, Union[Dict, List]] = self.get_users(
+                admin_token=admin_token, organization_id=organization_id, page=page
+            )
+
+            response_metadata = organization_users.get("meta", {})
+            response_users = organization_users.get("items", [])
+
+            if response_metadata["totalPages"] <= response_metadata["currentPage"]:
+                continue_searching = False
+            else:
+                page += 1
+
+            for user in response_users:
+                if user["id"] == int(user_id):
+                    response["user"] = user
+                    user_applications: List[Dict[str, Any]] = user["availableAppsIDs"]
+
+                    if searched_application_id in user_applications:
+                        response["access"] = True
+                        return response
+
+                    response["access"] = False
+                    return response
+
+        return response
+
+    def check_organization_user_has_application(
+        self,
+        admin_token: str,
+        organization_id: int,
+        user_id: int,
+        login_link: str,
+    ) -> Dict[str, Any]:
+
+        response: Dict[str, Any] = {"user": None, "application": None, "access": None}
+
+        organization_application: Dict = self.check_organization_has_application(
+            admin_token, organization_id, login_link
+        )
+        if not organization_application:
+            return response
+
+        response["application"] = organization_application
+
+        try:
+            user_information: Dict = self.get_user(admin_token, user_id)
+        except MissingUserException:
+            return response
+
+        if not user_information or user_information.get("organizationId") != int(organization_id):
+            return response
+
+        response["user"] = user_information
+
+        if user_information.get("role") == "admin":
+            response["access"] = True
+
+            return response
+
+        response = self._check_user_has_application(admin_token, organization_id, user_id, response)
+
+        return response

src/ngohub/network.py

@@ -6,7 +6,7 @@
 from http.client import HTTPResponse, HTTPSConnection
 from typing import Dict, Optional
 
-from ngohub.exceptions import HubDecodeException, HubHTTPException
+from ngohub.exceptions import HubBadRequestException, HubDecodeException, HubHTTPException
 
 
 logger = logging.getLogger(__name__)
@@ -80,16 +80,27 @@ def _api_request(
         try:
             conn.request(method=request_method, url=path, body=encoded_params, headers=headers)
         except socket.gaierror as e:
-            raise HubHTTPException(f"Failed to make request to '{path}': {e}")
+            raise HubBadRequestException(
+                message=f"Failed to make request to '{path}': {e}",
+                path=path,
+            )
 
         try:
             response: HTTPResponse = conn.getresponse()
         except ConnectionError as e:
-            raise HubHTTPException(f"Failed to get response from '{path}': {e}")
+            raise HubBadRequestException(
+                message=f"Failed to get response from '{path}': {e}",
+                path=path,
+            )
 
         if response.status != http.HTTPStatus.OK:
             logger.info(path)
-            raise HubHTTPException(f"{response.status} while retrieving '{path}'. Reason: {response.reason}")
+            raise HubHTTPException(
+                message=f"{response.status} while retrieving '{path}'. Reason: {response.reason}",
+                status_code=response.status,
+                path=path,
+                reason=response.reason,
+            )
 
         return HTTPClientResponse(response)
 

tests/conftest.py

@@ -2,16 +2,66 @@
 
 import pytest
 from dotenv import find_dotenv, load_dotenv
+from pycognito import Cognito
 
 
 def pytest_configure():
-    load_dotenv(find_dotenv(filename=".env.test"))
+    if not os.environ.get("SKIP_TEST_ENV_FILE"):
+        load_dotenv(find_dotenv(filename=".env.test"))
 
-    pytest.ngohub_api_url = os.environ.get("NGO_HUB_API_URL")
+    pytest.ngohub_api_url = os.environ["NGO_HUB_API_URL"]
 
+    pytest.app_login_link = os.environ["APP_LOGIN_LINK"]
 
-def ngohub_url():
-    return os.environ.get("NGO_HUB_API_URL")
+    pytest.ngo_id = os.environ["NGO_ID"]
+    pytest.ngo_admin_id = os.environ["NGO_ADMIN_ID"]
+    pytest.ngo_user_id = os.environ["NGO_USER_ID"]
+    pytest.ngo_user_id_no_app = os.environ["NGO_USER_ID_NO_APP"]
+
+    pytest.ngo_other_id = os.environ["NGO_OTHER_ID"]
+    pytest.ngo_other_admin_id = os.environ["NGO_OTHER_ADMIN_ID"]
+    pytest.ngo_other_user_id = os.environ["NGO_OTHER_USER_ID"]
+
+    pytest.admin_username = os.environ["NGOHUB_API_ACCOUNT"]
+    pytest.ngo_admin_username = os.environ["NGOHUB_NGO_ACCOUNT"]
+
+    pytest.admin_token = _authenticate_with_ngohub(
+        user_pool_id=os.environ["AWS_COGNITO_USER_POOL_ID"],
+        client_id=os.environ["AWS_COGNITO_CLIENT_ID"],
+        client_secret=os.environ["AWS_COGNITO_CLIENT_SECRET"],
+        user_pool_region=os.environ["AWS_COGNITO_REGION"],
+        username=os.environ["NGOHUB_API_ACCOUNT"],
+        api_key=os.environ["NGOHUB_API_KEY"],
+    )
+
+    pytest.ngo_admin_token = _authenticate_with_ngohub(
+        user_pool_id=os.environ["AWS_COGNITO_USER_POOL_ID"],
+        client_id=os.environ["AWS_COGNITO_CLIENT_ID"],
+        client_secret=os.environ["AWS_COGNITO_CLIENT_SECRET"],
+        user_pool_region=os.environ["AWS_COGNITO_REGION"],
+        username=os.environ["NGOHUB_NGO_ACCOUNT"],
+        api_key=os.environ["NGOHUB_NGO_KEY"],
+    )
+
+
+def _authenticate_with_ngohub(
+    user_pool_id: str,
+    client_id: str,
+    client_secret: str,
+    user_pool_region: str,
+    username: str,
+    api_key: str,
+) -> str:
+    u = Cognito(
+        user_pool_id=user_pool_id,
+        client_id=client_id,
+        client_secret=client_secret,
+        username=username,
+        user_pool_region=user_pool_region,
+    )
+    u.authenticate(password=api_key)
+
+    return u.id_token
 
 
 class FakeHTTPSConnection: