[TEST] for Heatmaps & Customizations 🧪

app/api/.devcontainer.json

@@ -8,6 +8,13 @@
     "service": "api",
     "workspaceFolder": "/app",
     "shutdownAction": "none",
+    "mounts": [
+        "source=${localWorkspaceFolder}/../..,target=/goat,type=bind,consistency=cached"
+    ],
+    "remoteEnv": {
+        "GIT_WORK_TREE": "/goat",
+        "GIT_DIR": "/goat/.git"
+    },
     "customizations": {
         "vscode": {
             "extensions": [
@@ -16,7 +23,9 @@
                 "ms-vscode.cpptools",
                 "twxs.cmake",
                 "ms-vscode.cmake-tools",
-                "njpwerner.autodocstring"
+                "njpwerner.autodocstring",
+                "donjayamanne.githistory",
+                "ms-python.black-formatter"
             ],
             "settings": {
                 "python.defaultInterpreterPath": "/usr/local/bin/python",

app/api/src/endpoints/deps.py

@@ -55,7 +55,7 @@ def get_current_active_superuser(
     current_user: models.User = Depends(get_current_user),
 ) -> models.User:
     if not crud.user.is_superuser(current_user):
-        raise HTTPException(status_code=400, detail="The user doesn't have enough privileges")
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="The user doesn't have enough privileges")
     return current_user
 
 

app/api/src/endpoints/v1/customizations.py

@@ -82,21 +82,17 @@ async def delete_user_settings(
     return update_settings
 
 
-@router.get("{user_id}/{study_area_id}", response_class=JSONResponse)
+@router.get("/{user_id}/{study_area_id}", response_class=JSONResponse)
 async def get_user_settings(
     *,
     db: AsyncSession = Depends(deps.get_db),
     study_area_id: int = None,
     user_id: int = None,
-    current_user: models.User = Depends(deps.get_current_active_user),
+    current_user: models.User = Depends(deps.get_current_active_superuser),
 ) -> Any:
     """
     Get customization settings for user.
     """
-    is_superuser = crud.user.is_superuser(user=current_user)
-
-    if user_id != current_user.id and not is_superuser:
-        raise HTTPException(status_code=400, detail="The user cannot get another user's settings")
     customizations = await crud.customization.get_multi(db)
     settings = {}
     for customization in customizations:
@@ -148,6 +144,7 @@ async def update_user_settings(
                 setting=user_customizations[user_customization_key],
                 user_id=user_id,
                 customization_id=customization.id,
+                study_area_id=study_area_id,
             )
             if user_customization is not None and len(user_customization) > 0:
                 del user_customization_in.id
@@ -182,7 +179,7 @@ async def delete_user_setting(
     customization = await CRUDBase(models.Customization).get_by_key(
         db, key="type", value=customization
     )
-    if customization is None:
+    if not customization:
         raise HTTPException(status_code=400, detail="Customization not found")
     else:
         customization = customization[0]

app/api/src/tests/api/api_v1/test_customizations.py

@@ -15,7 +15,7 @@
 
 
 # get user customization
-async def test_get_customizations(
+async def test_get_customizations_me(
     client: AsyncClient, superuser_token_headers: Dict[str, str]
 ) -> None:
     r = await client.get(
@@ -84,6 +84,83 @@ async def test_superuser_get_normal_user_setting(
     assert isinstance(all_customizations["poi_groups"], list)
 
 
-# TODO: test_superuser_update_normal_user_setting
+
+async def test_superuser_update_normal_user_setting(
+    client: AsyncClient, superuser_token_headers: Dict[str, str], db: AsyncSession
+) -> None:
+    # Create a random user
+    user = await create_random_user(db=db)
+
+    # Create settings for the user
+    await dynamic_customization.build_main_setting_json(db=db, current_user=user)
+
+    # Get the main settings and modify it
+    user_settings = request_examples["create"]
+    user_settings["map"]["zoom"] = 10
+
+    r = await client.post(
+        f"{settings.API_V1_STR}/customizations/{user.id}/{user.active_study_area_id}",
+        headers=superuser_token_headers,
+        json=user_settings,
+    )
+
+    assert 200 <= r.status_code < 300
+
+    # Check that the settings have been updated
+
+    r = await client.get(
+        f"{settings.API_V1_STR}/customizations/{user.id}/{user.active_study_area_id}",
+        headers=superuser_token_headers,
+    )
+    assert 200 <= r.status_code < 300
+    updated_settings = r.json()
+    assert updated_settings["map"]["zoom"] == 10
+
+
+
 # TODO: test_superuser_delete_normal_user_setting
-# TODO: test_normal_user_get_normal_user_setting
+
+async def test_superuser_delete_normal_user_setting(
+    client: AsyncClient, superuser_token_headers: Dict[str, str], db: AsyncSession
+) -> None:
+    # Create a random user
+    user = await create_random_user(db=db)
+
+    # Create settings for the user
+    await dynamic_customization.build_main_setting_json(db=db, current_user=user)
+
+    # Create poi settings
+    obj_dict = jsonable_encoder(request_examples["user_customization_insert"]["poi"]["value"])
+
+    await dynamic_customization.insert_opportunity_setting(
+        db=db,
+        current_user=user,
+        insert_settings=obj_dict
+    )
+
+    # Delete the settings
+    r = await client.delete(
+        f"{settings.API_V1_STR}/customizations/{user.id}/{user.active_study_area_id}/poi",
+        headers=superuser_token_headers,
+    )
+    assert 200 <= r.status_code < 300
+
+
+
+async def test_normal_user_get_normal_user_setting(
+    client: AsyncClient, normaluser_token_headers: Dict[str, str], db: AsyncSession
+) -> None:
+    # Create a random user
+    user = await create_random_user(db=db)
+
+    # Create settings for the user
+    await dynamic_customization.build_main_setting_json(db=db, current_user=user)
+
+
+    # Get the settings of other user
+    r = await client.get(
+        f"{settings.API_V1_STR}/customizations/{user.id}/{user.active_study_area_id}",
+        headers=normaluser_token_headers,
+    )
+    response = r.json()
+    assert r.status_code == 403

app/api/src/tests/api/api_v1/test_heatmap.py

@@ -0,0 +1,126 @@
+import asyncio
+import pyproj
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+from src import crud
+from src.crud.crud_scenario import scenario as crud_scenario
+from src.core.config import settings
+from src.db import models
+from src.schemas.heatmap import request_examples
+
+from src.tests.utils.utils import random_lower_string
+
+pytestmark = pytest.mark.asyncio
+
+async def heatmap_set_request(
+    client: AsyncClient, superuser_token_headers: dict[str, str], data: dict
+) -> None:
+    r = await client.post(
+        f"{settings.API_V1_STR}/indicators/heatmap",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    return r
+
+async def heatmap_get_results(
+    client: AsyncClient, superuser_token_headers: dict[str, str], task_id: str
+) -> None:
+    r = await client.get(
+        f"{settings.API_V1_STR}/indicators/result/{task_id}",
+        headers=superuser_token_headers,
+        params={"return_type": "geojson"},
+    )
+    return r
+
+async def heatmap_test_base(
+    client: AsyncClient, superuser_token_headers: dict[str, str], data: dict
+) -> None:
+    task_request = await heatmap_set_request(client, superuser_token_headers, data)
+    assert task_request.status_code == 200
+    task_id = task_request.json()["task_id"]
+
+    for i in range(10):
+        task_results = await heatmap_get_results(client, superuser_token_headers, task_id)
+        assert task_results.status_code >= 200 and task_results.status_code < 300
+        if task_results.status_code == 200:
+            break
+        else:
+            await asyncio.sleep(1)
+
+    assert task_results.status_code == 200
+
+
+
+async def test_connectivity_heatmap_6_walking(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["connectivity_heatmap_6_walking"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+
+async def test_modified_gaussian_hexagon_10(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["modified_gaussian_hexagon_10"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_connectivity_heatmap_6_transit(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["connectivity_heatmap_6_transit"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_modified_gaussian_hexagon_9(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["modified_gaussian_hexagon_9"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_modified_gaussian_hexagon_6(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["modified_gaussian_hexagon_6"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_combined_modified_gaussian_hexagon_6(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["combined_modified_gaussian_hexagon_6"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_closest_average_hexagon_10(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["closest_average_hexagon_10"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_closest_average_hexagon_9(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["closest_average_hexagon_9"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_closest_average_hexagon_6(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["closest_average_hexagon_6"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_connectivity_heatmap_10(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["connectivity_heatmap_10"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_aggregated_data_heatmap_10(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["aggregated_data_heatmap_10"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)
+
+async def test_modified_gaussian_population_6(
+    client: AsyncClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = request_examples["modified_gaussian_population_6"]["value"]
+    await heatmap_test_base(client, superuser_token_headers, data)

app/api/src/tests/conftest.py

@@ -3,10 +3,12 @@
 
 import pytest_asyncio
 from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.db.session import async_session
 from src.main import app
 from src.tests.utils.utils import get_superuser_token_headers
+from src.tests.utils.user import get_user_token_headers
 
 
 @pytest_asyncio.fixture(scope="session")
@@ -29,3 +31,7 @@ async def client() -> Generator:
 @pytest_asyncio.fixture(scope="module")
 async def superuser_token_headers(client: AsyncClient) -> Dict[str, str]:
     return await get_superuser_token_headers(client)
+
+@pytest_asyncio.fixture(scope="module")
+async def normaluser_token_headers(client: AsyncClient, db: AsyncSession) -> Dict[str, str]:
+    return await get_user_token_headers(client=client, db=db)