|
4 | 4 | from aiohttp.test_utils import TestClient |
5 | 5 | from aioresponses import aioresponses as AioResponsesMock |
6 | 6 | from common_library.users_enums import UserRole |
| 7 | +from models_library.api_schemas_catalog.service_access_rights import ( |
| 8 | + ServiceAccessRightsGet, |
| 9 | +) |
7 | 10 | from pytest_simcore.helpers.webserver_login import UserInfoDict |
8 | 11 | from servicelib.aiohttp import status |
9 | 12 | from simcore_service_webserver.catalog.catalog_service import ( |
| 13 | + get_service_access_rights, |
10 | 14 | get_services_for_user_in_product, |
11 | 15 | is_catalog_service_responsive, |
12 | 16 | ) |
|
16 | 20 | "user_role", |
17 | 21 | [UserRole.USER], |
18 | 22 | ) |
| 23 | +@pytest.mark.parametrize( |
| 24 | + "backend_status_code", [status.HTTP_200_OK, status.HTTP_500_INTERNAL_SERVER_ERROR] |
| 25 | +) |
19 | 26 | async def test_server_responsive( |
20 | | - client: TestClient, logged_user: UserInfoDict, aioresponses_mocker: AioResponsesMock |
| 27 | + client: TestClient, |
| 28 | + logged_user: UserInfoDict, |
| 29 | + aioresponses_mocker: AioResponsesMock, |
| 30 | + backend_status_code: int, |
21 | 31 | ): |
22 | | - aioresponses_mocker.get( |
23 | | - "http://catalog:8000", |
24 | | - status=status.HTTP_200_OK, |
25 | | - ) |
| 32 | + aioresponses_mocker.get("http://catalog:8000", status=backend_status_code) |
26 | 33 |
|
27 | 34 | assert client.app |
28 | | - assert await is_catalog_service_responsive(app=client.app) == True |
| 35 | + is_responsive = await is_catalog_service_responsive(app=client.app) |
| 36 | + if backend_status_code == status.HTTP_200_OK: |
| 37 | + assert is_responsive == True |
| 38 | + else: |
| 39 | + assert is_responsive == False |
29 | 40 |
|
30 | 41 |
|
31 | 42 | @pytest.mark.parametrize( |
@@ -53,3 +64,37 @@ async def test_get_services_for_user_in_product( |
53 | 64 | product_name="osparc", |
54 | 65 | only_key_versions=False, |
55 | 66 | ) |
| 67 | + |
| 68 | + |
| 69 | +@pytest.mark.parametrize( |
| 70 | + "user_role", |
| 71 | + [UserRole.USER], |
| 72 | +) |
| 73 | +async def test_get_service_access_rights( |
| 74 | + client: TestClient, |
| 75 | + logged_user: UserInfoDict, |
| 76 | + aioresponses_mocker: AioResponsesMock, |
| 77 | +): |
| 78 | + url_pattern = re.compile(r"http://catalog:8000/.*") |
| 79 | + example = ServiceAccessRightsGet( |
| 80 | + service_key="simcore/services/comp/itis/sleeper", |
| 81 | + service_version="2.1.4", |
| 82 | + gids_with_access_rights={ |
| 83 | + 1: {"execute_access": True}, |
| 84 | + 5: {"execute_access": True}, |
| 85 | + }, |
| 86 | + ) |
| 87 | + aioresponses_mocker.get( |
| 88 | + url_pattern, |
| 89 | + status=status.HTTP_200_OK, |
| 90 | + payload=example.model_dump(), |
| 91 | + ) |
| 92 | + assert client.app |
| 93 | + access_rights = await get_service_access_rights( |
| 94 | + app=client.app, |
| 95 | + user_id=logged_user["id"], |
| 96 | + service_key="simcore/services/comp/itis/sleeper", |
| 97 | + service_version="2.1.4", |
| 98 | + product_name="osparc", |
| 99 | + ) |
| 100 | + assert isinstance(access_rights, ServiceAccessRightsGet) |
0 commit comments