add(tests for store)

genotype_api/database/base_handler.py

@@ -1,8 +1,6 @@
 from dataclasses import dataclass
 from typing import Type
-
 from sqlalchemy.orm import Session, Query, DeclarativeBase
-
 from genotype_api.database.models import Analysis, Sample
 
 

genotype_api/database/crud/create.py

@@ -2,7 +2,7 @@
 
 
 from genotype_api.database.base_handler import BaseHandler
-from genotype_api.database.models import Analysis, Plate, Sample, User, SNP
+from genotype_api.database.models import Analysis, Plate, Sample, User, SNP, Genotype
 from genotype_api.dto.user import UserRequest
 from genotype_api.exceptions import SampleExistsError
 
@@ -42,14 +42,18 @@ def create_analyses_samples(self, analyses: list[Analysis]) -> list[Sample]:
             if not self.session.query(Sample).filter(Sample.id == analysis.sample_id).one_or_none()
         ]
 
-    def create_user(self, user: UserRequest) -> User:
-        db_user = User(email=user.email, name=user.name)
-        self.session.add(db_user)
+    def create_user(self, user: User) -> User:
+        self.session.add(user)
         self.session.commit()
-        self.session.refresh(db_user)
-        return db_user
+        self.session.refresh(user)
+        return user
 
     def create_snps(self, snps: list[SNP]) -> list[SNP]:
         self.session.add_all(snps)
         self.session.commit()
         return snps
+
+    def create_genotype(self, genotype: Genotype) -> Genotype:
+        self.session.add(genotype)
+        self.session.commit()
+        return genotype

genotype_api/database/crud/delete.py

@@ -26,7 +26,10 @@ def delete_user(self, user: User) -> None:
         self.session.delete(user)
         self.session.commit()
 
-    def delete_snps(self) -> any:
-        result = self.session.execute(delete(SNP))
+    def delete_snps(self) -> int:
+        snps: list[SNP] = self._get_query(SNP).all()
+        count: int = len(snps)
+        for snp in snps:
+            self.session.delete(snp)
         self.session.commit()
-        return result
+        return count

genotype_api/database/crud/read.py

@@ -23,17 +23,6 @@ class ReadHandler(BaseHandler):
     def get_analyses_from_plate(self, plate_id: int) -> list[Analysis]:
         return self.session.query(Analysis).filter(Analysis.plate_id == plate_id).all()
 
-    def get_analysis_by_type_sample(
-        self,
-        sample_id: str,
-        analysis_type: str,
-    ) -> Analysis | None:
-        return (
-            self.session.query(Analysis)
-            .filter(Analysis.sample_id == sample_id, Analysis.type == analysis_type)
-            .first()
-        )
-
     def get_analysis_by_id(self, analysis_id: int) -> Analysis:
         return self.session.query(Analysis).filter(Analysis.id == analysis_id).first()
 
@@ -130,7 +119,7 @@ def _get_samples(query: Query, sample_id: str) -> Query:
         """Returns a query for samples containing the given sample_id."""
         return query.filter(Sample.id.contains(sample_id))
 
-    def get_sample(self, sample_id: str) -> Sample:
+    def get_sample_by_id(self, sample_id: str) -> Sample:
         return self.session.query(Sample).filter(Sample.id == sample_id).first()
 
     def get_user_by_id(self, user_id: int) -> User:
@@ -145,7 +134,7 @@ def get_users_with_skip_and_limit(self, skip: int, limit: int) -> list[User]:
     def check_analyses_objects(self, analyses: list[Analysis], analysis_type: Types) -> None:
         """Raising 400 if any analysis in the list already exist in the database"""
         for analysis_obj in analyses:
-            existing_analysis = self.get_analysis_by_type_sample(
+            existing_analysis = self.get_analysis_by_type_and_sample_id(
                 sample_id=analysis_obj.sample_id,
                 analysis_type=analysis_type,
             )

genotype_api/database/crud/update.py

@@ -28,7 +28,7 @@ def refresh_sample_status(
         return sample
 
     def update_sample_comment(self, sample_id: str, comment: str) -> Sample:
-        sample: Sample = self.get_sample(sample_id=sample_id)
+        sample: Sample = self.get_sample_by_id(sample_id=sample_id)
         if not sample:
             raise SampleNotFoundError
         sample.comment = comment
@@ -38,7 +38,7 @@ def update_sample_comment(self, sample_id: str, comment: str) -> Sample:
         return sample
 
     def update_sample_status(self, sample_id: str, status: str | None) -> Sample:
-        sample: Sample = self.get_sample(sample_id=sample_id)
+        sample: Sample = self.get_sample_by_id(sample_id=sample_id)
         if not sample:
             raise SampleNotFoundError
         sample.status = status

genotype_api/services/endpoint_services/sample_service.py

@@ -60,7 +60,7 @@ def _get_sample_response(self, sample: Sample) -> SampleResponse:
         )
 
     def get_sample(self, sample_id: str) -> SampleResponse:
-        sample: Sample = self.store.get_sample(sample_id=sample_id)
+        sample: Sample = self.store.get_sample_by_id(sample_id=sample_id)
         if not sample:
             raise SampleNotFoundError
         if len(sample.analyses) == 2 and not sample.status:
@@ -82,13 +82,13 @@ def create_sample(self, sample_create: SampleCreate) -> None:
         self.store.create_sample(sample=sample)
 
     def delete_sample(self, sample_id: str) -> None:
-        sample: Sample = self.store.get_sample(sample_id=sample_id)
+        sample: Sample = self.store.get_sample_by_id(sample_id=sample_id)
         for analysis in sample.analyses:
             self.store.delete_analysis(analysis=analysis)
         self.store.delete_sample(sample=sample)
 
     def get_status_detail(self, sample_id: str) -> SampleDetail:
-        sample: Sample = self.store.get_sample(sample_id=sample_id)
+        sample: Sample = self.store.get_sample_by_id(sample_id=sample_id)
         if len(sample.analyses) != 2:
             return SampleDetail()
         return MatchGenotypeService.check_sample(sample=sample)

genotype_api/services/endpoint_services/snp_service.py

@@ -22,13 +22,13 @@ def get_snps(self, skip: int, limit: int) -> list[SNPResponse]:
 
     def upload_snps(self, snps_file: UploadFile) -> list[SNPResponse]:
         """Upload snps to the database, raises an error when SNPs already exist."""
-        existing_snps: list[SNP] = self.store.get_snps(self.session)
+        existing_snps: list[SNP] = self.store.get_snps()
         if existing_snps:
             raise SNPExistsError
         snps: list[SNP] = SNPReaderService.read_snps_from_file(snps_file)
         new_snps: list[SNP] = self.store.create_snps(snps=snps)
         return [self._get_snp_response(new_snp) for new_snp in new_snps]
 
     def delete_all_snps(self) -> int:
-        result = self.store.delete_snps(self.session)
+        result = self.store.delete_snps()
         return result.rowcount

genotype_api/services/endpoint_services/user_service.py

@@ -34,7 +34,8 @@ def create_user(self, user: UserRequest):
         existing_user: User = self.store.get_user_by_email(email=user.email)
         if existing_user:
             raise UserExistsError
-        new_user: User = self.store.create_user(user=user)
+        db_user = User(email=user.email, name=user.name)
+        new_user: User = self.store.create_user(user=db_user)
         return self._create_user_response(new_user)
 
     def get_users(self, skip: int, limit: int) -> list[UserResponse]:

tests/conftest.py

@@ -7,6 +7,8 @@
 import pytest
 
 from genotype_api.database.database import initialise_database, create_all_tables, drop_all_tables
+from genotype_api.database.filter_models.plate_models import PlateSignOff
+from genotype_api.database.filter_models.sample_models import SampleSexesUpdate
 from genotype_api.database.models import User, Plate, SNP, Sample, Genotype, Analysis
 from genotype_api.database.store import Store
 from tests.store_helpers import StoreHelpers
@@ -17,6 +19,21 @@ def timestamp_now() -> datetime:
     return datetime.datetime.now()
 
 
+@pytest.fixture
+def date_two_weeks_future() -> datetime.date:
+    return datetime.date.today() + datetime.timedelta(days=14)
+
+
+@pytest.fixture
+def date_yesterday() -> datetime:
+    return datetime.date.today() - datetime.timedelta(days=1)
+
+
+@pytest.fixture
+def date_tomorrow() -> datetime:
+    return datetime.date.today() + datetime.timedelta(days=1)
+
+
 @pytest.fixture
 def store() -> Generator[Store, None, None]:
     """Return a CG store."""
@@ -232,6 +249,39 @@ def base_store(
     return store
 
 
+@pytest.fixture
+def unsigned_plate() -> Plate:
+    return Plate(
+        id=1,
+        plate_id="ID_1",
+        signed_by=None,
+        method_document=None,
+        method_version=None,
+        created_at=datetime.datetime.now(),
+        signed_at=None,
+    )
+
+
+@pytest.fixture
+def plate_sign_off() -> PlateSignOff:
+    return PlateSignOff(
+        user_id=1,
+        signed_at=datetime.datetime.now(),
+        method_document="mdoc",
+        method_version="mdoc_ver",
+    )
+
+
+@pytest.fixture
+def sample_sex_update(test_sample_id) -> SampleSexesUpdate:
+    return SampleSexesUpdate(
+        sample_id=test_sample_id,
+        sex="female",
+        genotype_sex="female",
+        sequence_sex="female",
+    )
+
+
 @pytest.fixture(name="fixtures_dir")
 def fixture_fixtures_dir() -> Path:
     """Return the path to fixtures dir."""

tests/database/crud/test_create.py

@@ -0,0 +1,84 @@
+"""Module to test the create functionality of the genotype API CRUD."""
+
+from genotype_api.database.models import Analysis, SNP, User, Genotype, Sample, Plate
+from genotype_api.database.store import Store
+
+
+def test_create_analysis(store: Store, test_analysis: Analysis):
+    # GIVEN an analysis and an empty store
+    assert not store._get_query(Analysis).all()
+    # WHEN creating the analysis
+    store.create_analysis(analysis=test_analysis)
+
+    # THEN the analysis is created
+    assert store._get_query(Analysis).all()[0].id == test_analysis.id
+
+
+def test_create_genotype(store: Store, test_genotype: Genotype):
+    # GIVEN a genotype and an empty store
+    assert not store._get_query(Genotype).all()
+
+    # WHEN creating the genotype
+    store.create_genotype(genotype=test_genotype)
+
+    # THEN the genotype is created
+    assert store._get_query(Genotype).all()[0].id == test_genotype.id
+
+
+def test_create_snp(store: Store, test_snp: SNP):
+    # GIVEN a SNP and an empty store
+    assert not store._get_query(SNP).all()
+
+    # WHEN creating the SNP
+    store.create_snps(snps=[test_snp])
+
+    # THEN the SNP is created
+    assert store._get_query(SNP).all()[0].id == test_snp.id
+
+
+def test_create_user(store: Store, test_user: User):
+    # GIVEN a user and an empty store
+    assert not store._get_query(User).all()
+
+    # WHEN creating the user
+    store.create_user(user=test_user)
+
+    # THEN the user is created
+    assert store._get_query(User).all()[0].id == test_user.id
+
+
+def test_create_sample(store: Store, test_sample: Sample):
+    # GIVEN a sample and an empty store
+    assert not store._get_query(Sample).all()
+
+    # WHEN creating the sample
+    store.create_sample(sample=test_sample)
+
+    # THEN the sample is created
+    assert store._get_query(Sample).all()[0].id == test_sample.id
+
+
+def test_create_plate(store: Store, test_plate: Plate):
+    # GIVEN a plate and an empty store
+    assert not store._get_query(Plate).all()
+
+    # WHEN creating the plate
+    store.create_plate(plate=test_plate)
+
+    # THEN the plate is created
+    assert store._get_query(Plate).all()[0].id == test_plate.id
+
+
+def test_create_analyses_samples(store: Store, test_analysis: Analysis):
+    # GIVEN an analysis in a store
+    assert not store._get_query(Sample).all()
+    assert not store._get_query(Analysis).all()
+    store.create_analysis(test_analysis)
+
+    # WHEN creating the analyses
+    store.create_analyses_samples(analyses=[test_analysis])
+
+    # THEN the samples are created
+    sample: Sample = store._get_query(Sample).all()[0]
+    assert sample
+    assert sample.id == test_analysis.sample_id

tests/database/crud/test_delete.py

@@ -0,0 +1,59 @@
+"""Module to test the delete functionality of the genotype API CRUD."""
+
+from genotype_api.database.models import Analysis, Sample, User, Plate, SNP
+from genotype_api.database.store import Store
+
+
+def test_delete_analysis(base_store: Store, test_analysis: Analysis):
+    # GIVEN an analysis and a store with the analysis
+    assert test_analysis in base_store._get_query(Analysis).all()
+
+    # WHEN deleting the analysis
+    base_store.delete_analysis(analysis=test_analysis)
+
+    # THEN the analysis is deleted
+    assert test_analysis not in base_store._get_query(Analysis).all()
+
+
+def test_delete_sample(base_store: Store, test_sample: Sample):
+    # GIVEN a sample and a store with the sample
+    assert test_sample in base_store._get_query(Sample).all()
+
+    # WHEN deleting the sample
+    base_store.delete_sample(sample=test_sample)
+
+    # THEN the sample is deleted
+    assert test_sample not in base_store._get_query(Sample).all()
+
+
+def test_delete_plate(base_store: Store, test_plate: Plate):
+    # GIVEN a plate and a store with the plate
+    assert test_plate in base_store._get_query(Plate).all()
+
+    # WHEN deleting the plate
+    base_store.delete_plate(plate=test_plate)
+
+    # THEN the plate is deleted
+    assert test_plate not in base_store._get_query(Plate).all()
+
+
+def test_delete_user(base_store: Store, test_user: User):
+    # GIVEN a user and a store with the user
+    assert test_user in base_store._get_query(User).all()
+
+    # WHEN deleting the user
+    base_store.delete_user(user=test_user)
+
+    # THEN the user is deleted
+    assert test_user not in base_store._get_query(User).all()
+
+
+def test_delete_snps(base_store: Store, test_snp: SNP):
+    # GIVEN an SNP and a store with the SNP
+    assert base_store._get_query(SNP).all()
+
+    # WHEN deleting the SNP
+    base_store.delete_snps()
+
+    # THEN all SNPs are deleted
+    assert not base_store._get_query(SNP).all()