mirror of
https://github.com/cssgunc/compass.git
synced 2025-04-08 13:30:16 -04:00
Modify methods to filter resources based on user access
This commit is contained in:
Sushant Marella
pmoharana-cmd
parent
e8068ff2ea
commit
6cc0d697d4
backend
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user