diff options
| -rw-r--r-- | gn3/db/__init__.py | 47 | ||||
| -rw-r--r-- | tests/unit/db/test_db.py | 73 |
2 files changed, 0 insertions, 120 deletions
diff --git a/gn3/db/__init__.py b/gn3/db/__init__.py index 3b063f1..5676eb5 100644 --- a/gn3/db/__init__.py +++ b/gn3/db/__init__.py @@ -63,53 +63,6 @@ def update(conn: Any, conn.commit() return cursor.rowcount - -def fetchone(conn: Any, - table: str, - where: Optional[Dataclass], - columns: Union[str, List[str]] = "*") -> Optional[Dataclass]: - """Run a SELECT on a table. Returns only one result!""" - if not any(astuple(where)): - return None - where_ = {TABLEMAP[table].get(k): v for k, v in asdict(where).items() - if v is not None and k in TABLEMAP[table]} - sql = "" - if columns != "*": - sql = f"SELECT {', '.join(columns)} FROM {table} " - else: - sql = f"SELECT * FROM {table} " - if where: - sql += "WHERE " - sql += " AND ".join(f"{k} = " - "%s" for k in where_.keys()) - with conn.cursor() as cursor: - cursor.execute(sql, tuple(where_.values())) - return DATACLASSMAP[table](*cursor.fetchone()) - - -def fetchall(conn: Any, - table: str, - where: Optional[Dataclass], - columns: Union[str, List[str]] = "*") -> Optional[Generator]: - """Run a SELECT on a table. Returns all the results as a tuple!""" - if not any(astuple(where)): - return None - where_ = {TABLEMAP[table].get(k): v for k, v in asdict(where).items() - if v is not None and k in TABLEMAP[table]} - sql = "" - if columns != "*": - sql = f"SELECT {', '.join(columns)} FROM {table} " - else: - sql = f"SELECT * FROM {table} " - if where: - sql += "WHERE " - sql += " AND ".join(f"{k} = " - "%s" for k in where_.keys()) - with conn.cursor() as cursor: - cursor.execute(sql, tuple(where_.values())) - return (DATACLASSMAP[table](*record) for record in cursor.fetchall()) - - def insert(conn: Any, table: str, data: Dataclass) -> Optional[int]: diff --git a/tests/unit/db/test_db.py b/tests/unit/db/test_db.py index 821a6b6..5e146fd 100644 --- a/tests/unit/db/test_db.py +++ b/tests/unit/db/test_db.py @@ -4,13 +4,9 @@ from unittest import mock import pytest -from gn3.db import fetchall -from gn3.db import fetchone from gn3.db import update from gn3.db import diff_from_dict -from gn3.db.probesets import Probeset from gn3.db.phenotypes import Phenotype -from gn3.db.metadata_audit import MetadataAudit class TestCrudMethods(TestCase): @@ -51,75 +47,6 @@ class TestCrudMethods(TestCase): ('Test Pre Pub', 'Test Post Pub', 'Rob', 1, 'Rob')) @pytest.mark.unit_test - def test_fetch_phenotype(self): - """Test that a single phenotype is fetched properly - - """ - db_mock = mock.MagicMock() - with db_mock.cursor() as cursor: - test_data = ( - 35, "Test pre-publication", "Test post-publication", - "Original description A", "cm^2", "pre-abbrev", - "post-abbrev", "LAB001", "R. W.", "R. W.", "R. W." - ) - cursor.fetchone.return_value = test_data - phenotype = fetchone(db_mock, - "Phenotype", - where=Phenotype(id_=35, owner="Rob")) - self.assertEqual(phenotype.id_, 35) - self.assertEqual(phenotype.pre_pub_description, - "Test pre-publication") - cursor.execute.assert_called_once_with( - "SELECT * FROM Phenotype WHERE id = %s AND Owner = %s", - (35, 'Rob')) - - @pytest.mark.unit_test - def test_fetchall_metadataaudit(self): - """Test that multiple metadata_audit entries are fetched properly - - """ - db_mock = mock.MagicMock() - with db_mock.cursor() as cursor: - test_data = ( - 1, 35, "Rob", ('{"pages": ' - '{"old": "5099-5109", ' - '"new": "5099-5110"}, ' - '"month": {"old": "July", ' - '"new": "June"}, ' - '"year": {"old": "2001", ' - '"new": "2002"}}'), - "2021-06-04 09:01:05") - cursor.fetchall.return_value = (test_data,) - metadata = list(fetchall(db_mock, - "metadata_audit", - where=MetadataAudit(dataset_id=35, - editor="Rob")))[0] - self.assertEqual(metadata.dataset_id, 35) - self.assertEqual(metadata.time_stamp, - "2021-06-04 09:01:05") - cursor.execute.assert_called_once_with( - ("SELECT * FROM metadata_audit WHERE " - "dataset_id = %s AND editor = %s"), - (35, 'Rob')) - - @pytest.mark.unit_test - # pylint: disable=R0201 - def test_probeset_called_with_right_columns(self): - """Given a columns argument, test that the correct sql query is - constructed""" - db_mock = mock.MagicMock() - with db_mock.cursor() as cursor: - cursor.fetchall.return_value = None - fetchone(db_mock, - "ProbeSet", - where=Probeset(name="1446112_at"), - columns=["OMIM", "Probe_set_target_region"]) - cursor.execute.assert_called_once_with( - "SELECT OMIM, Probe_set_target_region FROM ProbeSet WHERE " - "Name = %s", - ("1446112_at",)) - - @pytest.mark.unit_test def test_diff_from_dict(self): """Test that a correct diff is generated""" self.assertEqual(diff_from_dict({"id": 1, "data": "a"}, |