diff --git a/tests/test_campaigns.py b/tests/test_campaigns.py index 83ff5dd6..5fbd0436 100755 --- a/tests/test_campaigns.py +++ b/tests/test_campaigns.py @@ -41,8 +41,9 @@ # FIXME: For now these tests assume you have a local postgres installed. One has the TM # database, the other for tm_admin. -user = UsersDB('localhost/testdata') -project = ProjectsDB('localhost/testdata') +dbname = os.getenv("TMDB", default="localhost/testdata") +user = UsersDB(dbname) +project = ProjectsDB(dbname) def get_campaign(): """Gets the specified campaign""" diff --git a/tests/test_messages.py b/tests/test_messages.py index 06de6b16..d4f33437 100755 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -41,8 +41,10 @@ # FIXME: For now these tests assume you have a local postgres installed. One has the TM # database, the other for tm_admin. -user = UsersDB('localhost/testdata') -project = ProjectsDB('localhost/testdata') +dbname = os.getenv("TMDB", default="localhost/testdata") +user = UsersDB(dbname) +project = ProjectsDB(dbname) +task = TasksDB(dbname) def send_welcome_message(): """Sends welcome message to new user at Sign up""" diff --git a/tests/test_notifications.py b/tests/test_notifications.py index a88663c7..c8586d2b 100755 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -41,8 +41,10 @@ # FIXME: For now these tests assume you have a local postgres installed. One has the TM # database, the other for tm_admin. -user = UsersDB('localhost/testdata') -project = ProjectsDB('localhost/testdata') +dbname = os.getenv("TMDB", default="localhost/testdata") +user = UsersDB(dbname) +project = ProjectsDB(dbname) +task = TasksDB(dbname) def update(): # user_id: int): diff --git a/tests/test_orgs.py b/tests/test_orgs.py index 101eb255..67b12248 100755 --- a/tests/test_orgs.py +++ b/tests/test_orgs.py @@ -39,7 +39,8 @@ # FIXME: For now these tests assume you have a local postgres installed. One has the TM # database, the other for tm_admin. -organization = OrganizationsDB('localhost/tm_admin') +dbname = os.getenv("TMDB", default="localhost/testdata") +organization = OrganizationsDB(dbname) def get_organisation_by_id(): log.debug(f"--- get_organisation_by_id() ---") @@ -130,6 +131,28 @@ def is_user_an_org_manager(): # def get_organisation_by_id_as_dto(): # def get_organisation_by_slug_as_dto(): +# FMTM API tests +def get_organisations(): + log.debug(f"--- get_organisations() unimplemented!") + +def generate_slug(): + # text: str) -> str: + log.debug(f"--- generate_slug() unimplemented!") + +def get_organisation_by_name(): + # db: Session, name: str): + log.debug(f"--- get_organisation_by_name() unimplemented!") + +def upload_image(): + # db: Session, file: UploadFile(None)): + log.debug(f"--- upload_image() unimplemented!") + +def create_organization(): + log.debug(f"--- create_organization() unimplemented!") + # get_organisation_by_id(db: Session, id: int): + +def update_organization_info(): + log.debug(f"--- update_organization_info() unimplemented!") if __name__ == "__main__": parser = argparse.ArgumentParser() @@ -168,3 +191,11 @@ def is_user_an_org_manager(): # get_organisation_by_slug_as_dto() # get_organisation_dto() # get_organisations_as_dto( + + # FMTM Aget_organisations(); + generate_slug() + get_organisation_by_name() + upload_image() + create_organization() + update_organization_info() + diff --git a/tests/test_projects.py b/tests/test_projects.py index cd55e9f4..e7b89006 100755 --- a/tests/test_projects.py +++ b/tests/test_projects.py @@ -44,7 +44,7 @@ # FIXME: For now these tests assume you have a local postgres installed. One has the TM # database, the other for tm_admin. -dbname = 'localhost/testdata' +dbname = os.getenv("TMDB", default="localhost/testdata") user = UsersDB(dbname) project = ProjectsDB(dbname) task = TasksDB(dbname) @@ -174,7 +174,7 @@ def evaluate_mapping_permission(): perm = Permissions.ANY userrole = user.getColumn(uid, 'role') team = user.getColumn(uid, 'team_members') - mapperms = projects.getColumn(pid, 'mapping_permission') + mapperms = project.getColumn(pid, 'mapping_permission') #result = team.getByWhere(f" id={uid}") #print(result) @@ -242,6 +242,148 @@ def send_email_on_project_progress(): # we use the UsersTable() to represent the table schema # def get_project_dto_for_mapper(): +# FMTM API tests +def get_projects(): + log.debug(f"--- FMTM get_projects() ---") + result = user.getAll() + assert len(result) > 0 + +def get_project(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_project() ---") + id = 150 + result = project.getByWhere(f" id='{id}'") + assert len(result) > 0 + +def get_project_by_id(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_project_by_id() unimplemented!") + id = 150 + result = project.getByWhere(f" id='{id}'") + +def get_project_geometry(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_project_geometry() ---") + pid = 150 + result = project.getColumn(pid, 'geometry') + # FIXME: this should test the geometry to make + # sure it's valid + assert len(result) > 0 + +def get_task_geometry(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_task_geometry() ---") + tid = 150 + result = task.getColumn(tid, 'geometry') + # FIXME: this should test the geometry to make + # sure it's valid + assert len(result) > 0 + +def get_project_summaries(): + log.debug(f"--- FMTM get_project_summaries() unimplemented!") + +def get_project_by_id_w_all_tasks(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_project_by_id_w_all_tasks() unimplemented!") + +def get_project_info_by_id(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_project_info_by_id() unimplemented!") + +def delete_project_by_id(): + # db: Session, project_id: int): + log.debug(f"--- FMTM delete_project_by_id() unimplemented!") + +def partial_update_project_info(): + log.debug(f"--- FMTM partial_update_project_info() unimplemented!") + +def update_project_info(): + log.debug(f"--- FMTM update_project_info() unimplemented!") + +def create_project_with_project_info(): + log.debug(f"--- FMTM create_project_with_project_info() unimplemented!") + +def upload_xlsform(): + log.debug(f"--- FMTM upload_xlsform() unimplemented!") + +def update_multi_polygon_project_boundary(): + log.debug(f"--- FMTM update_multi_polygon_project_boundary() unimplemented!") + +def preview_tasks(): + # boundary: str, dimension: int): + log.debug(f"--- FMTM preview_tasks() unimplemented!") + +def update_project_boundary(): + log.debug(f"--- FMTM update_project_boundary() unimplemented!") + +def read_xlsforms(): + log.debug(f"--- FMTM read_xlsforms() unimplemented!") + +def get_odk_id_for_project(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_odk_id_for_project() unimplemented!") + +def upload_custom_data_extracts(): + log.debug(f"--- FMTM upload_custom_data_extracts() unimplemented!") + +def get_project_features_geojson(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_project_features_geojson() unimplemented!") + +def get_project_features(): + # db: Session, project_id: int, task_id: int = None): + log.debug(f"--- FMTM get_project_features() unimplemented!") + +def add_features_into_database(): + log.debug(f"--- FMTM add_features_into_database() unimplemented!") + +def update_project_form(): + log.debug(f"--- FMTM update_project_form() unimplemented!") + +def update_odk_credentials(): + log.debug(f"--- FMTM update_odk_credentials() unimplemented!") + +def get_extracted_data_from_db(): + # db: Session, project_id: int, outfile: str): + log.debug(f"--- FMTM get_extracted_data_from_db() unimplemented!") + +def get_project_tiles(): + log.debug(f"--- FMTM get_project_tiles() unimplemented!") + +def get_mbtiles_list(): + # db: Session, project_id: int): + log.debug(f"--- FMTM get_mbtiles_list() unimplemented!") + +def update_project_location_info(): + log.debug(f"--- FMTM update_project_location_info() unimplemented!") + +# def get_extract_completion_count(): +# def get_osm_extracts(boundary: str): +# def get_shape_from_json_str(feature: str, error_detail: str): +# def get_dbqrcode_from_file(zip, qr_filename: str, error_detail: str): +# def get_outline_from_geojson_file_in_zip( +# def get_json_from_zip(zip, filename: str, error_detail: str): +# def create_task_grid(db: Session, project_id: int, delta: int): +# def generate_task_files_wrapper(project_id, task, xlsform, form_type, odk_credentials): +# def generate_appuser_files(): +# def create_qrcode(): +# def insert_background_task_into_database(): +# def update_background_task_status_in_database(): +# def get_background_task_status(task_id: uuid.UUID, db: Session): +# def convert_geojson_to_osm(geojson_file: str): +# def split_into_tasks(): +# def split_polygon_into_tasks(): +# def update_project_with_zip(): +# def generate_task_files(): + +# def convert_to_app_project(db_project: db_models.DbProject): +# def convert_to_app_project_info(db_project_info: db_models.DbProjectInfo): +# def convert_to_app_projects(db_projects: List[db_models.DbProject]): +# def convert_to_project_summary(db_project: db_models.DbProject): +# def convert_to_project_summaries(db_projects: List[db_models.DbProject]): +# def convert_to_project_feature(db_project_feature: db_models.DbFeatures): +# def convert_to_project_features(db_project_features: List[db_models.DbFeatures]): + if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("-v", "--verbose", nargs="?", const="0", help="verbose output") @@ -292,3 +434,33 @@ def send_email_on_project_progress(): send_email_on_project_progress() # is_user_in_the_allowed_list() # get_project_dto_for_mapper() + + # FMTM API tests + get_projects() + get_project_summaries() + get_project_by_id_w_all_tasks() + get_project() + get_project_by_id() + get_project_info_by_id() + delete_project_by_id() + partial_update_project_info() + update_project_info() + create_project_with_project_info() + upload_xlsform() + update_multi_polygon_project_boundary() + preview_tasks() + update_project_boundary() + read_xlsforms() + get_odk_id_for_project() + upload_custom_data_extracts() + get_project_geometry() + get_task_geometry() + get_project_features_geojson() + get_project_features() + add_features_into_database() + update_project_form() + update_odk_credentials() + get_extracted_data_from_db() + get_project_tiles() + get_mbtiles_list() + update_project_location_info() diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 696f2c9a..521c158f 100755 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -38,6 +38,11 @@ import tm_admin as tma rootdir = tma.__path__[0] +dbname = os.getenv("TMDB", default="localhost/testdata") +user = UsersDB(dbname) +project = ProjectsDB(dbname) +task = TasksDB(dbname) + # FIXME: For now these tests assume you have a local postgres installed. One has the TM # database, the other for tm_admin. @@ -109,9 +114,58 @@ def extend_task_lock_time(): log.debug(f"--- extend_task_lock_time() unimplemented!") #extend_dto: ExtendLockTimeDTO): +# FMTM API tests +def get_task_count_in_project(): + # db: Session, project_id: int): + log.debug(f"--- get_task_count_in_project() unimplemented!") + +def get_task_lists(): + # db: Session, project_id: int): + log.debug(f"--- get_task_lists() unimplemented!") + +def get_tasks(): + # + log.debug(f"--- get_tasks() unimplemented!") + +def get_task(): + # db: Session, task_id: int, db_obj: bool = False): + log.debug(f"--- get_task() unimplemented!") + +def update_task_status(): + # db: Session, user_id: int, task_id: int, new_status: TaskStatus): + log.debug(f"--- update_task_status() unimplemented!") + +def update_qrcode(): + log.debug(f"--- update_qrcode() unimplemented!") + +def create_task_history_for_status_change(): + log.debug(f"--- create_task_history_for_status_change() unimplemented!") -user = UsersDB('localhost/testdata') -project = ProjectsDB('localhost/testdata') +def convert_to_app_history(): + # db_histories: List[db_models.DbTaskHistory]): + log.debug(f"--- convert_to_app_history() unimplemented!") + +def convert_to_app_task(): + # db_task: db_models.DbTask): + log.debug(f"--- convert_to_app_task() unimplemented!") + +def convert_to_app_tasks(): + # db_tasks: List[db_models.DbTask]): + log.debug(f"--- convert_to_app_tasks() unimplemented!") + +def get_qr_codes_for_task(): + log.debug(f"--- get_qr_codes_for_task() unimplemented!") + +def get_task_by_id(): + # db: Session, task_id: int): + log.debug(f"--- get_task_by_id() unimplemented!") + +def update_task_files(): + log.debug(f"--- update_task_files() unimplemented!") + +def edit_task_boundary(): + # db: Session, task_id: int, boundary: str): + log.debug(f"--- edit_task_boundary() unimplemented!") if __name__ == "__main__": parser = argparse.ArgumentParser() @@ -146,3 +200,18 @@ def extend_task_lock_time(): reset_all_badimagery() lock_time_can_be_extended() extend_task_lock_time() + + # FMTM API tests + + get_tasks() + get_task() + update_task_status() + update_qrcode() + create_task_history_for_status_change() + convert_to_app_history() + convert_to_app_task() + convert_to_app_tasks() + get_qr_codes_for_task() + get_task_by_id() + update_task_files() + edit_task_boundary() diff --git a/tests/test_teams.py b/tests/test_teams.py index 88e8077c..f5d2d668 100755 --- a/tests/test_teams.py +++ b/tests/test_teams.py @@ -40,7 +40,8 @@ # FIXME: For now these tests assume you have a local postgres installed. One has the TM # database, the other for tm_admin. -team = TeamsDB('localhost/tm_admin') +dbname = os.getenv("TMDB", default="localhost/testdata") +teams = TeamsDB(dbname) def get_team_by_id(): log.debug(f"get_team_by_id() unimplemented!") diff --git a/tests/test_users.py b/tests/test_users.py index a698f1b8..ca40d324 100755 --- a/tests/test_users.py +++ b/tests/test_users.py @@ -44,8 +44,10 @@ #UPDATE users SET favorite_projects = ARRAY[1,2,16,5]; -user = UsersDB('localhost/testdata') -project = ProjectsDB('localhost/testdata') +tmdb = os.getenv("TMDB", default="localhost/testdata") + +user = UsersDB(tmdb) +project = ProjectsDB(tmdb) # def get_all_users(query: UserSearchQuery): def test_all(): @@ -73,7 +75,7 @@ def test_role(): log.debug("--- test_role() ---") id = 4606673 role = Userrole(Userrole.ADMIN) - result user.updateColumn(id, {'role': role.name}) + result = user.updateColumn(id, {'role': role.name}) assert result def get_mapping_level(): @@ -312,7 +314,40 @@ def get_recommended_projects(): # we use the UsersTable() to represent the table schema # def get_user_dto_by_username(): # def get_user_dto_by_id(user: int, request_user: int): + +# test FMTM API +def get_users(): + # db: Session, skip: int = 0, limit: int = 100): + log.debug(f"--- get_users() unimplemented!") + +def get_user(): + # db: Session, user_id: int, db_obj: bool = False): + log.debug(f"--- get_user() unimplemented!") + +def get_user_by_username(): + # db: Session, username: str): + log.debug(f"--- get_user_by_username() unimplemented!") + +def convert_to_app_user(): + # db_user: db_models.DbUser): + log.debug(f"--- convert_to_app_user() unimplemented!") + +def convert_to_app_users(): + # db_users: List[db_models.DbUser]): + log.debug(f"--- convert_to_app_users() unimplemented!") + +def get_user_role_by_user_id(): + # db: Session, user_id: int): + log.debug(f"--- get_user_role_by_user_id() unimplemented!") +def create_user_roles(): + # user_role: user_schemas.UserRoles, db: Session): + log.debug(f"--- create_user_roles() unimplemented!") + +def get_user_by_id(): + # db: Session, user_id: int): + log.debug(f"--- get_user_by_id() unimplemented!") + if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("-v", "--verbose", nargs="?", const="0", help="verbose output") @@ -331,8 +366,7 @@ def get_recommended_projects(): stream=sys.stdout, ) - # user = UsersDB(args.uri) - + # Test TM API # get_user_dto_by_username() # get_user_dto_by_id() test_by_id() @@ -370,5 +404,13 @@ def get_recommended_projects(): # is_user_the_project_author() # get_osm_details_for_user() # Not part of this API # notify_level_upgrade() # Not part of this API - test_all() + + # FMTM API tests + get_user() + get_user_by_username() + convert_to_app_user() + convert_to_app_users() + get_user_role_by_user_id() + create_user_roles() + get_user_by_id()