|
1 | | -# import pytest |
2 | | -# from api_test_utils.oauth_helper import OauthHelper |
3 | | -# from api_test_utils.apigee_api_apps import ApigeeApiDeveloperApps |
4 | | -# from api_test_utils.apigee_api_products import ApigeeApiProducts |
5 | | -# |
6 | | -# |
7 | | -# class TestEndpoints: |
8 | | -# |
9 | | -# @pytest.fixture() |
10 | | -# def app(self): |
11 | | -# """ |
12 | | -# Import the test utils module to be able to: |
13 | | -# - Create apigee test application |
14 | | -# - Update custom attributes |
15 | | -# - Update custom ratelimits |
16 | | -# - Update products to the test application |
17 | | -# """ |
18 | | -# return ApigeeApiDeveloperApps() |
19 | | -# |
20 | | -# @pytest.fixture() |
21 | | -# def product(self): |
22 | | -# """ |
23 | | -# Import the test utils module to be able to: |
24 | | -# - Create apigee test product |
25 | | -# - Update custom scopes |
26 | | -# - Update environments |
27 | | -# - Update product paths |
28 | | -# - Update custom attributes |
29 | | -# - Update proxies to the product |
30 | | -# - Update custom ratelimits |
31 | | -# """ |
32 | | -# return ApigeeApiProducts() |
33 | | -# |
34 | | -# @pytest.fixture() |
35 | | -# async def test_app_and_product(self, app, product): |
36 | | -# """Create a test app and product which can be modified in the test""" |
37 | | -# await product.create_new_product() |
38 | | -# |
39 | | -# await app.create_new_app() |
40 | | -# |
41 | | -# await product.update_scopes([ |
42 | | -# "urn:nhsd:apim:app:level3:splunk-monitor", |
43 | | -# "urn:nhsd:apim:user-nhs-id:aal3:splunk-monitor" |
44 | | -# ]) |
45 | | -# await app.add_api_product([product.name]) |
46 | | -# |
47 | | -# yield product, app |
48 | | -# |
49 | | -# await app.destroy_app() |
50 | | -# await product.destroy_product() |
51 | | -# |
52 | | -# @pytest.fixture() |
53 | | -# async def get_token(self, test_app_and_product): |
54 | | -# """Call identity server to get an access token""" |
55 | | -# test_product, test_app = test_app_and_product |
56 | | -# oauth = OauthHelper( |
57 | | -# client_id=test_app.client_id, |
58 | | -# client_secret=test_app.client_secret, |
59 | | -# redirect_uri=test_app.callback_url |
60 | | -# ) |
61 | | -# token_resp = await oauth.get_token_response(grant_type="authorization_code") |
62 | | -# assert token_resp["status_code"] == 200 |
63 | | -# return token_resp['body'] |
64 | | -# |
65 | | -# def test_user_restricted(self, get_token): |
66 | | -# """ |
67 | | -# In here you can add tests which call your proxy |
68 | | -# You can use the 'get_token' fixture to call the proxy with a access token |
69 | | -# """ |
70 | | -# pass |
| 1 | +from os import getenv |
| 2 | + |
| 3 | +import pytest |
| 4 | +import requests |
| 5 | + |
| 6 | + |
| 7 | +@pytest.mark.e2e |
| 8 | +@pytest.mark.smoketest |
| 9 | +@pytest.mark.asyncio |
| 10 | +async def test_wait_for_ping(nhsd_apim_proxy_url): |
| 11 | + """ |
| 12 | + test for _ping, this waits until the correct SOURCE_COMMIT_ID ( from env var ) |
| 13 | + is available |
| 14 | + """ |
| 15 | + retries = 0 |
| 16 | + resp = requests.get(f"{nhsd_apim_proxy_url}/_ping") |
| 17 | + deployed_commit_id = resp.json().get("commitId") |
| 18 | + |
| 19 | + while (deployed_commit_id != getenv('SOURCE_COMMIT_ID') |
| 20 | + and retries <= 30 |
| 21 | + and resp.status_code == 200): |
| 22 | + resp = requests.get(f"{nhsd_apim_proxy_url}/_ping") |
| 23 | + deployed_commit_id = resp.json().get("commitId") |
| 24 | + retries += 1 |
| 25 | + |
| 26 | + if resp.status_code != 200: |
| 27 | + pytest.fail(f"Status code {resp.status_code}, expecting 200") |
| 28 | + elif retries >= 30: |
| 29 | + pytest.fail("Timeout Error - max retries") |
| 30 | + |
| 31 | + assert deployed_commit_id == getenv('SOURCE_COMMIT_ID') |
0 commit comments