|
1 | | -import base64 |
2 | 1 | import logging |
3 | 2 | import os |
4 | 3 | import random |
5 | 4 | import time |
6 | 5 | import urllib.parse |
7 | | -import uuid |
8 | 6 |
|
9 | | -import jwt |
10 | 7 | import pytest |
11 | 8 | import requests |
12 | 9 | from faker import Faker |
13 | 10 |
|
14 | 11 | from mavis.test.constants import Programme |
| 12 | +from mavis.test.data.file_generator import FileGenerator |
15 | 13 | from mavis.test.data_models import ( |
16 | 14 | Child, |
17 | 15 | Clinic, |
|
30 | 28 | onboarding_faker.unique.clear() |
31 | 29 |
|
32 | 30 |
|
| 31 | +@pytest.fixture(scope="session", autouse=True) |
| 32 | +def delete_team_after_tests(base_url, team): |
| 33 | + yield |
| 34 | + |
| 35 | + url = urllib.parse.urljoin(base_url, f"api/testing/teams/{team.workgroup}") |
| 36 | + response = requests.delete(url, timeout=30) |
| 37 | + _check_response_status(response) |
| 38 | + |
| 39 | + |
| 40 | +@pytest.fixture(scope="module", autouse=True) |
| 41 | +def reset_before_each_module(base_url, team) -> None: |
| 42 | + url = urllib.parse.urljoin(base_url, f"api/testing/teams/{team.workgroup}") |
| 43 | + response = requests.delete(url, params={"keep_itself": "true"}, timeout=30) |
| 44 | + _check_response_status(response) |
| 45 | + |
| 46 | + |
33 | 47 | @pytest.fixture(scope="session") |
34 | 48 | def year_groups() -> dict[str, int]: |
35 | 49 | return { |
@@ -84,22 +98,6 @@ def _check_response_status(response) -> None: |
84 | 98 | response.raise_for_status() |
85 | 99 |
|
86 | 100 |
|
87 | | -@pytest.fixture(scope="session", autouse=True) |
88 | | -def delete_team_after_tests(base_url, team): |
89 | | - yield |
90 | | - |
91 | | - url = urllib.parse.urljoin(base_url, f"api/testing/teams/{team.workgroup}") |
92 | | - response = requests.delete(url, timeout=30) |
93 | | - _check_response_status(response) |
94 | | - |
95 | | - |
96 | | -@pytest.fixture(scope="module", autouse=True) |
97 | | -def reset_before_each_module(base_url, team) -> None: |
98 | | - url = urllib.parse.urljoin(base_url, f"api/testing/teams/{team.workgroup}") |
99 | | - response = requests.delete(url, params={"keep_itself": "true"}, timeout=30) |
100 | | - _check_response_status(response) |
101 | | - |
102 | | - |
103 | 101 | @pytest.fixture(scope="session") |
104 | 102 | def healthcare_assistant(onboarding) -> User: |
105 | 103 | return onboarding.users["healthcare_assistant"] |
@@ -157,57 +155,6 @@ def children(year_groups) -> dict[str, list[Child]]: |
157 | 155 | ) |
158 | 156 |
|
159 | 157 |
|
160 | | -def _read_imms_api_credentials() -> dict[str, str]: |
161 | | - return { |
162 | | - "pem": os.environ["IMMS_API_PEM"], |
163 | | - "key": os.environ["IMMS_API_KEY"], |
164 | | - "kid": os.environ["IMMS_API_KID"], |
165 | | - "url": os.environ["IMMS_BASE_URL"], |
166 | | - } |
167 | | - |
168 | | - |
169 | | -def _get_jwt_payload(api_auth: dict[str, str]) -> str: |
170 | | - _kid = api_auth["kid"] |
171 | | - _api_key = api_auth["key"] |
172 | | - _decoded_pem = base64.b64decode(api_auth["pem"]) |
173 | | - _auth_endpoint = urllib.parse.urljoin(api_auth["url"], "oauth2-mock/token") |
174 | | - headers = { |
175 | | - "alg": "RS512", |
176 | | - "typ": "JWT", |
177 | | - "kid": _kid, |
178 | | - } |
179 | | - claims = { |
180 | | - "sub": _api_key, |
181 | | - "iss": _api_key, |
182 | | - "jti": str(uuid.uuid4()), |
183 | | - "aud": _auth_endpoint, |
184 | | - "exp": int(time.time()) + 300, # 5mins in the future |
185 | | - } |
186 | | - return jwt.encode( |
187 | | - payload=claims, |
188 | | - key=_decoded_pem, |
189 | | - algorithm="RS512", |
190 | | - headers=headers, |
191 | | - ) |
192 | | - |
193 | | - |
194 | | -@pytest.fixture(scope="session", autouse=False) |
195 | | -def authenticate_api(): |
196 | | - _api_auth: dict[str, str] = _read_imms_api_credentials() |
197 | | - _endpoint = urllib.parse.urljoin(_api_auth["url"], "oauth2-mock/token") |
198 | | - _payload = { |
199 | | - "grant_type": "client_credentials", |
200 | | - "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer", # noqa: E501 |
201 | | - "client_assertion": _get_jwt_payload(api_auth=_api_auth), |
202 | | - } |
203 | | - _headers = {"Content-Type": "application/x-www-form-urlencoded"} |
204 | | - |
205 | | - response = requests.post(url=_endpoint, headers=_headers, data=_payload, timeout=30) |
206 | | - |
207 | | - _check_response_status(response=response) |
208 | | - return response.json()["access_token"] |
209 | | - |
210 | | - |
211 | | -@pytest.fixture(scope="session") |
212 | | -def imms_base_url(): |
213 | | - return os.environ["IMMS_AUTH_URL"] |
| 158 | +@pytest.fixture |
| 159 | +def file_generator(organisation, schools, nurse, children, clinics, year_groups): |
| 160 | + return FileGenerator(organisation, schools, nurse, children, clinics, year_groups) |
0 commit comments