rushilwiz/compass
1
0
Fork 0
mirror of https://github.com/cssgunc/compass.git synced 2025-04-09 14:00:15 -04:00
Code Issues Actions Packages Projects Releases Wiki Activity

implement service and resource api and corrected services

Browse Source
This commit is contained in:
blake hardee 2024-11-04 23:24:48 -05:00
parent b3db1144ca
commit e3dbe59fb9
17 changed files with 453 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
backend/api/resource.py
View File

@ -24,3 +24,58 @@ def get_all(
subject = user_svc.get_user_by_uuid(user_id)
return resource_svc.get_resource_by_user(subject)
@api.get("/{id}", response_model=Resource, tags=["Resource"])
def get_by_id(
user_id: str,
id: int,
resource_svc: ResourceService = Depends(),
user_svc: UserService = Depends(),
):
subject = user_svc.get_user_by_uuid(user_id)
resource = resource_svc.get_resource_by_id(id)
return resource
@api.post("/", response_model=Resource, tags=["Resource"])
def create_service(
user_id: str,
resource: Resource,
resource_svc: ResourceService = Depends(),
user_svc: UserService = Depends(),
):
subject = user_svc.get_user_by_uuid(user_id)
new_resource = resource_svc.create(subject, resource)
return new_resource
@api.put("/{resource_id}", response_model=Resource, tags=["Resource"])
def update_service(
resource_id: int,
user_id: str,
resource: Resource,
resource_svc: ResourceService = Depends(),
user_svc: UserService = Depends(),
):
resource.id = resource_id
subject = user_svc.get_user_by_uuid(user_id)
updated_resource = resource_svc.update(subject, resource)
return updated_resource
@api.delete("/{resource_id}", response_model=Resource, tags=["Resource"])
def delete_service_tag_by_id(
resource_id: int,
tag_id: int,
user_id: str,
resource_svc: ResourceService = Depends(),
user_svc: UserService = Depends(),
):
subject = user_svc.get_user_by_uuid(user_id)
resource = resource_svc.get_resource_by_id(resource_id)
tag = resource_svc._tag_service.get_tag_by_id(tag_id)
resource_svc.remove_tag(subject, resource, tag)
return resource_svc.get_resource_by_id(resource_id)

61
backend/api/service.py
View File

@ -1,6 +1,6 @@
from fastapi import APIRouter, Depends
from ..services import ServiceService, UserService
from ..models.service_model import Service
from fastapi import APIRouter, Depends, HTTPException
from ..services import ServiceService, UserService, TagService
from ..models import Service, Tag
from typing import List
@ -24,3 +24,58 @@ def get_all(
subject = user_svc.get_user_by_uuid(user_id)
return service_svc.get_service_by_user(subject)
@api.get("/{id}", response_model=Service, tags=["Service"])
def get_by_id(
user_id: str,
id: int,
service_svc: ServiceService = Depends(),
user_svc: UserService = Depends(),
):
subject = user_svc.get_user_by_uuid(user_id)
service = service_svc.get_service_by_id(id)
return service
@api.post("/", response_model=Service, tags=["Service"])
def create_service(
user_id: str,
service: Service,
service_svc: ServiceService = Depends(),
user_svc: UserService = Depends(),
):
subject = user_svc.get_user_by_uuid(user_id)
new_service = service_svc.create(subject, service)
return new_service
@api.put("/{service_id}", response_model=Service, tags=["Service"])
def update_service(
service_id: int,
user_id: str,
service: Service,
service_svc: ServiceService = Depends(),
user_svc: UserService = Depends(),
):
service.id = service_id
subject = user_svc.get_user_by_uuid(user_id)
updated_service = service_svc.update(subject, service)
return updated_service
@api.delete("/{service_id}", response_model=Service, tags=["Service"])
def delete_service_tag_by_id(
service_id: int,
tag_id: int,
user_id: str,
service_svc: ServiceService = Depends(),
user_svc: UserService = Depends(),
):
subject = user_svc.get_user_by_uuid(user_id)
service = service_svc.get_service_by_id(service_id)
tag = service_svc._tag_service.get_tag_by_id(tag_id)
service_svc.remove_tag(subject, service, tag)
return service_svc.get_service_by_id(service_id)

6
backend/entities/resource_entity.py
View File

@ -31,8 +31,8 @@ class ResourceEntity(EntityBase):
link: Mapped[str] = mapped_column(String, nullable=False)
program: Mapped[Program_Enum] = mapped_column(Enum(Program_Enum), nullable=False)
# relationships
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(
back_populates="resource", cascade="all,delete"
resource_tags: Mapped[list["ResourceTagEntity"]] = relationship(
back_populates="resource", cascade="all,delete-orphan"
)
@classmethod
@ -64,5 +64,5 @@ class ResourceEntity(EntityBase):
summary=self.summary,
link=self.link,
program=self.program,
tags=[tag.tag.to_model() for tag in self.resourceTags],
tags=[tag_entity.tag.to_model() for tag_entity in self.resource_tags],
)

29
backend/entities/resource_tag_entity.py
View File

@ -1,6 +1,7 @@
""" Defines the table for resource tags """
# Import our mapped SQL types from SQLAlchemy
from importlib.resources import Resource
from sqlalchemy import ForeignKey, Integer, String, DateTime
# Import mapping capabilities from the SQLAlchemy ORM
@ -29,20 +30,20 @@ class ResourceTagEntity(EntityBase):
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
# relationships
resource: Mapped["ResourceEntity"] = relationship(back_populates="resourceTags")
tag: Mapped["TagEntity"] = relationship(back_populates="resourceTags")
resource: Mapped["ResourceEntity"] = relationship(back_populates="resource_tags")
tag: Mapped["TagEntity"] = relationship(back_populates="resource")
@classmethod
def from_model(cls, model: ResourceTag) -> Self:
return cls(
id=model.id,
resourceId=model.resource_id,
tagId=model.tag_id,
def to_model(self) -> ResourceTag:
return ResourceTag(
id=self.id,
resourceId=self.resourceId,
tagId=self.tagId,
)
# def to_model (self) -> resource_tag_model:
# return user_model(
# id = self.id,
# resourceId = self.resourceId,
# tagId = self.tagId,
# )
@classmethod
def from_model(cls, model: "ResourceTag") -> Self:
return cls(
id=model.id,
resourceId=model.resourceId,
tagId=model.tagId,
)

30
backend/entities/service_entity.py
View File

@ -20,6 +20,7 @@ from backend.models.service_model import Service
from typing import Self
from backend.models.enum_for_models import ProgramTypeEnum
class ServiceEntity(EntityBase):
# set table name
@ -32,16 +33,33 @@ class ServiceEntity(EntityBase):
status: Mapped[str] = mapped_column(String(32), nullable=False)
summary: Mapped[str] = mapped_column(String(100), nullable=False)
requirements: Mapped[list[str]] = mapped_column(ARRAY(String))
program: Mapped[ProgramTypeEnum] = mapped_column(Enum(ProgramTypeEnum), nullable=False)
program: Mapped[ProgramTypeEnum] = mapped_column(
Enum(ProgramTypeEnum), nullable=False
)
# relationships
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(
back_populates="service", cascade="all,delete"
service_tags: Mapped[list["ServiceTagEntity"]] = relationship(
back_populates="service", cascade="all,delete-orphan"
)
def to_model(self) -> Service:
return Service(id=self.id, name=self.name, status=self.status, summary=self.summary, requirements=self.requirements, program=self.program)
return Service(
id=self.id,
name=self.name,
status=self.status,
summary=self.summary,
requirements=self.requirements,
program=self.program,
tags=[tag.tag.to_model() for tag in self.service_tags],
)
@classmethod
def from_model(cls, model:Service) -> Self:
return cls(id=model.id, name=model.name, status=model.status, summary=model.summary, requirements=model.requirements, program=model.program)
def from_model(cls, model: Service) -> Self:
return cls(
id=model.id,
name=model.name,
status=model.status,
summary=model.summary,
requirements=model.requirements,
program=model.program,
)

19
backend/entities/service_tag_entity.py
View File

@ -9,6 +9,10 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
# Import the EntityBase that we are extending
from .entity_base import EntityBase
from typing import Self
from ..models.service_tag_model import ServiceTag
class ServiceTagEntity(EntityBase):
@ -21,5 +25,16 @@ class ServiceTagEntity(EntityBase):
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
# relationships
service: Mapped["ServiceEntity"] = relationship(back_populates="serviceTags")
tag: Mapped["TagEntity"] = relationship(back_populates="serviceTags")
service: Mapped["ServiceEntity"] = relationship(back_populates="service_tags")
tag: Mapped["TagEntity"] = relationship(back_populates="service")
def to_model(self) -> ServiceTag:
return ServiceTag(id=self.id, serviceId=self.serviceId, tagId=self.tagId)
@classmethod
def from_model(cls, model: "ServiceTag") -> Self:
return cls(
id=model.id,
serviceId=model.serviceId,
tagId=model.tagId,
)

13
backend/entities/tag_entity.py
View File

@ -28,11 +28,11 @@ class TagEntity(EntityBase):
content: Mapped[str] = mapped_column(String(100), nullable=False)
# relationships
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(
back_populates="tag", cascade="all,delete"
resource: Mapped[list["ResourceTagEntity"]] = relationship(
back_populates="tag", cascade="all,delete", overlaps="tag"
)
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(
back_populates="tag", cascade="all,delete"
service: Mapped[list["ServiceTagEntity"]] = relationship(
back_populates="tag", cascade="all,delete", overlaps="tag"
)
@classmethod
@ -59,7 +59,4 @@ class TagEntity(EntityBase):
User: A User model for API usage
"""
return Tag(
id=self.id,
content=self.content,
)
return Tag(id=self.id, content=self.content, created_at=self.created_at)

4
backend/models/resource_tag_model.py
View File

@ -3,5 +3,5 @@ from pydantic import BaseModel
class ResourceTag(BaseModel):
id: int | None = None
tag_id: int
resource_id: int
tagId: int
resourceId: int

2
backend/models/service_model.py
View File

@ -4,6 +4,7 @@ from typing import List
from datetime import datetime
from typing import Optional
from .enum_for_models import ProgramTypeEnum
from .tag_model import Tag
class Service(BaseModel):
@ -14,3 +15,4 @@ class Service(BaseModel):
summary: str
requirements: List[str]
program: ProgramTypeEnum
tags: List[Tag] = []

10
backend/models/service_tag_model.py
View File

@ -4,16 +4,12 @@ from typing import List
from datetime import datetime
from typing import Optional
from .enum_for_models import ProgramTypeEnum
from .enum_for_models import UserTypeEnum
from .service_model import Service
from .tag_model import Tag
from pydantic import BaseModel
from datetime import datetime
class ServiceTag(Service, BaseModel):
class ServiceTag(BaseModel):
id: int | None = None
serviceid: int | None = None
tagId: List[Tag]
serviceId: int | None = None
tagId: int

8
backend/script/reset_demo.py
View File

@ -6,7 +6,12 @@ from ..database import engine, _engine_str
from ..env import getenv
from .. import entities
from ..test.services import user_test_data, service_test_data, resource_test_data
from ..test.services import (
user_test_data,
service_test_data,
resource_test_data,
tag_test_data,
)
database = getenv("POSTGRES_DATABASE")
@ -23,5 +28,6 @@ entities.EntityBase.metadata.create_all(engine)
with Session(engine) as session:
user_test_data.insert_test_data(session)
service_test_data.insert_test_data(session)
tag_test_data.insert_test_data(session)
resource_test_data.insert_test_data(session)
session.commit()

60
backend/services/resource.py
View File

@ -1,4 +1,6 @@
from fastapi import Depends
from backend.services.tag import TagService
from ..database import db_session
from sqlalchemy.orm import Session
from sqlalchemy import select
@ -6,13 +8,18 @@ from ..models.resource_model import Resource
from ..entities.resource_entity import ResourceEntity
from ..models.user_model import User, UserTypeEnum
from .exceptions import ResourceNotFoundException
from .exceptions import ProgramNotAssignedException, ResourceNotFoundException
class ResourceService:
def __init__(self, session: Session = Depends(db_session)):
def __init__(
self,
session: Session = Depends(db_session),
tag_service: TagService = Depends(),
):
self._session = session
self._tag_service = tag_service
def get_resource_by_user(self, subject: User):
"""Resource method getting all of the resources that a user has access to based on role"""
@ -43,15 +50,22 @@ class ResourceService:
Returns:
Resource: Object added to table
"""
if resource.role != user.role or resource.group != user.group:
raise PermissionError(
"User does not have permission to add resources in this role or group."
if user.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(
f"User is not {UserTypeEnum.ADMIN}, cannot update resource"
)
resource_entity = ResourceEntity.from_model(resource)
self._session.add(resource_entity)
self._session.commit()
self._session.flush()
for tag in resource.tags:
tag_entity = self._tag_service.get_or_create_tag(tag.content)
self._tag_service.add_tag_resource(
user, tag_entity, ResourceEntity.to_model(resource_entity)
)
self._session.commit()
return resource_entity.to_model()
def get_by_id(self, user: User, id: int) -> Resource:
@ -96,22 +110,38 @@ class ResourceService:
Raises:
ResourceNotFoundException: If no resource is found with the corresponding ID
"""
if resource.role != user.role or resource.group != user.group:
raise PermissionError(
"User does not have permission to update this resource."
if user.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
)
obj = self._session.get(ResourceEntity, resource.id) if resource.id else None
resource_entity = (
self._session.query(ResourceEntity)
.filter(ResourceEntity.id == resource.id)
.one_or_none()
)
if obj is None:
self._tag_service.delete_all_tags_service(resource)
if resource_entity is None:
raise ResourceNotFoundException(
f"No resource found with matching id: {resource.id}"
"The resource you are searching for does not exist."
)
obj.update_from_model(resource) # Assuming an update method exists
self._session.commit()
resource_entity.name = resource.name
resource_entity.status = resource.status
resource_entity.summary = resource.summary
resource_entity.requirements = resource.requirements
resource_entity.program = resource.program
return obj.to_model()
for tag in resource.tags:
tag_entity = self._tag_service.get_or_create_tag(tag.content)
self._tag_service.add_tag_service(
user, tag_entity, ResourceEntity.to_model(resource_entity)
)
self._session.commit()
return resource_entity.to_model()
def delete(self, user: User, id: int) -> None:
"""

78
backend/services/service.py
View File

@ -1,5 +1,9 @@
from typing import List
from fastapi import Depends
from backend.entities.service_tag_entity import ServiceTagEntity
from backend.entities.tag_entity import TagEntity
from ..database import db_session
from sqlalchemy.orm import Session
from sqlalchemy import func, select, and_, func, or_, exists, or_
@ -12,12 +16,19 @@ from backend.services.exceptions import (
ServiceNotFoundException,
ProgramNotAssignedException,
)
from . import TagService
from ..models import Tag
class ServiceService:
def __init__(self, session: Session = Depends(db_session)):
def __init__(
self,
session: Session = Depends(db_session),
tag_service: TagService = Depends(),
):
self._session = session
self._tag_service = tag_service
def get_service_by_program(self, program: ProgramTypeEnum) -> list[Service]:
"""Service method getting services belonging to a particular program."""
@ -33,8 +44,8 @@ class ServiceService:
if entity is None:
raise ServiceNotFoundException(f"Service with id: {id} does not exist")
return entity.to_model()
service = entity.to_model()
return service
def get_service_by_name(self, name: str) -> Service:
"""Service method getting services by id."""
@ -44,7 +55,9 @@ class ServiceService:
if entity is None:
raise ServiceNotFoundException(f"Service with name: {name} does not exist")
return entity.to_model()
service = entity.to_model()
# service.tags.extend(TagService.get_tags_for_service(TagService, service))
return service
def get_service_by_user(self, subject: User):
"""Service method getting all of the services that a user has access to based on role"""
@ -73,8 +86,8 @@ class ServiceService:
query = select(ServiceEntity)
entities = self._session.scalars(query).all()
return [service.to_model() for service in entities]
services = [service.to_model() for service in entities]
return services
def create(self, subject: User, service: Service) -> Service:
"""Creates/adds a service to the table."""
@ -85,6 +98,14 @@ class ServiceService:
service_entity = ServiceEntity.from_model(service)
self._session.add(service_entity)
self._session.flush()
for tag in service.tags:
tag_entity = self._tag_service.get_or_create_tag(tag.content)
self._tag_service.add_tag_service(
subject, tag_entity, ServiceEntity.to_model(service_entity)
)
self._session.commit()
return service_entity.to_model()
@ -95,7 +116,13 @@ class ServiceService:
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
)
service_entity = self._session.get(ServiceEntity, service.id)
service_entity = (
self._session.query(ServiceEntity)
.filter(ServiceEntity.id == service.id)
.one_or_none()
)
self._tag_service.delete_all_tags_service(service)
if service_entity is None:
raise ServiceNotFoundException(
@ -108,20 +135,49 @@ class ServiceService:
service_entity.requirements = service.requirements
service_entity.program = service.program
self._session.commit()
for tag in service.tags:
tag_entity = self._tag_service.get_or_create_tag(tag.content)
self._tag_service.add_tag_service(
subject, tag_entity, ServiceEntity.to_model(service_entity)
)
self._session.commit()
return service_entity.to_model()
def delete(self, subject: User, service: Service) -> None:
"""Deletes a service from the table."""
if subject.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(f"User is not {UserTypeEnum.ADMIN}")
service_entity = self._session.get(ServiceEntity, service.id)
service_entity = (
self._session.query(ServiceEntity)
.filter(ServiceEntity.id == service.id)
.one_or_none()
)
if service_entity is None:
raise ServiceNotFoundException(
"The service you are searching for does not exist."
)
self._tag_service.delete_all_tags_service(service_entity.to_model())
self._session.delete(service_entity)
self._session.commit()
def add_tag(self, subject: User, service: Service, tag: Tag):
service = self.get_service_by_id(service.id)
tag = self._tag_service.get_tag_by_id(tag.id)
self._tag_service.add_tag_service(subject, service.id, tag.id)
def remove_tag(self, subject: User, service: Service, tag: Tag) -> None:
service_tag = (
self._session.query(ServiceTagEntity)
.filter(
ServiceTagEntity.serviceId == service.id,
ServiceTagEntity.tagId == tag.id,
)
.one_or_none()
)
if service_tag is None:
raise Exception(
f"No tag with id {tag.id} found for service with id {service.id}."
)
self._session.delete(service_tag)
self._session.commit()

127
backend/services/tag.py
View File

@ -1,10 +1,13 @@
from fastapi import Depends
from backend.models.enum_for_models import UserTypeEnum
from ..database import db_session
from sqlalchemy.orm import Session
from sqlalchemy import select
from ..entities import TagEntity, ResourceTagEntity, ServiceTagEntity
from ..models import User, Resource, Service, Tag, ResourceTag
from .exceptions import ResourceNotFoundException
from ..models import User, Resource, Service, Tag, ResourceTag, ServiceTag
from .exceptions import ProgramNotAssignedException, ResourceNotFoundException
from datetime import datetime
class TagService:
@ -24,9 +27,9 @@ class TagService:
"""Returns tag based on it's id."""
tag = self._session.query(TagEntity).filter(TagEntity.id == id).one_or_none()
return tag.to_model()
return tag
def create(self, user: User, tag: Tag) -> Tag:
def create(self, tag: Tag) -> Tag:
"""Creates a tag in the database with id and content."""
# Add user permission check here if needed
@ -37,7 +40,7 @@ class TagService:
return tag_entity.to_model()
def delete(self, user: User, id: int) -> None:
def delete(self, id: int) -> None:
"""Method to delete a tag from the database, along with all connections."""
tag = self._session.query(TagEntity).filter(TagEntity.id == id).one_or_none()
@ -67,9 +70,9 @@ class TagService:
self._session.commit()
def get_tags_for_resource(self, user: User, resource: Resource) -> list[Tag]:
def get_tags_for_resource(self, resource: Resource) -> list[Tag]:
"""Get tags based on a resource."""
tags: list[Tag]
tags: list[Tag] = []
resource_tags = (
self._session.query(ResourceTagEntity)
.filter(ResourceTagEntity.tagId == resource.id)
@ -86,12 +89,12 @@ class TagService:
return tags
def get_tags_for_service(self, user: User, service: Service) -> list[Tag]:
def get_tags_for_service(self, service: Service) -> list[Tag]:
"""Get tags based on a resource."""
tags: list[Tag]
tags: list[Tag] = []
service_tags = (
self._session.query(ServiceTagEntity)
.filter(ServiceTagEntity.tagId == service.id)
.filter(ServiceTagEntity.serviceId == service.id)
.all()
)
@ -107,8 +110,106 @@ class TagService:
def add_tag_resource(self, user: User, tag: Tag, resource: Resource) -> None:
"""Adds a tag to a resource"""
resource_tag_entity = ResourceTagEntity.from_model(
ResourceTag(tag_id=tag.id, resource_id=resource.id)
if user.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(
f"User is not {UserTypeEnum.ADMIN}, cannot update resource tags"
)
existing_tag = (
self._session.query(ResourceTagEntity)
.filter(
ResourceTagEntity.tagId == tag.id,
ResourceTagEntity.resourceId == resource.id,
)
.first()
)
self._session.add(resource_tag_entity)
if existing_tag:
raise Exception(
f"Tag with id {tag.id} already exists for resource with id {resource.id}."
)
resource_tag_entity = ResourceTagEntity.from_model(
ResourceTag(tagId=tag.id, resourceId=resource.id)
)
try:
self._session.add(resource_tag_entity)
self._session.commit()
except Exception as e:
self._session.rollback() # Rollback in case of error
raise Exception("Failed to add tag to resource") from e
def add_tag_service(self, user: User, tag: Tag, service: Service) -> None:
"""Adds a tag to a service"""
existing_tag = (
self._session.query(ServiceTagEntity)
.filter(
ServiceTagEntity.tagId == tag.id,
ServiceTagEntity.serviceId == service.id,
)
.first()
)
if existing_tag:
raise Exception(
f"Tag with id {tag.id} already exists for service with id {service.id}."
)
service_tag_entity = ServiceTagEntity.from_model(
ServiceTag(tagId=tag.id, serviceId=service.id)
)
try:
self._session.add(service_tag_entity)
self._session.commit()
except Exception as e:
self._session.rollback() # Rollback in case of error
raise Exception("Failed to add tag to service") from e
def delete_all_tags_service(self, service: Service) -> None:
"""Deletes all service tags for a service"""
service_tags = self._session.query(ServiceTagEntity).filter(
ServiceTagEntity.serviceId == service.id
)
if service_tags.count() == 0:
raise ResourceNotFoundException
service_tags.delete(synchronize_session=False)
self._session.commit()
def delete_all_tags_resource(self, resource: Resource) -> None:
"""Deletes all resource tags for a resource"""
resource_tags = self._session.query(ResourceTagEntity).filter(
ResourceTagEntity.resourceId == resource.id
)
if resource_tags.count() == 0:
raise ResourceNotFoundException
resource_tags.delete(synchronize_session=False)
self._session.commit()
def get_or_create_tag(self, content: str) -> Tag:
existing_tag = (
self._session.query(TagEntity)
.filter(TagEntity.content == content)
.one_or_none()
)
if existing_tag:
return existing_tag.to_model()
try:
tag = Tag(content=content)
tag_entity = TagEntity.from_model(tag)
self._session.add(tag_entity)
self._session.commit()
return tag_entity.to_model()
except Exception as e:
self._session.rollback()
raise Exception(f"Failed to create tag with content: {tag.content}") from e

4
backend/test/services/service_test_data.py
View File

@ -3,7 +3,8 @@ from sqlalchemy.orm import Session
from ...entities import ServiceEntity
from ...models.enum_for_models import ProgramTypeEnum
from ...models.service_model import Service
from ...models import Service, Tag
from .tag_test_data import tags
service1 = Service(
id=1,
@ -95,6 +96,7 @@ service_1 = Service(
summary="24/7 support for individuals in crisis",
requirements=["Anonymous", "Confidential"],
program=ProgramTypeEnum.DOMESTIC,
tags=[tags[0], tags[1]],
)
service_2 = Service(

16
backend/test/services/tag_test.py
View File

@ -2,8 +2,8 @@
# PyTest
import pytest
from ...services import TagService, ResourceService
from .fixtures import tag_svc, resource_svc
from ...services import TagService, ResourceService, ServiceService
from .fixtures import tag_svc, resource_svc, service_svc
from .tag_test_data import tag_to_create, tag_to_create_no_id, tags
from .user_test_data import admin
@ -42,3 +42,15 @@ def test_resource_tag_creation(tag_svc: TagService, resource_svc: ResourceServic
assert len(resource.tags) == 0
assert len(updated_resource.tags) == 1
assert resource.id == updated_resource.id
def test_service_tag_creation(tag_svc: TagService, service_svc: ServiceService):
"""Test creation and attachment of service tag"""
service = service_svc.get_service_by_user(admin)[0]
tag_svc.add_tag_service(admin, tags[0], service)
updated_service = service_svc.get_service_by_id(service.id)
assert len(service.tags) == 0
assert len(updated_service.tags) == 1
assert service.id == updated_service.id

19
backend/test/services/tag_test_data.py
View File

@ -63,6 +63,25 @@ def insert_fake_data(session: Session):
session.commit()
def insert_test_data(session: Session):
"""Inserts test organization data into the test session."""
global tags
# Create entities for test organization data
entities = []
for tag in tags:
entity = TagEntity.from_model(tag)
session.add(entity)
entities.append(entity)
# Reset table IDs to prevent ID conflicts
reset_table_id_seq(session, TagEntity, TagEntity.id, len(tags) + 1)
# Commit all changes
session.commit()
@pytest.fixture(autouse=True)
def fake_data_fixture(session: Session):
"""Insert fake data the session automatically when test is run.