From 0665836986aa8e4df47b7f8488ba865e05ada12c Mon Sep 17 00:00:00 2001 From: sujanadh Date: Mon, 2 Oct 2023 13:19:30 +0545 Subject: [PATCH] feat: Implement comprehensive test coverage for the Create Project API and related functions --- src/backend/tests/conftest.py | 25 +++- src/backend/tests/test_projects_routes.py | 169 ++++++++++++++++++++++ 2 files changed, 193 insertions(+), 1 deletion(-) create mode 100644 src/backend/tests/test_projects_routes.py diff --git a/src/backend/tests/conftest.py b/src/backend/tests/conftest.py index 65a8c7a97a..7d1c9c5b57 100644 --- a/src/backend/tests/conftest.py +++ b/src/backend/tests/conftest.py @@ -17,19 +17,29 @@ # import pytest +from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy_utils import create_database, database_exists +from typing import Generator, Any from app.config import settings from app.db.database import Base, get_db -from app.main import api +from app.main import api, get_application +from app.db.db_models import DbOrganisation, DbProject, DbUser +from sqlalchemy.sql import text engine = create_engine(settings.FMTM_DB_URL) TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base.metadata.create_all(bind=engine) +@pytest.fixture(autouse=True) +def app() -> Generator[FastAPI, Any, None]: + """ + Create a fresh database on each test case. + """ + yield get_application() @pytest.fixture(scope="session") def db_engine(): @@ -56,6 +66,19 @@ def db(db_engine): db.rollback() connection.close() +@pytest.fixture(scope="function") +def get_ids(db): + user_id_query = text(f"SELECT id FROM {DbUser.__table__.name} LIMIT 1") + organization_id_query = text(f"SELECT id FROM {DbOrganisation.__table__.name} LIMIT 1") + project_id_query = text(f"SELECT id FROM {DbProject.__table__.name} LIMIT 1") + + user_id = db.execute(user_id_query).scalar() + organization_id = db.execute(organization_id_query).scalar() + project_id = db.execute(project_id_query).scalar() + + return {"user_id": user_id, "organization_id": organization_id, "project_id": project_id} + + @pytest.fixture(scope="function") def client(db): diff --git a/src/backend/tests/test_projects_routes.py b/src/backend/tests/test_projects_routes.py new file mode 100644 index 0000000000..761c00f08b --- /dev/null +++ b/src/backend/tests/test_projects_routes.py @@ -0,0 +1,169 @@ +from app.projects.project_schemas import BETAProjectUpload +from fastapi import FastAPI +from geoalchemy2.elements import WKBElement +from app.users.user_schemas import User +from app.projects.project_schemas import ProjectInfo, ODKCentral +import pytest +import os +from dotenv import load_dotenv +from app.db import db_models +from app.central.central_crud import create_odk_project +from unittest.mock import Mock +from app.projects import project_crud, project_schemas +from unittest.mock import patch +from shapely import Polygon +import json +import uuid +from app.tasks import tasks_crud + +load_dotenv() + +odk_central_url = os.getenv("ODK_CENTRAL_URL") +odk_central_user = os.getenv("ODK_CENTRAL_USER") +odk_central_password = os.getenv("ODK_CENTRAL_PASSWD") + +app=FastAPI() + +class MockSession: + def add(self, item): + pass + + def commit(self): + pass + + def refresh(self, item): + pass + +def test_create_project(client, db): + project_data = { + "author": { + "username": "test_user", + "id": 1 + }, + "project_info": { + "name": "test project", + "short_description": "test", + "description": "test" + }, + "xform_title": "buildings", + "odk_central": { + "odk_central_url": odk_central_url, + "odk_central_user": odk_central_user, + "odk_central_password": odk_central_password, + }, + "hashtags": [ + "hot-fmtm" + ], + "organisation_id": 1 + } + + response = client.post("/projects/create_project", json=project_data) + + assert response.status_code == 200 + + response_data = response.json() + assert "id" in response_data + + + +def test_create_odk_project(): + mock_project = Mock() + mock_project.createProject.return_value = {"status": "success"} + + with patch('app.central.central_crud.get_odk_project', return_value=mock_project): + result = create_odk_project("Test Project") + + assert result == {"status": "success"} + mock_project.createProject.assert_called_once_with("Test Project") + + +def test_convert_to_app_project(): + polygon = Polygon([ + (85.924707758, 26.727727503), + (85.922703741, 26.732440043), + (85.928284549, 26.735158727), + (85.930643709, 26.734365785), + (85.932368686, 26.732372075), + (85.924707758, 26.727727503) + ]) + +# Get the WKB representation of the Polygon + wkb_outline = WKBElement(polygon.wkb, srid=4326) + mock_db_project = db_models.DbProject( + id=1, + outline=wkb_outline, + ) + + result = project_crud.convert_to_app_project(mock_db_project) + + assert result is not None + assert isinstance(result, db_models.DbProject) + + assert result.outline_geojson is not None + + assert result.project_tasks is not None + assert isinstance(result.project_tasks, list) + +def test_create_project_with_project_info(db): + project_metadata = BETAProjectUpload( + author=User(username='test_user', id=1), + project_info=ProjectInfo( + name='test project', + short_description='test', + description='test', + ), + xform_title='buildings', + odk_central=ODKCentral( + odk_central_url= odk_central_url, + odk_central_user= odk_central_user, + odk_central_password= odk_central_password, + ), + hashtags=['hot-fmtm'], + organisation_id=1, + ) + try: + result = project_crud.create_project_with_project_info(db, project_metadata, project_id=123) + assert result is not None + except Exception as e: + pytest.fail(f"Test failed with exception: {str(e)}") + +def test_generate_app_user(db, get_ids): + custom_form = "/opt/app/test_data/buildings.xls" + with open(custom_form, "rb") as file: + contents = file.read() + data_extracts = "/opt/app/test_data/building_foot_jnk.geojson" + with open(data_extracts, "rb") as file: + extract_contents = file.read() + json.loads(extract_contents) + odk_credentials = { + "odk_central_url": odk_central_url, + "odk_central_user": odk_central_user, + "odk_central_password": odk_central_password, + } + odk_credentials = project_schemas.ODKCentral(**odk_credentials) + project_id = get_ids["project_id"] + test_data = { + "db": db, + "project_id": project_id, + "extract_polygon": True, + "upload": contents, + "extracts_contents": extract_contents, + "category": "buildings", + "form_type": "example_form_type", + "background_task_id": uuid.uuid4(), + } + result1 = project_crud.upload_custom_data_extracts(db, project_id, extract_contents) + assert result1 == True + + result2 = tasks_crud.get_task_lists(db, project_id) + assert isinstance(result2, list) + + for task in result2: + result3 = project_crud.generate_task_files(db, project_id, task, custom_form, 'xls', odk_credentials) + assert result3 == True + + result = project_crud.generate_appuser_files(**test_data) + assert result is None + +if __name__ == '__main__': + pytest.main() \ No newline at end of file