diff --git a/gs/backend/api/v1/aro/endpoints/user.py b/gs/backend/api/v1/aro/endpoints/user.py index 321b0cdee..416da7a40 100644 --- a/gs/backend/api/v1/aro/endpoints/user.py +++ b/gs/backend/api/v1/aro/endpoints/user.py @@ -1,15 +1,14 @@ -from uuid import UUID - -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter from gs.backend.api.v1.aro.models.requests import UserRequest from gs.backend.api.v1.aro.models.responses import AllUsersResponse, UserResponse -from gs.backend.data.data_wrappers.aro_wrapper.aro_user_data_wrapper import ( - add_user, - delete_user_by_id, - update_user_by_id, -) -from gs.backend.data.data_wrappers.aro_wrapper.aro_user_data_wrapper import get_all_users as get_all_db_users + +# from gs.backend.data.data_wrappers.aro_wrapper.aro_user_data_wrapper import ( +# add_user, +# delete_user_by_id, +# update_user_by_id, +# ) +# from gs.backend.data.data_wrappers.aro_wrapper.aro_user_data_wrapper import get_all_users as get_all_db_users aro_user_router = APIRouter(tags=["ARO", "User Information"]) @@ -21,8 +20,9 @@ async def get_all_users() -> AllUsersResponse: :return: all users """ - users = get_all_db_users() - return AllUsersResponse(data=users) + # users = get_all_db_users() + # return AllUsersResponse(data=users) + return None @aro_user_router.post("/create_user", response_model=UserResponse) @@ -33,15 +33,16 @@ def create_user(payload: UserRequest) -> UserResponse: :return: returns the user created """ - user = add_user( - call_sign=payload.call_sign, - email=payload.email, - f_name=payload.first_name, - l_name=payload.last_name, - phone_number=payload.phone_number, - ) + # user = add_user( + # call_sign=payload.call_sign, + # email=payload.email, + # f_name=payload.first_name, + # l_name=payload.last_name, + # phone_number=payload.phone_number, + # ) - return UserResponse(data=user) + # return UserResponse(data=user) + return None @aro_user_router.put("/update_user/{userid}", response_model=UserResponse) @@ -52,18 +53,19 @@ def update_user(userid: str, payload: UserRequest) -> UserResponse: :param payload: The data used to update a user :return: returns the user updated """ - try: - user = update_user_by_id( - userid=UUID(userid), - call_sign=payload.call_sign, - email=payload.email, - f_name=payload.first_name, - l_name=payload.last_name, - phone_number=payload.phone_number, - ) - return UserResponse(data=user) - except ValueError as e: - raise HTTPException(status_code=404, detail=str(e)) from e + # try: + # user = update_user_by_id( + # userid=UUID(userid), + # call_sign=payload.call_sign, + # email=payload.email, + # f_name=payload.first_name, + # l_name=payload.last_name, + # phone_number=payload.phone_number, + # ) + # return UserResponse(data=user) + # except ValueError as e: + # raise HTTPException(status_code=404, detail=str(e)) from e + return None @aro_user_router.delete("/delete_user/{userid}", response_model=AllUsersResponse) @@ -73,8 +75,9 @@ def delete_user(userid: str) -> AllUsersResponse: :param userid: The unique identifier of the user to be deleted :return: returns the user deleted """ - try: - users = delete_user_by_id(UUID(userid)) - return AllUsersResponse(data=users) - except ValueError as e: - raise HTTPException(status_code=404, detail=str(e)) from e + # try: + # users = delete_user_by_id(UUID(userid)) + # return AllUsersResponse(data=users) + # except ValueError as e: + # raise HTTPException(status_code=404, detail=str(e)) from e + return None diff --git a/gs/backend/api/v1/aro/models/responses.py b/gs/backend/api/v1/aro/models/responses.py index c2c02ea74..a99e1775b 100644 --- a/gs/backend/api/v1/aro/models/responses.py +++ b/gs/backend/api/v1/aro/models/responses.py @@ -1,19 +1,14 @@ -from pydantic import BaseModel +# class AllUsersResponse(BaseModel): +# """ +# The users response model. +# """ -from gs.backend.data.tables.aro_user_tables import AROUsers +# data: list[AROUsers] -class AllUsersResponse(BaseModel): - """ - The users response model. - """ +# class UserResponse(BaseModel): +# """ +# Single user response model. +# """ - data: list[AROUsers] - - -class UserResponse(BaseModel): - """ - Single user response model. - """ - - data: AROUsers +# data: AROUsers diff --git a/gs/backend/api/v1/mcc/endpoints/main_commands.py b/gs/backend/api/v1/mcc/endpoints/main_commands.py index 9b6edd3a9..24a0a7e91 100644 --- a/gs/backend/api/v1/mcc/endpoints/main_commands.py +++ b/gs/backend/api/v1/mcc/endpoints/main_commands.py @@ -2,7 +2,7 @@ from fastapi_cache.decorator import cache from gs.backend.api.v1.mcc.models.responses import MainCommandsResponse -from gs.backend.data.data_wrappers.mcc_wrappers.main_command_wrapper import get_all_main_commands +from gs.backend.data.data_wrappers.wrappers import MainCommandWrapper main_commands_router = APIRouter(tags=["MCC", "Main Commands"]) @@ -15,5 +15,5 @@ async def get_main_commands() -> MainCommandsResponse: :return: list of all commands """ - items = get_all_main_commands() + items = MainCommandWrapper().get_all() return MainCommandsResponse(data=items) diff --git a/gs/backend/data/data_wrappers/aro_wrapper/aro_request_wrapper.py b/gs/backend/data/data_wrappers/aro_wrapper/aro_request_wrapper.py deleted file mode 100644 index 113a9e21f..000000000 --- a/gs/backend/data/data_wrappers/aro_wrapper/aro_request_wrapper.py +++ /dev/null @@ -1,73 +0,0 @@ -from datetime import datetime -from decimal import Decimal -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.enums.aro_requests import ARORequestStatus -from gs.backend.data.tables.transactional_tables import ARORequest - - -def get_all_requests() -> list[ARORequest]: - """ - Get all the requests from aro - """ - with get_db_session() as session: - requests = list(session.exec(select(ARORequest)).all()) - return requests - - -def add_request( - aro_id: UUID, - long: Decimal, - lat: Decimal, - created_on: datetime, - request_sent_obc: datetime, - taken_date: datetime, - transmission: datetime, - status: ARORequestStatus, -) -> ARORequest: - """ - Add a request - - :param long: the longitude represented as a decimal of max 3 decimal places - :param lat: the latitude represented as a decimal of max 3 decimal places - :param created_on: datetime object representing the date this request was made. defaults to now - :param taken_date: datetime object representing the date that this picture was taken on - :param taken_date: datetime object representing the date that this picture was trasmitted - :param status: the status of the request, can only be from the requets in ARORequestStatus - """ - with get_db_session() as session: - request = ARORequest( - aro_id=aro_id, - latitude=lat, - longitude=long, - created_on=created_on, - request_sent_to_obc_on=request_sent_obc, - taken_date=taken_date, - transmission=transmission, - status=status, - ) - - session.add(request) - session.commit() - session.refresh(request) - return request - - -def delete_request_by_id(request_id: str) -> list[ARORequest]: - """ - Delete a request based on id - - :param request_id: unique identifier of the request - """ - with get_db_session() as session: - request = session.get(ARORequest, request_id) - if request: - session.delete(request) - session.commit() - else: - raise ValueError("Request not found, ID does not exist") - - return get_all_requests() diff --git a/gs/backend/data/data_wrappers/aro_wrapper/aro_user_auth_token_wrapper.py b/gs/backend/data/data_wrappers/aro_wrapper/aro_user_auth_token_wrapper.py deleted file mode 100644 index 0f0d52324..000000000 --- a/gs/backend/data/data_wrappers/aro_wrapper/aro_user_auth_token_wrapper.py +++ /dev/null @@ -1,54 +0,0 @@ -from datetime import datetime -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.enums.aro_auth_token import AROAuthToken as AROEnums -from gs.backend.data.tables.aro_user_tables import AROUserAuthToken - - -def get_all_auth_tokens() -> list[AROUserAuthToken]: - """ - Get all the auth tokens - """ - with get_db_session() as session: - auth_tokens = list(session.exec(select(AROUserAuthToken)).all()) - return auth_tokens - - -def add_auth_token(token: str, user_data_id: UUID, expiry: datetime, auth_type: AROEnums) -> AROUserAuthToken: - """ - Add auth token to the db - - :param token: the auth token we want to add - :param user_data_id: UUID which identifies which user this auth token belongs to - :param expiry: the date in which this token expires - :param auth_type: the type of auth token this is, can only be from AROAuthToken - """ - with get_db_session() as session: - auth_token = AROUserAuthToken(token=token, user_data_id=user_data_id, expiry=expiry, auth_type=auth_type) - - session.add(auth_token) - session.commit() - session.refresh(auth_token) - return auth_token - - -def delete_auth_token_by_id(token_id: UUID) -> list[AROUserAuthToken]: - """ - Delete the auth token based on the token id - - :param token_id: the unique identifier for a particular auth token - """ - - with get_db_session() as session: - auth_token = session.get(AROUserAuthToken, token_id) - - if auth_token: - session.delete(auth_token) - session.commit() - else: - print("Token does not exist") - - return get_all_auth_tokens() diff --git a/gs/backend/data/data_wrappers/aro_wrapper/aro_user_data_wrapper.py b/gs/backend/data/data_wrappers/aro_wrapper/aro_user_data_wrapper.py deleted file mode 100644 index cebce12c7..000000000 --- a/gs/backend/data/data_wrappers/aro_wrapper/aro_user_data_wrapper.py +++ /dev/null @@ -1,97 +0,0 @@ -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.aro_user_tables import AROUsers - - -# selects all objects of type AROUser from db and returns them in list -def get_all_users() -> list[AROUsers]: - """ - Gets all user - """ - with get_db_session() as session: - users = list(session.exec(select(AROUsers)).all()) - return users - - -# adds user to database of type AROUser then fetches the user from database -# so that the user now has an assigned ID -def add_user(call_sign: str, email: str, f_name: str, l_name: str, phone_number: str) -> AROUsers: - """ - Add a new user to the AROUser table in database - - :param call_sign: a 6 character string such as ABCDEF - :param email: unique email which is bound to the user - :param f_name: first name of user - :param l_name: last name of user - :param phone_numer: phone number of user - """ - with get_db_session() as session: - # check if the user already exists with email as it is unique - existing_user = session.exec(select(AROUsers).where(AROUsers.email == email)).first() - - if existing_user: - raise ValueError("User already exsits based on email") - - user = AROUsers( - call_sign=call_sign, email=email, first_name=f_name, last_name=l_name, phone_number=phone_number - ) - - session.add(user) - session.commit() - session.refresh(user) - return user - - -# updates user into database of type AROUser then fetches the user from database -def update_user_by_id( - userid: UUID, call_sign: str, email: str, f_name: str, l_name: str, phone_number: str -) -> AROUsers: - """ - Update an existing user in the AROUser table in database - - :param userid: the unique identifier of the user to be updated - :param call_sign: a 6 character string such as ABCDEF - :param email: unique email which is bound to the user - :param f_name: first name of user - :param l_name: last name of user - :param phone_numer: phone number of user - """ - with get_db_session() as session: - # check if the user already exists with email as it is unique - user = session.exec(select(AROUsers).where(AROUsers.id == userid)).first() - - if not user: - raise ValueError("User does not exist based on user ID") - - user.call_sign = call_sign - user.email = email - user.first_name = f_name - user.last_name = l_name - user.phone_number = phone_number - - session.add(user) - session.commit() - session.refresh(user) - return user - - -# deletes the user with given id and returns the remaining users -def delete_user_by_id(userid: UUID) -> list[AROUsers]: - """ - Use the user.id to delete a user from table - - :param userid: identifier unique to the user - """ - with get_db_session() as session: - user = session.get(AROUsers, userid) - - if user: - session.delete(user) - session.commit() - else: - raise ValueError("User ID does not exist") - - return get_all_users() diff --git a/gs/backend/data/data_wrappers/aro_wrapper/aro_user_login_wrapper.py b/gs/backend/data/data_wrappers/aro_wrapper/aro_user_login_wrapper.py deleted file mode 100644 index 6b5c26fbb..000000000 --- a/gs/backend/data/data_wrappers/aro_wrapper/aro_user_login_wrapper.py +++ /dev/null @@ -1,63 +0,0 @@ -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.aro_user_tables import AROUserLogin - - -def get_all_logins() -> list[AROUserLogin]: - """ - Gets all the logins - """ - with get_db_session() as session: - user_logins = list(session.exec(select(AROUserLogin)).all()) - return user_logins - - -def add_login(email: str, pwd: str, hash_algo: str, user_data_id: UUID, email_verification_token: str) -> AROUserLogin: - """ - Add a new user login - - :param email: the email which the user used to sign up - :param pwd: the password the user set - :hash_algo: the hashing algorithm used to encrypt the password w the salt - :user_data_id: the unique identifier which binds this login to the user which created it - :email_verification_token: email verification token - """ - with get_db_session() as session: - # check if the user exists already - existing_login = session.exec(select(AROUserLogin).where(AROUserLogin.email == email)).first() - - if existing_login: - raise ValueError("User login already exists based on email") - - user_login = AROUserLogin( - email=email, - password=pwd, - hashing_algorithm_name=hash_algo, - user_data_id=user_data_id, - email_verification_token=email_verification_token, - ) - - session.add(user_login) - session.commit() - session.refresh(user_login) - return user_login - - -def delete_login_by_id(loginid: UUID) -> list[AROUserLogin]: - """ - Use the .id to delete a user from table - - :param loginid: unique identifier of the target login - """ - with get_db_session() as session: - user_login = session.get(AROUserLogin, loginid) - if user_login: - session.delete(user_login) - session.commit() - else: - raise ValueError("Login ID does not exist") - - return get_all_logins() diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/commands_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/commands_wrapper.py deleted file mode 100644 index 9ea20a7a9..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/commands_wrapper.py +++ /dev/null @@ -1,49 +0,0 @@ -from typing import Any -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.transactional_tables import Commands - - -def get_all_commands() -> list[Commands]: - """ - Get all data wrapper for Commands - - :return: a list of all commands - """ - with get_db_session() as session: - commands = list(session.exec(select(Commands)).all()) - return commands - - -def create_commands(command_data: dict[str, Any]) -> Commands: - """ - Post data wrapper for Commands - - :param command_data: the JSON object of the command to be created - :return: the newly created command - """ - with get_db_session() as session: - command = Commands(**command_data) - session.add(command) - session.commit() - session.refresh(command) - return command - - -def delete_commands_by_id(command_id: UUID) -> Commands: - """ - Delete data wrapper for Commands - - :param command_id: UUID of command to be deleted - :return: the deleted command - """ - with get_db_session() as session: - command = session.get(Commands, command_id) - if not command: - raise ValueError("Command not found.") - session.delete(command) - session.commit() - return command diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/comms_session_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/comms_session_wrapper.py deleted file mode 100644 index 069aad3b6..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/comms_session_wrapper.py +++ /dev/null @@ -1,49 +0,0 @@ -from typing import Any -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.transactional_tables import CommsSession - - -def get_all_comms_sessions() -> list[CommsSession]: - """ - Get all data wrapper for CommsSession - - :return: a list of all sessions - """ - with get_db_session() as session: - sessions = list(session.exec(select(CommsSession)).all()) - return sessions - - -def create_comms_session(session_data: dict[str, Any]) -> CommsSession: - """ - Post data wrapper for CommsSession - - :param session_data: the JSON object of the comms_session to be created - :return: the newly created comms_session - """ - with get_db_session() as session: - comms_session = CommsSession(**session_data) - session.add(comms_session) - session.commit() - session.refresh(comms_session) - return comms_session - - -def delete_telemetry_by_id(session_id: UUID) -> CommsSession: - """ - Delete data wrapper for CommsSession - - :param session_id: UUID of session to be deleted - :return: the deleted session - """ - with get_db_session() as session: - comms_session = session.get(CommsSession, session_id) - if not comms_session: - raise ValueError("Comms session not found.") - session.delete(comms_session) - session.commit() - return comms_session diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/main_command_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/main_command_wrapper.py deleted file mode 100644 index 18352268c..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/main_command_wrapper.py +++ /dev/null @@ -1,48 +0,0 @@ -from typing import Any - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.main_tables import MainCommand - - -def get_all_main_commands() -> list[MainCommand]: - """ - Get all data wrapper for MainCommand - - :return: a list of all main_commands - """ - with get_db_session() as session: - commands = list(session.exec(select(MainCommand)).all()) - return commands - - -def create_main_command(command_data: dict[str, Any]) -> MainCommand: - """ - Post data wrapper for MainCommand - - :param command_data: the JSON object of the main_command to be created - :return: the newly created main_command - """ - with get_db_session() as session: - command = MainCommand(**command_data) - session.add(command) - session.commit() - session.refresh(command) - return command - - -def delete_main_command_by_id(command_id: int) -> MainCommand: - """ - Delete data wrapper for MainCommand - - :param command_id: id of main_command to be deleted - :return: the deleted main_command - """ - with get_db_session() as session: - command = session.get(MainCommand, command_id) - if not command: - raise ValueError("Main command not found.") - session.delete(command) - session.commit() - return command diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/main_telemetry_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/main_telemetry_wrapper.py deleted file mode 100644 index d93e507ad..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/main_telemetry_wrapper.py +++ /dev/null @@ -1,48 +0,0 @@ -from typing import Any - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.main_tables import MainTelemetry - - -def get_all_main_telemetries() -> list[MainTelemetry]: - """ - Get all data wrapper for MainTelemetry - - :return: a list of all main_telemetries - """ - with get_db_session() as session: - telemetries = list(session.exec(select(MainTelemetry)).all()) - return telemetries - - -def create_main_telemetry(telemetry_data: dict[str, Any]) -> MainTelemetry: - """ - Post data wrapper for MainTelemetry - - :param command_data: the JSON object of the main_telemetry to be created - :return: the newly created main_telemetry - """ - with get_db_session() as session: - telemetry = MainTelemetry(**telemetry_data) - session.add(telemetry) - session.commit() - session.refresh(telemetry) - return telemetry - - -def delete_main_telemetry_by_id(telemetry_id: int) -> MainTelemetry: - """ - Delete data wrapper for MainTelemetry - - :param command_id: id of main_telemetry to be deleted - :return: the deleted main_telemetry - """ - with get_db_session() as session: - telemetry = session.get(MainTelemetry, telemetry_id) - if not telemetry: - raise ValueError("Main telemetry not found.") - session.delete(telemetry) - session.commit() - return telemetry diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/packet_commands_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/packet_commands_wrapper.py deleted file mode 100644 index 260916497..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/packet_commands_wrapper.py +++ /dev/null @@ -1,49 +0,0 @@ -from typing import Any -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.transactional_tables import PacketCommands - - -def get_all_packet_commands() -> list[PacketCommands]: - """ - Get all data wrapper for PacketCommands - - :return: a list of all packet_commands - """ - with get_db_session() as session: - commands = list(session.exec(select(PacketCommands)).all()) - return commands - - -def create_packet_command(command_data: dict[str, Any]) -> PacketCommands: - """ - Post data wrapper for PacketCommands - - :param command_data: the JSON object of the packet_command to be created - :return: the newly created packet_command - """ - with get_db_session() as session: - command = PacketCommands(**command_data) - session.add(command) - session.commit() - session.refresh(command) - return command - - -def delete_packet_command_by_id(command_id: UUID) -> PacketCommands: - """ - Delete data wrapper for PacketCommands - - :param command_id: UUID of packet_command to be deleted - :return: the deleted packet_command - """ - with get_db_session() as session: - command = session.get(PacketCommands, command_id) - if not command: - raise ValueError("Packet command not found.") - session.delete(command) - session.commit() - return command diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/packet_telemetry_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/packet_telemetry_wrapper.py deleted file mode 100644 index 31c2dae11..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/packet_telemetry_wrapper.py +++ /dev/null @@ -1,49 +0,0 @@ -from typing import Any -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.transactional_tables import PacketTelemetry - - -def get_all_packet_telemetries() -> list[PacketTelemetry]: - """ - Get all data wrapper for PacketTelemetry - - :return: a list of all packet_telemetries - """ - with get_db_session() as session: - telemetries = list(session.exec(select(PacketTelemetry)).all()) - return telemetries - - -def create_packet_telemetry(telemetry_data: dict[str, Any]) -> PacketTelemetry: - """ - Post data wrapper for PacketTelemetry - - :param command_data: the JSON object of the packet_telemetry to be created - :return: the newly created packet_telemetry - """ - with get_db_session() as session: - telemetry = PacketTelemetry(**telemetry_data) - session.add(telemetry) - session.commit() - session.refresh(telemetry) - return telemetry - - -def delete_packet_telemetry_by_id(telemetry_id: UUID) -> PacketTelemetry: - """ - Delete data wrapper for PacketTelemetry - - :param command_id: UUID of packet_telemetry to be deleted - :return: the deleted packet_telemetry - """ - with get_db_session() as session: - telemetry = session.get(PacketTelemetry, telemetry_id) - if not telemetry: - raise ValueError("Packet telemetry not found.") - session.delete(telemetry) - session.commit() - return telemetry diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/packet_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/packet_wrapper.py deleted file mode 100644 index c44029d62..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/packet_wrapper.py +++ /dev/null @@ -1,49 +0,0 @@ -from typing import Any -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.transactional_tables import Packet - - -def get_all_packets() -> list[Packet]: - """ - Get all data wrapper for Packet - - :return: a list of all packets - """ - with get_db_session() as session: - packets = list(session.exec(select(Packet)).all()) - return packets - - -def create_packet(packet_data: dict[str, Any]) -> Packet: - """ - Post data wrapper for Packet - - :param packet_data: the JSON object of the packet to be created - :return: the newly created packet - """ - with get_db_session() as session: - packet = Packet(**packet_data) - session.add(packet) - session.commit() - session.refresh(packet) - return packet - - -def delete_packet_by_id(packet_id: UUID) -> Packet: - """ - Delete data wrapper for Packet - - :param packet_id: UUID of packet to be deleted - :return: the deleted packet - """ - with get_db_session() as session: - packet = session.get(Packet, packet_id) - if not packet: - raise ValueError("Packet not found.") - session.delete(packet) - session.commit() - return packet diff --git a/gs/backend/data/data_wrappers/mcc_wrappers/telemetry_wrapper.py b/gs/backend/data/data_wrappers/mcc_wrappers/telemetry_wrapper.py deleted file mode 100644 index 49066c1c2..000000000 --- a/gs/backend/data/data_wrappers/mcc_wrappers/telemetry_wrapper.py +++ /dev/null @@ -1,49 +0,0 @@ -from typing import Any -from uuid import UUID - -from sqlmodel import select - -from gs.backend.data.database.engine import get_db_session -from gs.backend.data.tables.transactional_tables import Telemetry - - -def get_all_telemetries() -> list[Telemetry]: - """ - Get all data wrapper for Telemetry - - :return: a list of all telemetries - """ - with get_db_session() as session: - telemetries = list(session.exec(select(Telemetry)).all()) - return telemetries - - -def create_telemetry(telemetry_data: dict[str, Any]) -> Telemetry: - """ - Post data wrapper for Telemetry - - :param telemetry_data: the JSON object of the telemetry to be created - :return: the newly created telemetry - """ - with get_db_session() as session: - telemetry = Telemetry(**telemetry_data) - session.add(telemetry) - session.commit() - session.refresh(telemetry) - return telemetry - - -def delete_telemetry_by_id(telemetry_id: UUID) -> Telemetry: - """ - Delete data wrapper for Telemetry - - :param telemetry_id: UUID of telemetry to be deleted - :return: the deleted telemetry - """ - with get_db_session() as session: - telemetry = session.get(Telemetry, telemetry_id) - if not telemetry: - raise ValueError("Telemetry not found.") - session.delete(telemetry) - session.commit() - return telemetry diff --git a/python_test/conftest.py b/python_test/conftest.py index 1bf9d8bd4..0f50cdb0a 100644 --- a/python_test/conftest.py +++ b/python_test/conftest.py @@ -49,19 +49,7 @@ def test_get_db_session(monkeypatch, db_session: Session): When testing any database function that requires the `get_db_session()` function, you must add the module path to the list below. """ path_list: list[str] = [ - "gs.backend.data.data_wrappers.abstract_wrapper", - "gs.backend.data.data_wrappers.aro_wrapper.aro_request_wrapper", - "gs.backend.data.data_wrappers.aro_wrapper.aro_user_data_wrapper", - "gs.backend.data.data_wrappers.aro_wrapper.aro_user_auth_token_wrapper", - "gs.backend.data.data_wrappers.aro_wrapper.aro_user_login_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.commands_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.comms_session_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.main_command_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.main_telemetry_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.packet_commands_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.packet_telemetry_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.packet_wrapper", - "gs.backend.data.data_wrappers.mcc_wrappers.telemetry_wrapper", + "gs.backend.data.data_wrappers.wrappers", ] for path in path_list: diff --git a/python_test/test_aro_data_wrapper.py b/python_test/test_aro_data_wrapper.py deleted file mode 100644 index e5c67389d..000000000 --- a/python_test/test_aro_data_wrapper.py +++ /dev/null @@ -1,117 +0,0 @@ -# python_test/test_aro_data_wrapper.py -from datetime import datetime, timedelta -from decimal import Decimal -from uuid import UUID - -import pytest - -# funcs under test -from gs.backend.data.data_wrappers.aro_wrapper.aro_request_wrapper import ( - add_request, - get_all_requests, -) -from gs.backend.data.data_wrappers.aro_wrapper.aro_user_auth_token_wrapper import ( - add_auth_token, - get_all_auth_tokens, -) -from gs.backend.data.data_wrappers.aro_wrapper.aro_user_data_wrapper import add_user -from gs.backend.data.data_wrappers.aro_wrapper.aro_user_login_wrapper import add_login -from gs.backend.data.enums.aro_auth_token import AROAuthToken -from gs.backend.data.enums.aro_requests import ARORequestStatus - - -def test_user_creation(): - user = add_user( - call_sign="KEVWAN", - email="kevian@gmail.com", - f_name="kevin", - l_name="wan", - phone_number="8888888888", - ) - assert user.email == "kevian@gmail.com" - assert user.call_sign == "KEVWAN" - assert user.first_name == "kevin" - assert user.last_name == "wan" - assert user.phone_number == "8888888888" - - -def test_login_creation(): - user = add_user( - call_sign="BEVWAN", - email="bevian@gmail.com", - f_name="bevin", - l_name="ban", - phone_number="9999999999", - ) - login = add_login( - email=user.email, - pwd="password", - hash_algo="kevalgo", - user_data_id=user.id, - email_verification_token="abcABC19201", - ) - assert login.email == "bevian@gmail.com" - assert login.password == "password" - assert login.hashing_algorithm_name == "kevalgo" - assert login.user_data_id == user.id - assert login.email_verification_token == "abcABC19201" - - -def test_auth_token_creation(): - user = add_user( - call_sign="TOKUSR", - email="tokuser@example.com", - f_name="tok", - l_name="user", - phone_number="1112223333", - ) - before = {t.id for t in get_all_auth_tokens()} - expiry = datetime.now() + timedelta(hours=2) - token = add_auth_token( - token="tok123", - user_data_id=user.id, - expiry=expiry, - auth_type=AROAuthToken.DUMMY, - ) - after = {t.id for t in get_all_auth_tokens()} - assert token.token == "tok123" - assert isinstance(token.user_data_id, UUID) - assert token.user_data_id == user.id - assert isinstance(token.expiry, datetime) - assert token.auth_type == AROAuthToken.DUMMY - assert token.id in after - assert token.id not in before - - -def test_request_creation(): - user = add_user( - call_sign="BEVWAN", - email="bevian@gmail.com", - f_name="bevin", - l_name="ban", - phone_number="9999999999", - ) - - created_on = datetime.now() - request_sent_obc = created_on + timedelta(minutes=1) - taken_date = created_on + timedelta(minutes=2) - transmission = created_on + timedelta(minutes=3) - before = {r.id for r in get_all_requests()} - req = add_request( - aro_id=user.id, - long=Decimal("123.456"), - lat=Decimal("49.282"), - created_on=created_on, - request_sent_obc=request_sent_obc, - taken_date=taken_date, - transmission=transmission, - status=ARORequestStatus.PENDING, - ) - after = {r.id for r in get_all_requests()} - assert req.longitude == Decimal("123.456") - assert req.latitude == Decimal("49.282") - assert abs((req.created_on - created_on).total_seconds()) < 1 - assert abs((req.request_sent_to_obc_on - request_sent_obc).total_seconds()) < 1 - assert req.status == ARORequestStatus.PENDING - assert req.id in after - assert req.id not in before diff --git a/python_test/test_aro_user_api.py b/python_test/test_aro_user_api.py deleted file mode 100644 index 07754550d..000000000 --- a/python_test/test_aro_user_api.py +++ /dev/null @@ -1,141 +0,0 @@ -import pytest -from fastapi.testclient import TestClient -from gs.backend.main import app - - -@pytest.fixture -def client(): - return TestClient(app) - - -# Test data for user 1 -@pytest.fixture -def user1_data(): - return { - "call_sign": "ABCDEF", - "email": "bob@test.com", - "first_name": "Bob", - "last_name": "Smith", - "phone_number": "123456789", - } - - -# Test data for user 2 -@pytest.fixture -def user2_data(): - return { - "call_sign": "KEVWAN", - "email": "kevian@gmail.com", - "first_name": "kevin", - "last_name": "wan", - "phone_number": "8888888888", - } - - -# Test creating user1 -@pytest.fixture -def test_user1_creation(client, user1_data): - response = client.post( - "/api/v1/aro/user/create_user", json=user1_data, headers={"Content-Type": "application/json"} - ) - - assert response.status_code == 200 - user = response.json()["data"] - assert user["email"] == user1_data["email"] - assert user["call_sign"] == user1_data["call_sign"] - assert user["first_name"] == user1_data["first_name"] - assert user["last_name"] == user1_data["last_name"] - assert user["phone_number"] == user1_data["phone_number"] - - return user - - -# Test creating user2 -@pytest.fixture -def test_user2_creation(client, user2_data): - response = client.post( - "/api/v1/aro/user/create_user", json=user2_data, headers={"Content-Type": "application/json"} - ) - - assert response.status_code == 200 - user = response.json()["data"] - assert user["email"] == user2_data["email"] - assert user["call_sign"] == user2_data["call_sign"] - assert user["first_name"] == user2_data["first_name"] - assert user["last_name"] == user2_data["last_name"] - assert user["phone_number"] == user2_data["phone_number"] - - return user - - -# Test updating user1 (depends on test_user1_creation) -@pytest.fixture -def test_user1_update(client, test_user1_creation): - user_id = test_user1_creation["id"] - update_data = { - "call_sign": "123456", - "email": "bob2@test.com", - "first_name": "Rob", - "last_name": "Smith", - "phone_number": "234567890", - } - res = client.put( - f"/api/v1/aro/user/update_user/{user_id}", json=update_data, headers={"Content-Type": "application/json"} - ) - - assert res.status_code == 200 - updated_user = res.json()["data"] - assert updated_user["email"] == update_data["email"] - assert updated_user["call_sign"] == update_data["call_sign"] - assert updated_user["first_name"] == update_data["first_name"] - assert updated_user["last_name"] == update_data["last_name"] - assert updated_user["phone_number"] == update_data["phone_number"] - - return updated_user - - -# Test getting all users (after creating user1 and user2, and updating user1 to ensure creation and update work oncorrect user objects) -@pytest.fixture -def test_get_users(client, test_user1_update, test_user2_creation): - res = client.get("/api/v1/aro/user/get_all_users") - - assert res.status_code == 200 - all_users = res.json()["data"] - assert len(all_users) == 2 - - # Check user1 - user1_id = test_user1_update["id"] - user1_from_response = next(user for user in all_users if user["id"] == user1_id) - assert user1_from_response["call_sign"] == test_user1_update["call_sign"] - assert user1_from_response["email"] == test_user1_update["email"] - assert user1_from_response["first_name"] == test_user1_update["first_name"] - assert user1_from_response["last_name"] == test_user1_update["last_name"] - assert user1_from_response["phone_number"] == test_user1_update["phone_number"] - - # Check user2 - user2_id = test_user2_creation["id"] - user2_from_response = next(user for user in all_users if user["id"] == user2_id) - assert user2_from_response["email"] == test_user2_creation["email"] - assert user2_from_response["call_sign"] == test_user2_creation["call_sign"] - assert user2_from_response["first_name"] == test_user2_creation["first_name"] - assert user2_from_response["last_name"] == test_user2_creation["last_name"] - assert user2_from_response["phone_number"] == test_user2_creation["phone_number"] - - -# Test deleting user1 (after test_get_users to ensure both users exist) -def test_user1_deletion(client, test_user1_update, test_user2_creation, test_get_users): - user_id = test_user1_update["id"] - res = client.delete(f"/api/v1/aro/user/delete_user/{user_id}", headers={"Content-Type": "application/json"}) - - assert res.status_code == 200 - all_users = res.json()["data"] - assert len(all_users) == 1 - assert all_users[0]["id"] != user_id # Ensure user1 is deleted - assert all_users[0]["id"] == test_user2_creation["id"] - assert all_users[0]["email"] == test_user2_creation["email"] - assert all_users[0]["call_sign"] == test_user2_creation["call_sign"] - assert all_users[0]["first_name"] == test_user2_creation["first_name"] - assert all_users[0]["last_name"] == test_user2_creation["last_name"] - assert all_users[0]["phone_number"] == test_user2_creation["phone_number"] - - return all_users diff --git a/python_test/test_mcc_command_wrapper.py b/python_test/test_mcc_command_wrapper.py deleted file mode 100644 index b5b33606c..000000000 --- a/python_test/test_mcc_command_wrapper.py +++ /dev/null @@ -1,42 +0,0 @@ -import pytest -from gs.backend.data.data_wrappers.mcc_wrappers import main_command_wrapper as wrapper -from gs.backend.data.database import engine as prod_engine -from gs.backend.data.tables.main_tables import MAIN_SCHEMA_METADATA, MainCommand -from sqlmodel import Session, SQLModel, create_engine - - -def test_create_main_command(db_session): - data = {"id": 1, "name": "Test Command", "params": None, "format": None, "data_size": 4, "total_size": 4} - command = wrapper.create_main_command(data) - - assert command.id == 1 - assert command.name == "Test Command" - - # Verify it was actually inserted - stored = db_session.get(MainCommand, 1) - assert stored is not None - assert stored.name == "Test Command" - - -def test_get_all_main_commands(db_session): - # Insert a couple of rows manually - db_session.add(MainCommand(id=1, name="CmdA", params=None, format=None, data_size=1, total_size=1)) - db_session.add(MainCommand(id=2, name="CmdB", params="x", format="y", data_size=2, total_size=2)) - db_session.commit() - - commands = wrapper.get_all_main_commands() - - assert len(commands) == 2 - names = {c.name for c in commands} - assert names == {"CmdA", "CmdB"} - - -def test_delete_main_command_by_id(db_session): - # Insert one command - db_session.add(MainCommand(id=1, name="CmdX", params=None, format=None, data_size=1, total_size=1)) - db_session.commit() - - # Should delete successfully - result = wrapper.delete_main_command_by_id(1) - assert isinstance(result, MainCommand) - assert db_session.get(MainCommand, 1) is None