Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor CatalogueCategoryRepo unit tests #305 #307

Merged
merged 15 commits into from
Jul 26, 2024
Merged
Show file tree
Hide file tree
Changes from 14 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions inventory_management_system_api/models/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@

class SystemBase(BaseModel):
"""
Base database model for a System
Base database model for a system
"""

parent_id: Optional[CustomObjectIdField] = None
Expand All @@ -28,13 +28,13 @@ class SystemBase(BaseModel):

class SystemIn(CreatedModifiedTimeInMixin, SystemBase):
"""
Input database model for a System
Input database model for a system
"""


class SystemOut(CreatedModifiedTimeOutMixin, SystemBase):
"""
Output database model for a System
Output database model for a system
"""

id: StringObjectIdField = Field(alias="_id")
Expand Down
62 changes: 31 additions & 31 deletions inventory_management_system_api/repositories/catalogue_category.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,29 +72,6 @@ def create(self, catalogue_category: CatalogueCategoryIn, session: ClientSession
catalogue_category = self.get(str(result.inserted_id), session=session)
return catalogue_category

def delete(self, catalogue_category_id: str, session: ClientSession = None) -> None:
"""
Delete a catalogue category by its ID from a MongoDB database.

The method checks if the catalogue category has child elements and raises a `ChildElementsExistError` if it
does.

:param catalogue_category_id: The ID of the catalogue category to delete.
:param session: PyMongo ClientSession to use for database operations
:raises ChildElementsExistError: If the catalogue category has child elements.
:raises MissingRecordError: If the catalogue category doesn't exist.
"""
catalogue_category_id = CustomObjectId(catalogue_category_id)
if self.has_child_elements(catalogue_category_id, session=session):
raise ChildElementsExistError(
f"Catalogue category with ID {str(catalogue_category_id)} has child elements and cannot be deleted"
)

logger.info("Deleting catalogue category with ID: %s from the database", catalogue_category_id)
result = self._catalogue_categories_collection.delete_one({"_id": catalogue_category_id}, session=session)
if result.deleted_count == 0:
raise MissingRecordError(f"No catalogue category found with ID: {str(catalogue_category_id)}")

def get(self, catalogue_category_id: str, session: ClientSession = None) -> Optional[CatalogueCategoryOut]:
"""
Retrieve a catalogue category by its ID from a MongoDB database.
Expand Down Expand Up @@ -134,6 +111,20 @@ def get_breadcrumbs(self, catalogue_category_id: str, session: ClientSession = N
collection_name="catalogue_categories",
)

def list(self, parent_id: Optional[str], session: ClientSession = None) -> List[CatalogueCategoryOut]:
"""
Retrieve catalogue categories from a MongoDB database based on the provided filters.

:param parent_id: The parent_id to filter catalogue categories by.
:param session: PyMongo ClientSession to use for database operations
:return: A list of catalogue categories, or an empty list if no catalogue categories are returned by the
database.
"""
query = utils.list_query(parent_id, "catalogue categories")

catalogue_categories = self._catalogue_categories_collection.find(query, session=session)
return [CatalogueCategoryOut(**catalogue_category) for catalogue_category in catalogue_categories]

def update(
self, catalogue_category_id: str, catalogue_category: CatalogueCategoryIn, session: ClientSession = None
) -> CatalogueCategoryOut:
Expand Down Expand Up @@ -192,19 +183,28 @@ def update(
catalogue_category = self.get(str(catalogue_category_id), session=session)
return catalogue_category

def list(self, parent_id: Optional[str], session: ClientSession = None) -> List[CatalogueCategoryOut]:
def delete(self, catalogue_category_id: str, session: ClientSession = None) -> None:
"""
Retrieve catalogue categories from a MongoDB database based on the provided filters.
Delete a catalogue category by its ID from a MongoDB database.

:param parent_id: The parent_id to filter catalogue categories by.
The method checks if the catalogue category has child elements and raises a `ChildElementsExistError` if it
does.

:param catalogue_category_id: The ID of the catalogue category to delete.
:param session: PyMongo ClientSession to use for database operations
:return: A list of catalogue categories, or an empty list if no catalogue categories are returned by the
database.
:raises ChildElementsExistError: If the catalogue category has child elements.
:raises MissingRecordError: If the catalogue category doesn't exist.
"""
query = utils.list_query(parent_id, "catalogue categories")
catalogue_category_id = CustomObjectId(catalogue_category_id)
if self.has_child_elements(catalogue_category_id, session=session):
raise ChildElementsExistError(
f"Catalogue category with ID {str(catalogue_category_id)} has child elements and cannot be deleted"
)

catalogue_categories = self._catalogue_categories_collection.find(query, session=session)
return [CatalogueCategoryOut(**catalogue_category) for catalogue_category in catalogue_categories]
logger.info("Deleting catalogue category with ID: %s from the database", catalogue_category_id)
result = self._catalogue_categories_collection.delete_one({"_id": catalogue_category_id}, session=session)
if result.deleted_count == 0:
raise MissingRecordError(f"No catalogue category found with ID: {str(catalogue_category_id)}")

def _is_duplicate_catalogue_category(
self,
Expand Down
72 changes: 36 additions & 36 deletions inventory_management_system_api/repositories/system.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
"""
Module for providing a repository for managing Systems in a MongoDB database
Module for providing a repository for managing systems in a MongoDB database
"""

import logging
Expand All @@ -25,7 +25,7 @@

class SystemRepo:
"""
Repository for managing Systems in a MongoDB database
Repository for managing systems in a MongoDB database
"""

def __init__(self, database: DatabaseDep) -> None:
Expand All @@ -40,37 +40,37 @@ def __init__(self, database: DatabaseDep) -> None:

def create(self, system: SystemIn, session: ClientSession = None) -> SystemOut:
"""
Create a new System in a MongoDB database
Create a new system in a MongoDB database

If a parent system is specified by `parent_id`, then checks if that exists in the database and raises a
`MissingRecordError` if it doesn't exist. It also checks if a duplicate System is found within the parent
System and raises a `DuplicateRecordError` if it is.
`MissingRecordError` if it doesn't exist. It also checks if a duplicate system is found within the parent
system and raises a `DuplicateRecordError` if it is.

:param system: System to be created
:param session: PyMongo ClientSession to use for database operations
:return: Created System
:raises MissingRecordError: If the parent System specified by `parent_id` doesn't exist
:raises DuplicateRecordError: If a duplicate System is found within the parent System
:return: Created system
:raises MissingRecordError: If the parent system specified by `parent_id` doesn't exist
:raises DuplicateRecordError: If a duplicate system is found within the parent system
"""
parent_id = str(system.parent_id) if system.parent_id else None
if parent_id and not self.get(parent_id, session=session):
raise MissingRecordError(f"No parent System found with ID: {parent_id}")
raise MissingRecordError(f"No parent system found with ID: {parent_id}")

if self._is_duplicate_system(parent_id, system.code, session=session):
raise DuplicateRecordError("Duplicate System found within the parent System")
raise DuplicateRecordError("Duplicate system found within the parent system")

logger.info("Inserting the new System into the database")
logger.info("Inserting the new system into the database")
result = self._systems_collection.insert_one(system.model_dump(), session=session)
system = self.get(str(result.inserted_id), session=session)
return system

def get(self, system_id: str, session: ClientSession = None) -> Optional[SystemOut]:
"""
Retrieve a System by its ID from a MongoDB database
Retrieve a system by its ID from a MongoDB database

:param system_id: ID of the System to retrieve
:param system_id: ID of the system to retrieve
:param session: PyMongo ClientSession to use for database operations
:return: Retrieved System or `None` if not found
:return: Retrieved system or `None` if not found
"""
system_id = CustomObjectId(system_id)
logger.info("Retrieving system with ID: %s from the database", system_id)
Expand Down Expand Up @@ -101,11 +101,11 @@ def get_breadcrumbs(self, system_id: str, session: ClientSession = None) -> Brea

def list(self, parent_id: Optional[str], session: ClientSession = None) -> list[SystemOut]:
"""
Retrieve Systems from a MongoDB database based on the provided filters
Retrieve systems from a MongoDB database based on the provided filters

:param parent_id: parent_id to filter Systems by
:param parent_id: parent_id to filter systems by
:param session: PyMongo ClientSession to use for database operations
:return: List of Systems or an empty list if no Systems are retrieved
:return: List of systems or an empty list if no systems are retrieved
"""
query = utils.list_query(parent_id, "systems")

Expand All @@ -115,26 +115,26 @@ def list(self, parent_id: Optional[str], session: ClientSession = None) -> list[
def update(self, system_id: str, system: SystemIn, session: ClientSession = None) -> SystemOut:
"""Update a system by its ID in a MongoDB database

:param system_id: ID of the System to update
:param system_id: ID of the system to update
:param system: System containing the update data
:param session: PyMongo ClientSession to use for database operations
:return: The updated System
:raises MissingRecordError: If the parent System specified by `parent_id` doesn't exist
:raises DuplicateRecordError: If a duplicate System is found within the parent System
:return: The updated system
:raises MissingRecordError: If the parent system specified by `parent_id` doesn't exist
:raises DuplicateRecordError: If a duplicate system is found within the parent system
:raises InvalidActionError: If attempting to change the `parent_id` to one of its own child system ids
"""
system_id = CustomObjectId(system_id)

parent_id = str(system.parent_id) if system.parent_id else None
if parent_id and not self.get(parent_id, session=session):
raise MissingRecordError(f"No parent System found with ID: {parent_id}")
raise MissingRecordError(f"No parent system found with ID: {parent_id}")

stored_system = self.get(str(system_id), session=session)
moving_system = parent_id != stored_system.parent_id
if (system.name != stored_system.name or moving_system) and self._is_duplicate_system(
parent_id, system.code, system_id, session=session
):
raise DuplicateRecordError("Duplicate System found within the parent System")
raise DuplicateRecordError("Duplicate system found within the parent system")

# Prevent a system from being moved to one of its own children
if moving_system:
Expand All @@ -157,14 +157,14 @@ def update(self, system_id: str, system: SystemIn, session: ClientSession = None

def delete(self, system_id: str, session: ClientSession = None) -> None:
"""
Delete a System by its ID from a MongoDB database
Delete a system by its ID from a MongoDB database

The method checks if the system has any child and raises a `ChildElementsExistError` if it does

:param system_id: ID of the System to delete
:param system_id: ID of the system to delete
:param session: PyMongo ClientSession to use for database operations
:raises ChildElementsExistError: If the System has child elements
:raises MissingRecordError: If the System doesn't exist
:raises ChildElementsExistError: If the system has child elements
:raises MissingRecordError: If the system doesn't exist
"""
system_id = CustomObjectId(system_id)
if self._has_child_elements(system_id, session=session):
Expand All @@ -173,21 +173,21 @@ def delete(self, system_id: str, session: ClientSession = None) -> None:
logger.info("Deleting system with ID: %s from the database", system_id)
result = self._systems_collection.delete_one({"_id": system_id}, session=session)
if result.deleted_count == 0:
raise MissingRecordError(f"No System found with ID: {str(system_id)}")
raise MissingRecordError(f"No system found with ID: {str(system_id)}")

def _is_duplicate_system(
self, parent_id: Optional[str], code: str, system_id: CustomObjectId = None, session: ClientSession = None
) -> bool:
"""
Check if a System with the same code already exists within the parent System
Check if a system with the same code already exists within the parent system

:param parent_id: ID of the parent System which can also be `None`
:param code: Code of the System to check for duplicates
:param parent_id: ID of the parent system which can also be `None`
:param code: Code of the system to check for duplicates
:param system_id: The ID of the system to check if the duplicate system found is itself.
:param session: PyMongo ClientSession to use for database operations
:return: `True` if a duplicate System code is found under the given parent, `False` otherwise
:return: `True` if a duplicate system code is found under the given parent, `False` otherwise
"""
logger.info("Checking if System with code '%s' already exists within the parent System", code)
logger.info("Checking if system with code '%s' already exists within the parent System", code)
if parent_id:
parent_id = CustomObjectId(parent_id)

Expand All @@ -198,11 +198,11 @@ def _is_duplicate_system(

def _has_child_elements(self, system_id: CustomObjectId, session: ClientSession = None) -> bool:
"""
Check if a System has any child System's or any Item's based on its ID
Check if a system has any child system's or any Item's based on its ID
joelvdavies marked this conversation as resolved.
Show resolved Hide resolved

:param system_id: ID of the System to check
:param system_id: ID of the system to check
:param session: PyMongo ClientSession to use for database operations
:return: True if the System has child elements, False otherwise
:return: True if the system has child elements, False otherwise
joelvdavies marked this conversation as resolved.
Show resolved Hide resolved
"""
logger.info("Checking if system with ID '%s' has child elements", str(system_id))

Expand Down
26 changes: 13 additions & 13 deletions inventory_management_system_api/routers/v1/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,31 +29,31 @@

@router.post(
path="",
summary="Create a new System",
response_description="The created System",
summary="Create a new system",
response_description="The created system",
status_code=status.HTTP_201_CREATED,
)
def create_system(system: SystemPostSchema, system_service: SystemServiceDep) -> SystemSchema:
# pylint: disable=missing-function-docstring
logger.info("Creating a new System")
logger.info("Creating a new system")
logger.debug("System data: %s", system)
try:
system = system_service.create(system)
return SystemSchema(**system.model_dump())
except (MissingRecordError, InvalidObjectIdError) as exc:
message = "The specified parent System does not exist"
message = "The specified parent system does not exist"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message) from exc
except DuplicateRecordError as exc:
message = "A System with the same name already exists within the same parent System"
message = "A system with the same name already exists within the same parent system"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc


@router.get(path="", summary="Get Systems", response_description="List of Systems")
@router.get(path="", summary="Get systems", response_description="List of systems")
def get_systems(
system_service: SystemServiceDep,
parent_id: Annotated[Optional[str], Query(description="Filter Systems by parent ID")] = None,
parent_id: Annotated[Optional[str], Query(description="Filter systems by parent ID")] = None,
) -> list[SystemSchema]:
# pylint: disable=missing-function-docstring
logger.info("Getting Systems")
Expand All @@ -69,12 +69,12 @@ def get_systems(
return []


@router.get(path="/{system_id}", summary="Get a System by ID", response_description="Single System")
@router.get(path="/{system_id}", summary="Get a system by ID", response_description="Single system")
def get_system(
system_id: Annotated[str, Path(description="ID of the System to get")], system_service: SystemServiceDep
system_id: Annotated[str, Path(description="ID of the system to get")], system_service: SystemServiceDep
) -> SystemSchema:
# pylint: disable=missing-function-docstring
logger.info("Getting System with ID: %s", system_service)
logger.info("Getting system with ID: %s", system_service)
message = "System not found"
try:
system = system_service.get(system_id)
Expand Down Expand Up @@ -110,7 +110,7 @@ def get_system_breadcrumbs(
# pylint: enable=duplicate-code


@router.patch(path="/{system_id}", summary="Update a System by ID", response_description="System updated successfully")
@router.patch(path="/{system_id}", summary="Update a system by ID", response_description="System updated successfully")
def partial_update_system(system_id: str, system: SystemPatchSchema, system_service: SystemServiceDep) -> SystemSchema:
# pylint: disable=missing-function-docstring
logger.info("Partially updating system with ID: %s", system_id)
Expand All @@ -121,15 +121,15 @@ def partial_update_system(system_id: str, system: SystemPatchSchema, system_serv
return SystemSchema(**updated_system.model_dump())
except (MissingRecordError, InvalidObjectIdError) as exc:
if system.parent_id and system.parent_id in str(exc) or "parent system" in str(exc).lower():
message = "The specified parent System does not exist"
message = "The specified parent system does not exist"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message) from exc

message = "System not found"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message) from exc
except DuplicateRecordError as exc:
message = "A System with the same name already exists within the parent System"
message = "A system with the same name already exists within the parent system"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc
# pylint:disable=duplicate-code
Expand Down
Loading