fix: Add env variable for test database

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"""

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"""

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):

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()
+

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()

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()

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!")