From 6cc0d697d44e0d4e69ecc075662272c48e8677eb Mon Sep 17 00:00:00 2001 From: Sushant Marella Date: Tue, 23 Apr 2024 11:56:32 -0400 Subject: [PATCH] Modify methods to filter resources based on user access --- backend/services/resource.py | 133 ++++++++++--------------- backend/test/services/resource_test.py | 107 ++++++++++++++++++++ 2 files changed, 161 insertions(+), 79 deletions(-) diff --git a/backend/services/resource.py b/backend/services/resource.py index 26bc0fa..f51e617 100644 --- a/backend/services/resource.py +++ b/backend/services/resource.py @@ -8,87 +8,83 @@ from ..models.user_model import User from .exceptions import ResourceNotFoundException - -""" - Field reference: - - id: int | None = None - name: str = Field(..., max_length=150, description="The name of the resource") - summary: str = Field(..., max_length=300, description="The summary of the resource") - link: str = Field(..., max_length=150, description="link to the resource") - programtype: ProgramTypeEnum - created_at: Optional[datetime] -""" - - class ResourceService: def __init__(self, session: Session = Depends(db_session)): self._session = session - def all(self) -> list[Resource]: + def all(self, user: User) -> list[Resource]: """ - Retrieves all Resources from the table + Retrieves all Resources from the table that the user has access to + + Parameters: + user: a valid User model representing the currently logged in User Returns: - list[Resource]: list of all `Resource` + list[Resource]: list of accessible `Resource` for the user """ - # Select all entries in the `Resource` table - query = select(ResourceEntity) + # Assuming user has 'categories' attribute listing accessible resource categories + accessible_categories = user.categories + query = select(ResourceEntity).where(ResourceEntity.category.in_(accessible_categories)) entities = self._session.scalars(query).all() - # Convert entries to a model and return return [entity.to_model() for entity in entities] - def create(self, subject: User, resource: Resource) -> Resource: + def create(self, user: User, resource: Resource) -> Resource: """ - Creates a resource based on the input object and adds it to the table. + Creates a resource based on the input object and adds it to the table if the user has the right to do so. + + Parameters: + user: a valid User model representing the currently logged in User + resource: Resource object to add to table Returns: Resource: Object added to table """ - # Check if user has admin permissions - # TODO + # Assuming we check user's right to create resources in specific categories + if resource.category not in user.categories: + raise PermissionError("User does not have permission to add resources to this category") - # Create new resource object - recource_entity = ResourceEntity.from_model(resource) - - self._session.add(recource_entity) + resource_entity = ResourceEntity.from_model(resource) + self._session.add(resource_entity) self._session.commit() - # Return added object - return recource_entity.to_model() + return resource_entity.to_model() - def get_by_id(self, id: int) -> Resource: + def get_by_id(self, user: User, id: int) -> Resource: """ - Gets a resource based on the resource id + Gets a resource based on the resource id that the user has access to + + Parameters: + user: a valid User model representing the currently logged in User + id: int, the id of the resource Returns: Resource - """ - # Query the resource table with id + Raises: + ResourceNotFoundException: If no resource is found with id + """ + accessible_categories = user.categories resource = ( self._session.query(ResourceEntity) - .filter(ResourceEntity.id == id) + .filter(ResourceEntity.id == id, ResourceEntity.category.in_(accessible_categories)) .one_or_none() ) - # Check if result is null if resource is None: raise ResourceNotFoundException(f"No resource found with id: {id}") return resource.to_model() - def update(self, subject: User, resource: ResourceEntity) -> Resource: + def update(self, user: User, resource: ResourceEntity) -> Resource: """ - Update the resource - If none found with that id, a debug description is displayed. + Update the resource if the user has access Parameters: - subject: a valid User model representing the currently logged in User - resource (Resource): Resource to add to table + user: a valid User model representing the currently logged in User + resource (ResourceEntity): Resource to update Returns: Resource: Updated resource object @@ -96,70 +92,50 @@ class ResourceService: Raises: ResourceNotFoundException: If no resource is found with the corresponding ID """ + # Check if user has permission to update the resource + if resource.category not in user.categories: + raise PermissionError("User does not have permission to update this category") - # Check if user has admin permissions - # TODO - - # Query the resource table with matching id obj = self._session.get(ResourceEntity, resource.id) if resource.id else None - # Check if result is null if obj is None: - raise ResourceNotFoundException( - f"No resource found with matching id: {resource.id}" - ) + raise ResourceNotFoundException(f"No resource found with matching id: {resource.id}") - # Update resource object - obj.name = resource.name - obj.summary = resource.summary - obj.link = resource.link - obj.programtype = resource.programtype - - # Save Changes + obj.update_from_model(resource) # Assuming an update method exists self._session.commit() - # Return updated object return obj.to_model() - def delete(self, subject: User, id: int) -> None: + def delete(self, user: User, id: int) -> None: """ - Delete resource based on id - If no resource exists, a debug description is displayed. + Delete resource based on id that the user has access to Parameters: - subject: a valid User model representing the currently logged in User - id: a string representing a unique resource id + user: a valid User model representing the currently logged in User + id: int, a unique resource id Raises: ResourceNotFoundException: If no resource is found with the corresponding id """ - - # Check if user has admin permissions - # TODO - - # Query the resource table with matching id + accessible_categories = user.categories resource = ( self._session.query(ResourceEntity) - .filter(ResourceEntity.id == id) + .filter(ResourceEntity.id == id, ResourceEntity.category.in_(accessible_categories)) .one_or_none() ) - # Check if result is null if resource is None: raise ResourceNotFoundException(f"No resource found with matching id: {id}") - # Delete object and commit self._session.delete(resource) - - # Save Changes self._session.commit() - def get_by_slug(self, search_string: str) -> list[Resource]: + def get_by_slug(self, user: User, search_string: str) -> list[Resource]: """ - Get a list of resources given a search string - If none retrieved, a debug description is displayed. + Get a list of resources given a search string that the user has access to Parameters: + user: a valid User model representing the currently logged in User search_string: a string to search resources by Returns: @@ -168,16 +144,15 @@ class ResourceService: Raises: ResourceNotFoundException if no resource is found with the corresponding slug """ - - # Query the resource with matching slug + accessible_categories = user.categories query = select(ResourceEntity).where( or_( ResourceEntity.title.ilike(f"%{search_string}%"), ResourceEntity.details.ilike(f"%{search_string}%"), - ResourceEntity.location.ilike(f"%{search_string}%"), - ) + ResourceEntity.location.ilike(f"%{search_string}%") + ), + ResourceEntity.category.in_(accessible_categories) ) entities = self._session.scalars(query).all() - # Convert entries to a model and return return [entity.to_model() for entity in entities] diff --git a/backend/test/services/resource_test.py b/backend/test/services/resource_test.py index e69de29..67457d2 100644 --- a/backend/test/services/resource_test.py +++ b/backend/test/services/resource_test.py @@ -0,0 +1,107 @@ +import pytest +from sqlalchemy.orm import Session +from sqlalchemy.exc import NoResultFound +from .resource_service import ResourceService +from .models.resource_model import Resource +from .entities.resource_entity import ResourceEntity +from .models.user_model import User +from .exceptions import ResourceNotFoundException + +# Example of a Resource and User object creation for use in tests +@pytest.fixture +def sample_resource(): + return Resource(id=1, name="Sample Resource", summary="A brief summary", link="http://example.com", programtype="TypeA") + +@pytest.fixture +def sample_user(): + return User(id=1, username="admin", is_admin=True) + +@pytest.fixture +def resource_service(mocker): + # Mock the session and its methods + mock_session = mocker.MagicMock(spec=Session) + return ResourceService(session=mock_session) + +def test_all(resource_service, mocker): + # Setup + mock_query_all = mocker.MagicMock(return_value=[ResourceEntity(id=1, name="Resource One"), ResourceEntity(id=2, name="Resource Two")]) + mocker.patch.object(resource_service._session, 'scalars', return_value=mock_query_all) + + # Execution + results = resource_service.all() + + # Verification + assert len(results) == 2 + assert results[0].id == 1 + assert results[1].name == "Resource Two" + +def test_create(resource_service, mocker, sample_resource, sample_user): + # Mock the add and commit methods of session + mocker.patch.object(resource_service._session, 'add') + mocker.patch.object(resource_service._session, 'commit') + + # Execution + result = resource_service.create(sample_user, sample_resource) + + # Verification + resource_service._session.add.assert_called_once() + resource_service._session.commit.assert_called_once() + assert result.id == sample_resource.id + assert result.name == sample_resource.name + +def test_get_by_id_found(resource_service, mocker): + # Setup + resource_entity = ResourceEntity(id=1, name="Existing Resource") + mocker.patch.object(resource_service._session, 'query', return_value=mocker.MagicMock(one_or_none=mocker.MagicMock(return_value=resource_entity))) + + # Execution + result = resource_service.get_by_id(1) + + # Verification + assert result.id == 1 + assert result.name == "Existing Resource" + +def test_get_by_id_not_found(resource_service, mocker): + # Setup + mocker.patch.object(resource_service._session, 'query', return_value=mocker.MagicMock(one_or_none=mocker.MagicMock(return_value=None))) + + # Execution & Verification + with pytest.raises(ResourceNotFoundException): + resource_service.get_by_id(999) + +def test_update(resource_service, mocker, sample_resource, sample_user): + # Setup + mocker.patch.object(resource_service._session, 'get', return_value=sample_resource) + mocker.patch.object(resource_service._session, 'commit') + + # Execution + result = resource_service.update(sample_user, sample_resource) + + # Verification + assert result.id == sample_resource.id + resource_service._session.commit.assert_called_once() + +def test_delete(resource_service, mocker): + # Setup + mock_resource = ResourceEntity(id=1, name="Delete Me") + mocker.patch.object(resource_service._session, 'query', return_value=mocker.MagicMock(one_or_none=mocker.MagicMock(return_value=mock_resource))) + mocker.patch.object(resource_service._session, 'delete') + mocker.patch.object(resource_service._session, 'commit') + + # Execution + resource_service.delete(sample_user(), 1) + + # Verification + resource_service._session.delete.assert_called_with(mock_resource) + resource_service._session.commit.assert_called_once() + +def test_get_by_slug(resource_service, mocker): + # Setup + mock_query_all = mocker.MagicMock(return_value=[ResourceEntity(id=1, name="Resource One"), ResourceEntity(id=2, name="Resource Two")]) + mocker.patch.object(resource_service._session, 'scalars', return_value=mock_query_all) + + # Execution + results = resource_service.get_by_slug("Resource") + + # Verification +