add(storehelpers and tests)

genotype_api/database/crud/read.py

@@ -12,6 +12,7 @@
     Sample,
     User,
     SNP,
+    Genotype,
 )
 
 LOG = logging.getLogger(__name__)
@@ -34,7 +35,7 @@ def get_analysis_by_type_sample(
         )
 
     def get_analysis_by_id(self, analysis_id: int) -> Analysis:
-        return self.session.query(Analysis).filter(Analysis.id == analysis_id).one()
+        return self.session.query(Analysis).filter(Analysis.id == analysis_id).first()
 
     def get_analyses(self) -> list[Analysis]:
         return self.session.query(Analysis).all()
@@ -63,10 +64,10 @@ def get_analysis_by_type_and_sample_id(self, analysis_type: str, sample_id: str)
         )
 
     def get_plate_by_id(self, plate_id: int) -> Plate:
-        return self.session.query(Plate).filter(Plate.id == plate_id).one()
+        return self.session.query(Plate).filter(Plate.id == plate_id).first()
 
     def get_plate_by_plate_id(self, plate_id: str) -> Plate:
-        return self.session.query(Plate).filter(Plate.plate_id == plate_id).one()
+        return self.session.query(Plate).filter(Plate.plate_id == plate_id).first()
 
     def get_ordered_plates(self, order_params: PlateOrderParams) -> list[Plate]:
         sort_func = desc if order_params.sort_order == "descend" else asc
@@ -78,6 +79,9 @@ def get_ordered_plates(self, order_params: PlateOrderParams) -> list[Plate]:
             .all()
         )
 
+    def get_genotype_by_id(self, entry_id: int) -> Genotype:
+        return self.session.query(Genotype).filter(Genotype.id == entry_id).first()
+
     def get_filtered_samples(self, filter_params: SampleFilterParams) -> list[Sample]:
         query = self.session.query(Sample).distinct().join(Analysis)
         if filter_params.sample_id:
@@ -127,11 +131,10 @@ def _get_samples(query: Query, sample_id: str) -> Query:
         return query.filter(Sample.id.contains(sample_id))
 
     def get_sample(self, sample_id: str) -> Sample:
-        """Get sample or raise 404."""
-        return self.session.query(Sample).filter(Sample.id == sample_id).one()
+        return self.session.query(Sample).filter(Sample.id == sample_id).first()
 
     def get_user_by_id(self, user_id: int) -> User:
-        return self.session.query(User).filter(User.id == user_id).one()
+        return self.session.query(User).filter(User.id == user_id).first()
 
     def get_user_by_email(self, email: str) -> User | None:
         return self.session.query(User).filter(User.email == email).first()

tests/conftest.py

@@ -1,29 +1,118 @@
 """General fixtures"""
 
+import datetime
 from pathlib import Path
+from typing import Generator
 
 import pytest
 
+from genotype_api.database.database import initialise_database, create_all_tables, drop_all_tables
+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
+
+
+@pytest.fixture
+def timestamp_now() -> datetime:
+    return datetime.datetime.now()
+
+
+@pytest.fixture
+def store() -> Generator[Store, None, None]:
+    """Return a CG store."""
+    initialise_database("sqlite:///")
+    _store = Store()
+    create_all_tables()
+    yield _store
+    drop_all_tables()
+
+
+@pytest.fixture
+def helpers(store: Store):
+    return StoreHelpers(store)
+
+
+@pytest.fixture
+def test_user() -> User:
+    return User(id=1, email="[email protected]", name="Test Testorus")
+
+
+@pytest.fixture
+def test_plate(test_user: User, timestamp_now: datetime) -> Plate:
+    return Plate(
+        id=1,
+        plate_id="ID_1",
+        signed_by=test_user.id,
+        method_document="mdoc",
+        method_version="mdoc_ver",
+        created_at=timestamp_now,
+        signed_at=timestamp_now,
+    )
+
+
+@pytest.fixture
+def test_snp() -> SNP:
+    return SNP(id=1, ref="A", chrom="2", pos=12341)
+
+
+@pytest.fixture
+def test_sample_id() -> str:
+    return "test_sample"
+
+
+@pytest.fixture
+def sex_male() -> str:
+    return "male"
+
+
+@pytest.fixture
+def test_sample(timestamp_now: datetime, test_sample_id: str, sex_male: str) -> Sample:
+    return Sample(
+        id=test_sample_id,
+        status="test_status",
+        comment="test_comment",
+        sex=sex_male,
+        created_at=timestamp_now,
+    )
+
+
+@pytest.fixture
+def test_genotype() -> Genotype:
+    return Genotype(id=1, rsnumber="12315", analysis_id=1, allele_1="A", allele_2="G")
+
+
+@pytest.fixture
+def test_analysis(sex_male, timestamp_now: datetime, test_sample_id: str) -> Analysis:
+    return Analysis(
+        id=1,
+        type="genotype",
+        source="source",
+        sex=sex_male,
+        created_at=timestamp_now,
+        sample_id=test_sample_id,
+        plate_id=1,
+    )
+
 
 @pytest.fixture(name="fixtures_dir")
 def fixture_fixtures_dir() -> Path:
-    """Return the path to fixtures dir"""
+    """Return the path to fixtures dir."""
     return Path("tests/fixtures")
 
 
 @pytest.fixture(name="excel_file")
 def fixture_excel_file(fixtures_dir: Path) -> Path:
-    """Return the path to an excel file"""
+    """Return the path to an excel file."""
     return fixtures_dir / "excel" / "genotype.xlsx"
 
 
 @pytest.fixture(name="include_key")
 def fixture_include_key() -> str:
-    """Return the include key"""
+    """Return the include key."""
     return "-CG-"
 
 
 @pytest.fixture(name="sample_name")
 def fixture_sample_name() -> str:
-    """Return a sample name"""
+    """Return a sample name."""
     return "sample"

tests/store_helpers.py

@@ -0,0 +1,64 @@
+"""Module that holds the store helper to be used to setup the testing environment."""
+
+from genotype_api.database.models import Sample, Analysis, User, SNP, Plate, Genotype
+from genotype_api.database.store import Store
+
+
+class StoreHelpers:
+    """Class to hold helper functions that needs to be used all over."""
+
+    def __init__(self, store: Store):
+        self.store: Store = store
+
+    def add_entity(self, entity: Sample | Analysis | SNP | Plate | User | Genotype):
+        self.store.session.add(entity)
+        self.store.session.commit()
+
+    def ensure_sample(self, sample: Sample, analyses: list[Analysis] = None):
+        """Add a sample to the store. Ensure its analyses are present."""
+        if analyses:
+            for analysis in analyses:
+                if not self.store.get_analysis_by_id(analysis.id):
+                    self.add_entity(entity=analysis)
+        self.add_entity(entity=sample)
+
+    def ensure_snp(self, snp: SNP):
+        self.add_entity(entity=snp)
+
+    def ensure_plate(self, plate: Plate, analyses: list[Analysis] = None, user: User = None):
+        """Add a plate to the store ensure the associated user and analyses are present."""
+        if user and not self.store.get_user_by_email(user.email):
+            self.add_entity(user)
+        if analyses:
+            for analysis in analyses:
+                if not self.store.get_analysis_by_id(analysis.id):
+                    self.add_entity(analysis)
+        self.add_entity(plate)
+
+    def ensure_user(self, user: User, plates: list[Plate] = None):
+        """Add a user to the store and ensure the associated plates are present."""
+        if plates:
+            for plate in plates:
+                if not self.store.get_plate_by_id(plate.id):
+                    self.add_entity(entity=plate)
+        self.add_entity(user)
+
+    def ensure_analysis(
+        self, analysis: Analysis, sample: Sample, plate: Plate, genotypes: list[Genotype]
+    ):
+        """Add an analysis to the store and ensure the associated sample, plate and genotypes are present."""
+        if sample and not self.store.get_sample(sample.id):
+            self.add_entity(sample)
+        if plate and not self.store.get_plate_by_id(plate.id):
+            self.add_entity(plate)
+        if genotypes:
+            for genotype in genotypes:
+                if not self.store.get_genotype_by_id(genotype.id):
+                    self.add_entity(genotype)
+        self.add_entity(analysis)
+
+    def ensure_genotype(self, genotype: Genotype, analysis: Analysis = None):
+        """Add a genotype to the database and ensure the associated analysis is present."""
+        if analysis and not self.store.get_analysis_by_id(analysis.id):
+            self.add_entity(analysis)
+        self.add_entity(genotype)

tests/test_store_helpers.py

@@ -0,0 +1,107 @@
+"""Module to test the store helpers."""
+
+from genotype_api.database.models import Plate, User, SNP, Analysis, Genotype, Sample
+from tests.store_helpers import StoreHelpers
+
+
+def test_ensure_user(helpers: StoreHelpers, test_user: User, test_plate: Plate):
+
+    # GIVEN a user and plates
+
+    # WHEN ensuring a user
+    helpers.ensure_user(user=test_user, plates=[test_plate])
+
+    # THEN a user and the associated plates are added
+    added_user: User = helpers.store.get_user_by_email(test_user.email)
+    assert added_user
+
+    added_plate: Plate = helpers.store.get_plate_by_id(test_plate.id)
+    assert added_plate
+    assert added_plate.signed_by == test_user.id
+
+
+def test_ensure_snp(helpers: StoreHelpers, test_snp: SNP):
+    # GIVEN a snp
+
+    # WHEN adding a snp to the store
+    helpers.ensure_snp(test_snp)
+
+    # THEN a snp is added
+    snp: SNP = helpers.store.get_snps()[0]
+    assert snp
+    assert snp.id == test_snp.id
+
+
+def test_ensure_plate(
+    helpers: StoreHelpers,
+    test_plate: Plate,
+    test_user: User,
+    test_analysis: Analysis,
+):
+    # GIVEN plates, a user and analyses
+
+    # WHEN adding a plate to the store
+    helpers.ensure_plate(plate=test_plate, analyses=[test_analysis], user=test_user)
+
+    # THEN a plate and associated user and analyses are added
+    added_user: User = helpers.store.get_user_by_email(test_user.email)
+    assert added_user
+
+    added_analysis: Analysis = helpers.store.get_analysis_by_id(test_analysis.id)
+    assert added_analysis
+
+    added_plate: Plate = helpers.store.get_plate_by_id(test_plate.id)
+    assert added_plate
+
+
+def test_ensure_genotype(helpers: StoreHelpers, test_genotype: Genotype, test_analysis: Analysis):
+    # GIVEN a genotype and an associated analysis
+
+    # WHEN adding a genotype and analysis to the store
+    helpers.ensure_genotype(genotype=test_genotype, analysis=test_analysis)
+
+    # THEN a genotype and analysis has been added
+    added_genotype: Genotype = helpers.store.get_genotype_by_id(test_genotype.id)
+    assert added_genotype
+
+    added_analysis: Analysis = helpers.store.get_analysis_by_id(test_analysis.id)
+    assert added_analysis
+
+
+def test_ensure_analysis(
+    helpers: StoreHelpers,
+    test_analysis: Analysis,
+    test_sample: Sample,
+    test_plate: Plate,
+    test_genotype: Genotype,
+):
+    # GIVEN an analysis, sample, plate and genotypes
+
+    # WHEN adding an analysis to the store
+    helpers.ensure_analysis(
+        analysis=test_analysis, sample=test_sample, plate=test_plate, genotypes=[test_genotype]
+    )
+
+    # THEN an analysis and associated sample, plate and genotypes are added
+    added_analysis: Analysis = helpers.store.get_analysis_by_id(test_analysis.id)
+    assert added_analysis
+
+    added_sample: Sample = helpers.store.get_sample(test_sample.id)
+    assert added_sample
+
+    added_plate: Plate = helpers.store.get_plate_by_id(test_plate.id)
+    assert added_plate
+
+    added_genotype: Genotype = helpers.store.get_genotype_by_id(test_genotype.id)
+    assert added_genotype
+
+
+def test_ensure_sample(test_sample: Sample, helpers: StoreHelpers):
+    # GIVEN a sample
+
+    # WHEN adding a sample to the store
+    helpers.ensure_sample(sample=test_sample)
+
+    # THEN a sample is added
+    added_sample: Sample = helpers.store.get_sample(test_sample.id)
+    assert added_sample