From bdef0ae2c689e18179c9f75bd31af9e761d7fb0a Mon Sep 17 00:00:00 2001 From: Bryann Valderrama Date: Wed, 30 Oct 2024 13:57:29 -0500 Subject: [PATCH] fix: move integration test utils to separate file --- .../api/v1/tests/integration/test_views.py | 2 +- eox_core/test_utils.py | 104 ----------------- eox_core/tests/integration/utils.py | 107 ++++++++++++++++++ 3 files changed, 108 insertions(+), 105 deletions(-) create mode 100644 eox_core/tests/integration/utils.py diff --git a/eox_core/api/v1/tests/integration/test_views.py b/eox_core/api/v1/tests/integration/test_views.py index dea3561c7..f87825ddb 100644 --- a/eox_core/api/v1/tests/integration/test_views.py +++ b/eox_core/api/v1/tests/integration/test_views.py @@ -14,7 +14,7 @@ from rest_framework import status from eox_core.api.v1.tests.integration.data.fake_users import FAKE_USER_DATA -from eox_core.test_utils import BaseIntegrationTest, make_request +from eox_core.tests.integration.utils import BaseIntegrationTest, make_request settings = ds.INTEGRATION_TEST_SETTINGS diff --git a/eox_core/test_utils.py b/eox_core/test_utils.py index 0b9f1c4aa..04530b425 100644 --- a/eox_core/test_utils.py +++ b/eox_core/test_utils.py @@ -1,16 +1,10 @@ """ Utils for testing""" -from __future__ import annotations - from datetime import datetime import factory -import requests -from django.conf import settings as ds from django.contrib.auth.models import User -from django.test import TestCase DEFAULT_PASSWORD = 'test' -settings = ds.INTEGRATION_TEST_SETTINGS class SuperUserFactory(factory.django.DjangoModelFactory): @@ -46,101 +40,3 @@ def url(self, name): return the name of the asset """ return name - - -def get_access_token() -> str: - """ - Get an access token for all requests in the test suite. - - Returns: - str: The access token. - """ - data = { - "client_id": settings["CLIENT_ID"], - "client_secret": settings["CLIENT_SECRET"], - "grant_type": "client_credentials", - } - url = f"http://{settings['LMS_BASE']}/oauth2/access_token/" - response = requests.post(url, data=data, timeout=settings["API_TIMEOUT"]) - return response.json()["access_token"] - - -ACCESS_TOKEN = get_access_token() - - -# pylint: disable=too-many-arguments -def make_request( - tenant: dict, - method: str, - url: str, - json: dict | None = None, - data: dict | None = None, - params: dict | None = None, - with_auth: bool = True, -) -> requests.Response: - """ - Make a request to a site (default site or tenant). - - Args: - tenant (dict): The tenant data. - method (str): The HTTP method ('GET', 'POST', etc.). - url (str): The URL to make the request to. - json (dict, optional): The JSON data for POST, PATCH and PUT requests. - data (dict, optional): The data for POST, PATCH and PUT requests. - params (dict, optional): The parameters for GET and DELETE requests. - with_auth (bool, optional): Whether to include the access token in the request headers. - - Returns: - requests.Response: The response object. - """ - headers = {"Host": tenant["domain"]} - if with_auth: - headers["Authorization"] = f"Bearer {ACCESS_TOKEN}" - full_url = f"{tenant['base_url']}/{url}" - - method = method.upper() - if method not in ("GET", "POST", "PATCH", "PUT", "DELETE"): - raise ValueError(f"Unsupported HTTP method: {method}.") - - return requests.request( - method, - full_url, - json=json, - data=data, - params=params, - headers=headers, - timeout=settings["API_TIMEOUT"], - ) - - -class BaseIntegrationTest(TestCase): - """ - Base class for the integration test suite. - """ - - def setUp(self): - """ - Set up the test suite. - """ - self.default_site = self.get_tenant_data() - self.tenant_x = self.get_tenant_data("tenant-x") - self.tenant_y = self.get_tenant_data("tenant-y") - self.demo_course_id = settings["DEMO_COURSE_ID"] - - def get_tenant_data(self, prefix: str = "") -> dict: - """ - Get the tenant data. - - If no prefix is provided, the default site data is returned. - - Args: - prefix (str, optional): The tenant prefix. Defaults to "". - - Returns: - dict: The tenant data. - """ - domain = f"{prefix}.{settings['LMS_BASE']}" if prefix else settings["LMS_BASE"] - return { - "base_url": f"http://{domain}", - "domain": domain, - } diff --git a/eox_core/tests/integration/utils.py b/eox_core/tests/integration/utils.py new file mode 100644 index 000000000..9c9710ff8 --- /dev/null +++ b/eox_core/tests/integration/utils.py @@ -0,0 +1,107 @@ +"""Utilities for the integration test suite.""" + +from __future__ import annotations + +import requests +from django.conf import settings as ds +from django.test import TestCase + +settings = ds.INTEGRATION_TEST_SETTINGS + + +def get_access_token() -> str: + """ + Get an access token for all requests in the test suite. + + Returns: + str: The access token. + """ + data = { + "client_id": settings["CLIENT_ID"], + "client_secret": settings["CLIENT_SECRET"], + "grant_type": "client_credentials", + } + url = f"http://{settings['LMS_BASE']}/oauth2/access_token/" + response = requests.post(url, data=data, timeout=settings["API_TIMEOUT"]) + return response.json()["access_token"] + + +ACCESS_TOKEN = get_access_token() + + +# pylint: disable=too-many-arguments +def make_request( + tenant: dict, + method: str, + url: str, + json: dict | None = None, + data: dict | None = None, + params: dict | None = None, + with_auth: bool = True, +) -> requests.Response: + """ + Make a request to a site (default site or tenant). + + Args: + tenant (dict): The tenant data. + method (str): The HTTP method ('GET', 'POST', etc.). + url (str): The URL to make the request to. + json (dict, optional): The JSON data for POST, PATCH and PUT requests. + data (dict, optional): The data for POST, PATCH and PUT requests. + params (dict, optional): The parameters for GET and DELETE requests. + with_auth (bool, optional): Whether to include the access token in the request headers. + + Returns: + requests.Response: The response object. + """ + headers = {"Host": tenant["domain"]} + if with_auth: + headers["Authorization"] = f"Bearer {ACCESS_TOKEN}" + full_url = f"{tenant['base_url']}/{url}" + + method = method.upper() + if method not in ("GET", "POST", "PATCH", "PUT", "DELETE"): + raise ValueError(f"Unsupported HTTP method: {method}.") + + return requests.request( + method, + full_url, + json=json, + data=data, + params=params, + headers=headers, + timeout=settings["API_TIMEOUT"], + ) + + +class BaseIntegrationTest(TestCase): + """ + Base class for the integration test suite. + """ + + def setUp(self): + """ + Set up the test suite. + """ + self.default_site = self.get_tenant_data() + self.tenant_x = self.get_tenant_data("tenant-x") + self.tenant_y = self.get_tenant_data("tenant-y") + self.demo_course_id = settings["DEMO_COURSE_ID"] + + def get_tenant_data(self, prefix: str = "") -> dict: + """ + Get the tenant data. + + If no prefix is provided, the default site data is returned. + + Args: + prefix (str, optional): The tenant prefix. Defaults to "". + + Returns: + dict: The tenant data. + """ + domain = f"{prefix}.{settings['LMS_BASE']}" if prefix else settings["LMS_BASE"] + return { + "base_url": f"http://{domain}", + "domain": domain, + }