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 manufacturer tests #313

Merged
merged 9 commits into from
Jul 24, 2024
Merged
Show file tree
Hide file tree
Changes from 4 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
102 changes: 50 additions & 52 deletions inventory_management_system_api/repositories/manufacturer.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,79 +21,76 @@


class ManufacturerRepo:
"""Repository for managing manufacturer in MongoDb database"""
"""
Repository for managing manufacturers in a MongoDb database.
"""

def __init__(self, database: DatabaseDep) -> None:
"""Initialize the `ManufacturerRepo` with MongoDB database instance
"""
Initialise the `ManufacturerRepo` with a MongoDB database instance.
:param database: The database to use.
"""

self._database = database
self._manufacturers_collection: Collection = self._database.manufacturers
self._catalogue_item_collection: Collection = self._database.catalogue_items
self._catalogue_items_collection: Collection = self._database.catalogue_items

def create(self, manufacturer: ManufacturerIn, session: ClientSession = None) -> ManufacturerOut:
"""
Create a new manufacturer in MongoDB database
Create a new manufacturer in a MongoDB database.
:param manufacturer: The manufacturer to be created
:param session: PyMongo ClientSession to use for database operations
:return: The created manufacturer
:raises DuplicateRecordError: If a duplicate manufacturer is found within collection
:param manufacturer: The manufacturer to be created.
:param session: PyMongo ClientSession to use for database operations.
:raises DuplicateRecordError: If a duplicate manufacturer is found.
:return: The created manufacturer.
"""

if self._is_duplicate_manufacturer(manufacturer.code, session=session):
raise DuplicateRecordError("Duplicate manufacturer found")

logger.info("Inserting new manufacturer into database")
logger.info("Inserting the new manufacturer into database")

result = self._manufacturers_collection.insert_one(manufacturer.model_dump(), session=session)
manufacturer = self.get(str(result.inserted_id), session=session)

return manufacturer

def get(self, manufacturer_id: str, session: ClientSession = None) -> Optional[ManufacturerOut]:
"""Retrieve a manufacturer from database by its id
:param manufacturer_id: The ID of the manufacturer
:param session: PyMongo ClientSession to use for database operations
:return: The retrieved manufacturer, or `None` if not found
"""
Retrieve a manufacturer by its ID from a MondoDB database.
:param manufacturer_id: The ID of the manufacturer to retrieve.
:param session: PyMongo ClientSession to use for database operations.
:return: The retrieved manufacturer, or `None` if not found.
"""
manufacturer_id = CustomObjectId(manufacturer_id)

logger.info("Retrieving manufacturer with ID %s from database", manufacturer_id)
logger.info("Retrieving manufacturer with ID: %s from database", manufacturer_id)
manufacturer = self._manufacturers_collection.find_one({"_id": manufacturer_id}, session=session)
if manufacturer:
return ManufacturerOut(**manufacturer)
return None

def list(self, session: ClientSession = None) -> List[ManufacturerOut]:
"""Get all manufacturers from database
:param session: PyMongo ClientSession to use for database operations
:return: List of manufacturers, or empty list if no manufacturers
"""
Retrieve all manufacturers from a MongoDB database.
logger.info("Getting all manufacturers from database")

:param session: PyMongo ClientSession to use for database operations.
:return: List of manufacturers, or empty list if no manufacturers.
"""
logger.info("Getting all manufacturers from the database")
manufacturers = self._manufacturers_collection.find(session=session)

return [ManufacturerOut(**manufacturer) for manufacturer in manufacturers]

def update(
self, manufacturer_id: str, manufacturer: ManufacturerIn, session: ClientSession = None
) -> ManufacturerOut:
"""Update manufacturer by its ID in database
:param: manufacturer_id: The id of the manufacturer to be updated
:param: manufacturer: The manufacturer with the update data
:param session: PyMongo ClientSession to use for database operations
:raises: DuplicateRecordError: if changed manufacturer name is a duplicate name
"""
Update manufacturer by its ID in a MongoDB database.
:returns: the updated manufacturer
:param manufacturer_id: The ID of the manufacturer to update.
:param manufacturer: The manufacturer containing the update data.
:param session: PyMongo ClientSession to use for database operations.
:raises: DuplicateRecordError: if a duplicate manufacturer is found.
:return: the updated manufacturer.
joelvdavies marked this conversation as resolved.
Show resolved Hide resolved
"""
manufacturer_id = CustomObjectId(manufacturer_id)

Expand All @@ -102,7 +99,7 @@ def update(
if self._is_duplicate_manufacturer(manufacturer.code, manufacturer_id, session=session):
raise DuplicateRecordError("Duplicate manufacturer found")

logger.info("Updating manufacturer with ID %s", manufacturer_id)
logger.info("Updating manufacturer with ID: %s", manufacturer_id)
self._manufacturers_collection.update_one(
{"_id": manufacturer_id}, {"$set": manufacturer.model_dump()}, session=session
)
Expand All @@ -112,21 +109,21 @@ def update(

def delete(self, manufacturer_id: str, session: ClientSession = None) -> None:
"""
Delete a manufacturer by its ID from MongoDB database.
Checks if manufacturer is a part of an item, and does not delete if it is
Delete a manufacturer by its ID from a MongoDB database.
The method Checks if the manufacturer is part of a catalogue item, and raises a `PartOfCatalogueItemError` if it
joelvdavies marked this conversation as resolved.
Show resolved Hide resolved
is.
:param manufacturer_id: The ID of the manufacturer to delete
:param session: PyMongo ClientSession to use for database operations
:raises PartOfCatalogueItemError: if manufacturer is a part of a catalogue item
:raises MissingRecordError: if supplied manufacturer ID does not exist in the database
:param manufacturer_id: The ID of the manufacturer to delete.
:param session: PyMongo ClientSession to use for database operations.
:raises PartOfCatalogueItemError: if the manufacturer is part of a catalogue item.
:raises MissingRecordError: if the manufacturer doesn't exist.
joelvdavies marked this conversation as resolved.
Show resolved Hide resolved
"""
manufacturer_id = CustomObjectId(manufacturer_id)
if self._is_manufacturer_in_catalogue_item(str(manufacturer_id), session=session):
raise PartOfCatalogueItemError(
f"The manufacturer with id {str(manufacturer_id)} is a part of a Catalogue Item"
)
raise PartOfCatalogueItemError(f"Manufacturer with ID '{str(manufacturer_id)}' is part of a catalogue item")

logger.info("Deleting manufacturer with ID %s from the database", manufacturer_id)
logger.info("Deleting manufacturer with ID: %s from the database", manufacturer_id)
result = self._manufacturers_collection.delete_one({"_id": manufacturer_id}, session=session)
if result.deleted_count == 0:
raise MissingRecordError(f"No manufacturer found with ID: {str(manufacturer_id)}")
Expand All @@ -135,12 +132,12 @@ def _is_duplicate_manufacturer(
self, code: str, manufacturer_id: CustomObjectId = None, session: ClientSession = None
) -> bool:
"""
Check if manufacturer with the same name already exists in the manufacturer collection
Check if a manufacturer with the same code already exists.
:param code: The code of the manufacturer to check for duplicates.
:param manufacturer_id: The ID of the manufacturer to check if the duplicate manufacturer found is itself.
:param session: PyMongo ClientSession to use for database operations
:return `True` if duplicate manufacturer, `False` otherwise
:param session: PyMongo ClientSession to use for database operations.
:return `True` if a duplicate manufacturer is found, `False` otherwise.
"""
logger.info("Checking if manufacturer with code '%s' already exists", code)
manufacturer = self._manufacturers_collection.find_one(
Expand All @@ -149,13 +146,14 @@ def _is_duplicate_manufacturer(
return manufacturer is not None

def _is_manufacturer_in_catalogue_item(self, manufacturer_id: str, session: ClientSession = None) -> bool:
"""Checks to see if any of the documents in the database have a specific manufacturer id
"""
Check if a manufacturer is part of a catalogue item based on its ID.
:param manufacturer_id: The ID of the manufacturer that is looked for
:param session: PyMongo ClientSession to use for database operations
:return: Returns True if 1 or more documents have the manufacturer ID, false if none do
:param manufacturer_id: The ID of the manufacturer to check.
:param session: PyMongo ClientSession to use for database operations.
:return: Returns `True` if the manufacturer is part of a catalogue item, `False` otherwise.
"""
manufacturer_id = CustomObjectId(manufacturer_id)
return (
self._catalogue_item_collection.find_one({"manufacturer_id": manufacturer_id}, session=session) is not None
self._catalogue_items_collection.find_one({"manufacturer_id": manufacturer_id}, session=session) is not None
)
42 changes: 22 additions & 20 deletions inventory_management_system_api/services/manufacturer.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,25 @@


class ManufacturerService:
"""Service for managing manufacturers"""
"""
Service for managing manufacturers.
"""

def __init__(
self,
manufacturer_repository: Annotated[ManufacturerRepo, Depends(ManufacturerRepo)],
) -> None:
"""
Initialise the manufacturer service with a ManufacturerRepo
Initialise the `ManufacturerService` with a `ManufacturerRepo` repository.

:param manufacturer_repository: The `ManufacturerRepo` repository to use.
"""

self._manufacturer_repository = manufacturer_repository

def create(self, manufacturer: ManufacturerPostSchema) -> ManufacturerOut:
"""
Create a new manufacturer.

:param manufacturer: The manufacturer to be created.
:return: The created manufacturer.
"""
Expand All @@ -52,35 +54,36 @@ def create(self, manufacturer: ManufacturerPostSchema) -> ManufacturerOut:

def get(self, manufacturer_id: str) -> Optional[ManufacturerOut]:
"""
Get manufacturer by its ID.
Retrieve a manufacturer by its ID.

:param: manufacturer_id: The ID of the requested manufacturer
:return: The retrieved manufacturer, or None if not found
:param manufacturer_id: The ID of the manufacturer to retrieve.
:return: The retrieved manufacturer, or `None` if not found.
"""
return self._manufacturer_repository.get(manufacturer_id)

def list(self) -> List[ManufacturerOut]:
"""Get all manufacturers
"""
Retrieve all manufacturers.

:return: list of all manufacturers
:return: List of manufacturers, or empty list if no manufacturers.
"""
return self._manufacturer_repository.list()

def update(self, manufacturer_id: str, manufacturer: ManufacturerPatchSchema) -> ManufacturerOut:
"""Update a manufacturer by its ID


:params: manufacturer_id: The ID of the manufacturer to be updated
:return: The updated manufacturer
:raises MissingRecordError: If manufacturer does not exist in database
"""
update_data = manufacturer.model_dump(exclude_unset=True)
Update a manufacturer by its ID.

:params: manufacturer_id: The ID of the manufacturer to be updated.
:raises MissingRecordError: If the manufacturer with the given ID does not exist.
:return: The updated manufacturer.
"""
stored_manufacturer = self.get(manufacturer_id)
if not stored_manufacturer:
raise MissingRecordError(f"No manufacturer found with ID {manufacturer_id}")
raise MissingRecordError(f"No manufacturer found with ID: {manufacturer_id}")

if "name" in update_data and update_data["name"] != stored_manufacturer.name:
update_data = manufacturer.model_dump(exclude_unset=True)

if "name" in update_data and manufacturer.name != stored_manufacturer.name:
update_data["code"] = utils.generate_code(manufacturer.name, "manufacturer")

stored_manufacturer = stored_manufacturer.model_copy(
Expand All @@ -91,9 +94,8 @@ def update(self, manufacturer_id: str, manufacturer: ManufacturerPatchSchema) ->

def delete(self, manufacturer_id: str) -> None:
"""
Delete a manufacturer by its ID

:param manufacturer_id: The ID of the manufacturer to delete
Delete a manufacturer by its ID.

:param manufacturer_id: The ID of the manufacturer to delete.
"""
return self._manufacturer_repository.delete(manufacturer_id)
39 changes: 39 additions & 0 deletions test/mock_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,45 @@
# at runtime
CREATED_MODIFIED_GET_DATA_EXPECTED = {"created_time": ANY, "modified_time": ANY}

# --------------------------------- MANUFACTURERS ---------------------------------
joelvdavies marked this conversation as resolved.
Show resolved Hide resolved

MANUFACTURER_POST_DATA_A = {
"name": "Manufacturer A",
"url": "http://testUrl.co.uk",
"address": {
"address_line": "1 Example Street",
"town": "Oxford",
"county": "Oxfordshire",
"country": "United Kingdom",
"postcode": "OX1 2AB",
},
"telephone": "0932348348",
}

MANUFACTURER_POST_DATA_B = {
"name": "Manufacturer B",
"url": "http://example.co.uk",
"address": {
"address_line": "2 Example Street",
"town": "Oxford",
"county": "Oxfordshire",
"country": "United Kingdom",
"postcode": "OX1 2AB",
},
"telephone": "073434394",
}

MANUFACTURER_IN_DATA_A = {
**MANUFACTURER_POST_DATA_A,
"code": "manufacturer-a",
}

MANUFACTURER_IN_DATA_B = {
**MANUFACTURER_POST_DATA_B,
"code": "manufacturer-b",
}

# --------------------------------- SYSTEMS ---------------------------------

SYSTEM_POST_DATA_NO_PARENT_A = {
"parent_id": None,
Expand Down
Loading
Loading