diff --git a/pepdbagent/db_utils.py b/pepdbagent/db_utils.py
index c9f93b3..f59be18 100644
--- a/pepdbagent/db_utils.py
+++ b/pepdbagent/db_utils.py
@@ -324,3 +324,15 @@ def check_db_connection(self):
self.session_execute(select(Projects).limit(1))
except ProgrammingError:
raise SchemaError()
+
+ def delete_schema(self, engine=None) -> None:
+ """
+ Delete sql schema in the database.
+
+ :param engine: sqlalchemy engine [Default: None]
+ :return: None
+ """
+ if not engine:
+ engine = self._engine
+ Base.metadata.drop_all(engine)
+ return None
diff --git a/pepdbagent/pepdbagent.py b/pepdbagent/pepdbagent.py
index fc1a386..1519c33 100644
--- a/pepdbagent/pepdbagent.py
+++ b/pepdbagent/pepdbagent.py
@@ -45,47 +45,48 @@ def __init__(
)
sa_engine = pep_db_engine.engine
- self.__sa_engine = sa_engine
+ self.pep_db_engine = pep_db_engine
+ self._sa_engine = sa_engine
- self.__project = PEPDatabaseProject(pep_db_engine)
- self.__annotation = PEPDatabaseAnnotation(pep_db_engine)
- self.__namespace = PEPDatabaseNamespace(pep_db_engine)
- self.__sample = PEPDatabaseSample(pep_db_engine)
- self.__user = PEPDatabaseUser(pep_db_engine)
- self.__view = PEPDatabaseView(pep_db_engine)
+ self._project = PEPDatabaseProject(pep_db_engine)
+ self._annotation = PEPDatabaseAnnotation(pep_db_engine)
+ self._namespace = PEPDatabaseNamespace(pep_db_engine)
+ self._sample = PEPDatabaseSample(pep_db_engine)
+ self._user = PEPDatabaseUser(pep_db_engine)
+ self._view = PEPDatabaseView(pep_db_engine)
- self.__db_name = database
+ self._db_name = database
@property
def project(self) -> PEPDatabaseProject:
- return self.__project
+ return self._project
@property
def annotation(self) -> PEPDatabaseAnnotation:
- return self.__annotation
+ return self._annotation
@property
def namespace(self) -> PEPDatabaseNamespace:
- return self.__namespace
+ return self._namespace
@property
def user(self) -> PEPDatabaseUser:
- return self.__user
+ return self._user
@property
def sample(self) -> PEPDatabaseSample:
- return self.__sample
+ return self._sample
@property
def view(self) -> PEPDatabaseView:
- return self.__view
+ return self._view
def __str__(self):
return f"Connection to the database: '{self.__db_name}' is set!"
def __exit__(self):
- self.__sa_engine.__exit__()
+ self._sa_engine.__exit__()
@property
def connection(self):
- return self.__sa_engine
+ return self._sa_engine
diff --git a/tests/conftest.py b/tests/conftest.py
index 19ec436..d1a1c6b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,83 +1,83 @@
-import os
-
-import peppy
-import pytest
-from sqlalchemy import create_engine, text
-
-from pepdbagent import PEPDatabaseAgent
-
-DNS = "postgresql+psycopg://postgres:docker@localhost:5432/pep-db"
-
-
-DATA_PATH = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "tests",
- "data",
-)
-
-
-def get_path_to_example_file(namespace, project_name):
- return os.path.join(DATA_PATH, namespace, project_name, "project_config.yaml")
-
-
-@pytest.fixture
-def list_of_available_peps():
- pep_namespaces = os.listdir(DATA_PATH)
- projects = {}
- for np in pep_namespaces:
- pep_name = os.listdir(os.path.join(DATA_PATH, np))
- projects[np] = {p: get_path_to_example_file(np, p) for p in pep_name}
- return projects
-
-
-@pytest.fixture(scope="function")
-def initiate_pepdb_con(
- list_of_available_peps,
-):
- sa_engine = create_engine(DNS)
- with sa_engine.begin() as conn:
- conn.execute(text("DROP table IF EXISTS projects CASCADE"))
- conn.execute(text("DROP table IF EXISTS samples CASCADE"))
- conn.execute(text("DROP table IF EXISTS subsamples CASCADE"))
- conn.execute(text("DROP table IF EXISTS stars CASCADE"))
- conn.execute(text("DROP table IF EXISTS users CASCADE"))
- conn.execute(text("DROP table IF EXISTS views CASCADE"))
- conn.execute(text("DROP table IF EXISTS views_samples CASCADE"))
-
- pepdb_con = PEPDatabaseAgent(dsn=DNS, echo=True)
- for namespace, item in list_of_available_peps.items():
- if namespace == "private_test":
- private = True
- else:
- private = False
- for name, path in item.items():
- prj = peppy.Project(path)
- pepdb_con.project.create(
- namespace=namespace,
- name=name,
- tag="default",
- is_private=private,
- project=prj,
- overwrite=True,
- pep_schema="random_schema_name",
- )
-
- yield pepdb_con
-
-
-@pytest.fixture(scope="function")
-def initiate_empty_pepdb_con(
- list_of_available_peps,
-):
- """
- create connection without adding peps to the db
- """
- # sa_engine = create_engine(DNS)
- # with sa_engine.begin() as conn:
- # conn.execute(text("DROP table IF EXISTS projects CASCADE"))
- # conn.execute(text("DROP table IF EXISTS samples CASCADE"))
- # conn.execute(text("DROP table IF EXISTS subsamples CASCADE"))
-
- pepdb_con = PEPDatabaseAgent(dsn=DNS, echo=False)
-
- yield pepdb_con
+# import os
+#
+# import peppy
+# import pytest
+# from sqlalchemy import create_engine, text
+#
+# from pepdbagent import PEPDatabaseAgent
+#
+# DNS = "postgresql+psycopg://postgres:docker@localhost:5432/pep-db"
+#
+#
+# DATA_PATH = os.path.join(
+# os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+# "tests",
+# "data",
+# )
+#
+#
+# def get_path_to_example_file(namespace, project_name):
+# return os.path.join(DATA_PATH, namespace, project_name, "project_config.yaml")
+#
+#
+# @pytest.fixture
+# def list_of_available_peps():
+# pep_namespaces = os.listdir(DATA_PATH)
+# projects = {}
+# for np in pep_namespaces:
+# pep_name = os.listdir(os.path.join(DATA_PATH, np))
+# projects[np] = {p: get_path_to_example_file(np, p) for p in pep_name}
+# return projects
+#
+#
+# @pytest.fixture(scope="function")
+# def initiate_pepdb_con(
+# list_of_available_peps,
+# ):
+# sa_engine = create_engine(DNS)
+# with sa_engine.begin() as conn:
+# conn.execute(text("DROP table IF EXISTS projects CASCADE"))
+# conn.execute(text("DROP table IF EXISTS samples CASCADE"))
+# conn.execute(text("DROP table IF EXISTS subsamples CASCADE"))
+# conn.execute(text("DROP table IF EXISTS stars CASCADE"))
+# conn.execute(text("DROP table IF EXISTS users CASCADE"))
+# conn.execute(text("DROP table IF EXISTS views CASCADE"))
+# conn.execute(text("DROP table IF EXISTS views_samples CASCADE"))
+#
+# pepdb_con = PEPDatabaseAgent(dsn=DNS, echo=True)
+# for namespace, item in list_of_available_peps.items():
+# if namespace == "private_test":
+# private = True
+# else:
+# private = False
+# for name, path in item.items():
+# prj = peppy.Project(path)
+# pepdb_con.project.create(
+# namespace=namespace,
+# name=name,
+# tag="default",
+# is_private=private,
+# project=prj,
+# overwrite=True,
+# pep_schema="random_schema_name",
+# )
+#
+# yield pepdb_con
+#
+#
+# @pytest.fixture(scope="function")
+# def initiate_empty_pepdb_con(
+# list_of_available_peps,
+# ):
+# """
+# create connection without adding peps to the db
+# """
+# # sa_engine = create_engine(DNS)
+# # with sa_engine.begin() as conn:
+# # conn.execute(text("DROP table IF EXISTS projects CASCADE"))
+# # conn.execute(text("DROP table IF EXISTS samples CASCADE"))
+# # conn.execute(text("DROP table IF EXISTS subsamples CASCADE"))
+#
+# pepdb_con = PEPDatabaseAgent(dsn=DNS, echo=False)
+#
+# yield pepdb_con
diff --git a/tests/test_annotation.py b/tests/test_annotation.py
index f70addb..583f4eb 100644
--- a/tests/test_annotation.py
+++ b/tests/test_annotation.py
@@ -1,38 +1,12 @@
import datetime
-import os
-import warnings
-
import pytest
-from sqlalchemy.exc import OperationalError
-
-import pepdbagent
from pepdbagent.exceptions import FilterError, ProjectNotFoundError
-from .conftest import DNS
-
-DATA_PATH = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "tests",
- "data",
-)
-
-
-def db_setup():
- # Check if the database is setup
- try:
- pepdbagent.PEPDatabaseAgent(dsn=DNS)
- except OperationalError:
- warnings.warn(
- UserWarning(
- f"Skipping tests, because DB is not setup. {DNS}. To setup DB go to README.md"
- )
- )
- return False
- return True
+from .utils import PEPDBAgentContextManager
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestAnnotation:
@@ -50,13 +24,14 @@ class TestAnnotation:
["namespace3", "subtable1"],
],
)
- def test_annotation_of_one_project(self, initiate_pepdb_con, namespace, name):
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace,
- name=name,
- tag="default",
- )
- assert result.results[0].namespace == namespace
+ def test_annotation_of_one_project(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ )
+ assert result.results[0].namespace == namespace
@pytest.mark.parametrize(
"namespace, name",
@@ -64,13 +39,14 @@ def test_annotation_of_one_project(self, initiate_pepdb_con, namespace, name):
["namespace6", "amendments1"],
],
)
- def test_annotation_of_one_non_existing_project(self, initiate_pepdb_con, namespace, name):
- with pytest.raises(ProjectNotFoundError):
- initiate_pepdb_con.annotation.get(
- namespace=namespace,
- name=name,
- tag="default",
- )
+ def test_annotation_of_one_non_existing_project(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ with pytest.raises(ProjectNotFoundError):
+ agent.annotation.get(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ )
@pytest.mark.parametrize(
"namespace, n_projects",
@@ -82,12 +58,13 @@ def test_annotation_of_one_non_existing_project(self, initiate_pepdb_con, namesp
["private_test", 0],
],
)
- def test_annotation_all(self, initiate_pepdb_con, namespace, n_projects):
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace,
- )
- assert result.count == n_projects
- assert len(result.results) == n_projects
+ def test_annotation_all(self, namespace, n_projects):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(
+ namespace=namespace,
+ )
+ assert result.count == n_projects
+ assert len(result.results) == n_projects
@pytest.mark.parametrize(
"namespace, n_projects",
@@ -100,10 +77,11 @@ def test_annotation_all(self, initiate_pepdb_con, namespace, n_projects):
],
)
@pytest.mark.parametrize("admin", ("private_test", ["private_test", "bbb"]))
- def test_annotation_all_private(self, initiate_pepdb_con, namespace, n_projects, admin):
- result = initiate_pepdb_con.annotation.get(namespace=namespace, admin=admin)
- assert result.count == n_projects
- assert len(result.results) == n_projects
+ def test_annotation_all_private(self, namespace, n_projects, admin):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(namespace=namespace, admin=admin)
+ assert result.count == n_projects
+ assert len(result.results) == n_projects
@pytest.mark.parametrize(
"namespace, limit, n_projects",
@@ -116,10 +94,11 @@ def test_annotation_all_private(self, initiate_pepdb_con, namespace, n_projects,
],
)
@pytest.mark.parametrize("admin", ("private_test", ["private_test", "bbb"]))
- def test_annotation_limit(self, initiate_pepdb_con, namespace, limit, admin, n_projects):
- result = initiate_pepdb_con.annotation.get(namespace=namespace, admin=admin, limit=limit)
- assert result.count == n_projects
- assert len(result.results) == limit
+ def test_annotation_limit(self, namespace, limit, admin, n_projects):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(namespace=namespace, admin=admin, limit=limit)
+ assert result.count == n_projects
+ assert len(result.results) == limit
@pytest.mark.parametrize(
"namespace, order_by, first_name",
@@ -131,11 +110,12 @@ def test_annotation_limit(self, initiate_pepdb_con, namespace, limit, admin, n_p
],
)
@pytest.mark.parametrize("admin", ["private_test"])
- def test_order_by(self, initiate_pepdb_con, namespace, admin, order_by, first_name):
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace, admin=admin, order_by=order_by
- )
- assert result.results[0].name == first_name
+ def test_order_by(self, namespace, admin, order_by, first_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(
+ namespace=namespace, admin=admin, order_by=order_by
+ )
+ assert result.results[0].name == first_name
@pytest.mark.parametrize(
"namespace, order_by, last_name",
@@ -147,14 +127,15 @@ def test_order_by(self, initiate_pepdb_con, namespace, admin, order_by, first_na
],
)
@pytest.mark.parametrize("admin", ["private_test"])
- def test_order_by_desc(self, initiate_pepdb_con, namespace, admin, order_by, last_name):
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace,
- admin=admin,
- order_by=order_by,
- order_desc=True,
- )
- assert result.results[0].name == last_name
+ def test_order_by_desc(self, namespace, admin, order_by, last_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(
+ namespace=namespace,
+ admin=admin,
+ order_by=order_by,
+ order_desc=True,
+ )
+ assert result.results[0].name == last_name
@pytest.mark.parametrize(
"namespace, query, found_number",
@@ -166,9 +147,10 @@ def test_order_by_desc(self, initiate_pepdb_con, namespace, admin, order_by, las
[None, "re", 2],
],
)
- def test_name_search(self, initiate_pepdb_con, namespace, query, found_number):
- result = initiate_pepdb_con.annotation.get(namespace=namespace, query=query)
- assert len(result.results) == found_number
+ def test_name_search(self, namespace, query, found_number):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(namespace=namespace, query=query)
+ assert len(result.results) == found_number
@pytest.mark.parametrize(
"namespace, query, found_number",
@@ -180,11 +162,12 @@ def test_name_search(self, initiate_pepdb_con, namespace, query, found_number):
[None, "re", 3],
],
)
- def test_name_search_private(self, initiate_pepdb_con, namespace, query, found_number):
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace, query=query, admin="private_test"
- )
- assert len(result.results) == found_number
+ def test_name_search_private(self, namespace, query, found_number):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(
+ namespace=namespace, query=query, admin="private_test"
+ )
+ assert len(result.results) == found_number
@pytest.mark.parametrize(
"namespace, name",
@@ -196,27 +179,28 @@ def test_name_search_private(self, initiate_pepdb_con, namespace, query, found_n
["namespace3", "subtable1"],
],
)
- def test_all_annotations_are_returned(self, initiate_pepdb_con, namespace, name):
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace,
- name=name,
- tag="default",
- )
- assert result.results[0].model_fields_set == {
- "is_private",
- "tag",
- "namespace",
- "digest",
- "description",
- "number_of_samples",
- "name",
- "last_update_date",
- "submission_date",
- "pep_schema",
- "pop",
- "stars_number",
- "forked_from",
- }
+ def test_all_annotations_are_returned(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ )
+ assert result.results[0].model_fields_set == {
+ "is_private",
+ "tag",
+ "namespace",
+ "digest",
+ "description",
+ "number_of_samples",
+ "name",
+ "last_update_date",
+ "submission_date",
+ "pep_schema",
+ "pop",
+ "stars_number",
+ "forked_from",
+ }
@pytest.mark.parametrize(
"namespace, query, found_number",
@@ -225,18 +209,19 @@ def test_all_annotations_are_returned(self, initiate_pepdb_con, namespace, name)
[None, "re", 3],
],
)
- def test_search_filter_success(self, initiate_pepdb_con, namespace, query, found_number):
- date_now = datetime.datetime.now() + datetime.timedelta(days=1)
- date_old = datetime.datetime.now() - datetime.timedelta(days=5)
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace,
- query=query,
- admin="private_test",
- filter_by="submission_date",
- filter_start_date=date_old.strftime("%Y/%m/%d"),
- filter_end_date=date_now.strftime("%Y/%m/%d"),
- )
- assert len(result.results) == found_number
+ def test_search_filter_success(self, namespace, query, found_number):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ date_now = datetime.datetime.now() + datetime.timedelta(days=1)
+ date_old = datetime.datetime.now() - datetime.timedelta(days=5)
+ result = agent.annotation.get(
+ namespace=namespace,
+ query=query,
+ admin="private_test",
+ filter_by="submission_date",
+ filter_start_date=date_old.strftime("%Y/%m/%d"),
+ filter_end_date=date_now.strftime("%Y/%m/%d"),
+ )
+ assert len(result.results) == found_number
@pytest.mark.parametrize(
"namespace, query, found_number",
@@ -245,18 +230,19 @@ def test_search_filter_success(self, initiate_pepdb_con, namespace, query, found
[None, "re", 0],
],
)
- def test_search_filter_zero_prj(self, initiate_pepdb_con, namespace, query, found_number):
- date_now = datetime.datetime.now() - datetime.timedelta(days=2)
- date_old = date_now - datetime.timedelta(days=2)
- result = initiate_pepdb_con.annotation.get(
- namespace=namespace,
- query=query,
- admin="private_test",
- filter_by="submission_date",
- filter_start_date=date_old.strftime("%Y/%m/%d"),
- filter_end_date=date_now.strftime("%Y/%m/%d"),
- )
- assert len(result.results) == found_number
+ def test_search_filter_zero_prj(self, namespace, query, found_number):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ date_now = datetime.datetime.now() - datetime.timedelta(days=2)
+ date_old = date_now - datetime.timedelta(days=2)
+ result = agent.annotation.get(
+ namespace=namespace,
+ query=query,
+ admin="private_test",
+ filter_by="submission_date",
+ filter_start_date=date_old.strftime("%Y/%m/%d"),
+ filter_end_date=date_now.strftime("%Y/%m/%d"),
+ )
+ assert len(result.results) == found_number
@pytest.mark.parametrize(
"namespace, query, found_number",
@@ -265,19 +251,20 @@ def test_search_filter_zero_prj(self, initiate_pepdb_con, namespace, query, foun
],
)
def test_search_incorrect_filter_by_string(
- self, initiate_pepdb_con, namespace, query, found_number
+ self, namespace, query, found_number
):
- date_now = datetime.datetime.now() - datetime.timedelta(days=2)
- date_old = date_now - datetime.timedelta(days=2)
- with pytest.raises(FilterError):
- initiate_pepdb_con.annotation.get(
- namespace=namespace,
- query=query,
- admin="private_test",
- filter_by="incorrect",
- filter_start_date=date_old.strftime("%Y/%m/%d"),
- filter_end_date=date_now.strftime("%Y/%m/%d"),
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ date_now = datetime.datetime.now() - datetime.timedelta(days=2)
+ date_old = date_now - datetime.timedelta(days=2)
+ with pytest.raises(FilterError):
+ agent.annotation.get(
+ namespace=namespace,
+ query=query,
+ admin="private_test",
+ filter_by="incorrect",
+ filter_start_date=date_old.strftime("%Y/%m/%d"),
+ filter_end_date=date_now.strftime("%Y/%m/%d"),
+ )
@pytest.mark.parametrize(
"rp_list, admin, found_number",
@@ -304,13 +291,16 @@ def test_search_incorrect_filter_by_string(
],
],
)
- def test_get_annotation_by_rp_list(self, initiate_pepdb_con, rp_list, admin, found_number):
- result = initiate_pepdb_con.annotation.get_by_rp_list(rp_list)
- assert len(result.results) == found_number
+ def test_get_annotation_by_rp_list(self, rp_list, admin, found_number):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+
+ result = agent.annotation.get_by_rp_list(rp_list)
+ assert len(result.results) == found_number
- def test_get_annotation_by_rp_enpty_list(self, initiate_pepdb_con):
- result = initiate_pepdb_con.annotation.get_by_rp_list([])
- assert len(result.results) == 0
+ def test_get_annotation_by_rp_enpty_list(self):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get_by_rp_list([])
+ assert len(result.results) == 0
@pytest.mark.parametrize(
"namespace, query, found_number",
@@ -319,10 +309,12 @@ def test_get_annotation_by_rp_enpty_list(self, initiate_pepdb_con):
],
)
def test_search_incorrect_incorrect_pep_type(
- self, initiate_pepdb_con, namespace, query, found_number
+ self, namespace, query, found_number
):
- with pytest.raises(ValueError):
- initiate_pepdb_con.annotation.get(namespace=namespace, pep_type="incorrect")
+ with PEPDBAgentContextManager(add_data=True) as agent:
+
+ with pytest.raises(ValueError):
+ agent.annotation.get(namespace=namespace, pep_type="incorrect")
@pytest.mark.parametrize(
"namespace, query, found_number",
@@ -331,10 +323,11 @@ def test_search_incorrect_incorrect_pep_type(
],
)
def test_project_list_without_annotation(
- self, initiate_pepdb_con, namespace, query, found_number
+ self, namespace, query, found_number
):
- result = initiate_pepdb_con.annotation.get_projects_list(
- namespace=namespace,
- search_str=query,
- )
- assert len(result) == found_number
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get_projects_list(
+ namespace=namespace,
+ search_str=query,
+ )
+ assert len(result) == found_number
diff --git a/tests/test_namespace.py b/tests/test_namespace.py
index aa49889..ebb4cc4 100644
--- a/tests/test_namespace.py
+++ b/tests/test_namespace.py
@@ -1,37 +1,11 @@
-import os
-import warnings
-
import pytest
-from sqlalchemy.exc import OperationalError
-
-import pepdbagent
from pepdbagent.exceptions import ProjectAlreadyInFavorites, ProjectNotInFavorites
-from .conftest import DNS
-
-DATA_PATH = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "tests",
- "data",
-)
-
-
-def db_setup():
- # Check if the database is setup
- try:
- pepdbagent.PEPDatabaseAgent(dsn=DNS)
- except OperationalError:
- warnings.warn(
- UserWarning(
- f"Skipping tests, because DB is not setup. {DNS}. To setup DB go to README.md"
- )
- )
- return False
- return True
+from .utils import PEPDBAgentContextManager
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestNamespace:
@@ -39,32 +13,36 @@ class TestNamespace:
Test function within namespace class
"""
- def test_annotation(self, initiate_pepdb_con):
- result = initiate_pepdb_con.namespace.get()
- assert len(result.results) == 3
-
- def test_annotation_private(self, initiate_pepdb_con):
- result = initiate_pepdb_con.namespace.get(admin="private_test")
- assert len(result.results) == 4
-
- def test_namespace_info(self, initiate_pepdb_con):
- initiate_pepdb_con.project.update(
- namespace="private_test",
- name="derive",
- tag="default",
- update_dict={"is_private": False},
- )
- result = initiate_pepdb_con.namespace.info()
- assert len(result.results) == 4
- assert result.results[3].number_of_projects == 1
+ def test_annotation(self):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.namespace.get()
+ assert len(result.results) == 3
+
+ def test_annotation_private(self):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.namespace.get(admin="private_test")
+ assert len(result.results) == 4
+
+ def test_namespace_info(self):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.update(
+ namespace="private_test",
+ name="derive",
+ tag="default",
+ update_dict={"is_private": False},
+ )
+ result = agent.namespace.info()
+ assert len(result.results) == 4
+ assert result.results[3].number_of_projects == 1
- def test_namespace_stats(self, initiate_pepdb_con):
- stat_result = initiate_pepdb_con.namespace.stats(monthly=True)
- assert next(iter(stat_result.projects_created.values()), 0) == 30
+ def test_namespace_stats(self):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ stat_result = agent.namespace.stats(monthly=True)
+ assert next(iter(stat_result.projects_created.values()), 0) == 30
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestFavorites:
@@ -72,24 +50,26 @@ class TestFavorites:
Test function within user class
"""
- def test_add_projects_to_favorites(self, initiate_pepdb_con):
- result = initiate_pepdb_con.annotation.get(
- namespace="namespace1",
- )
- for project in result.results:
- initiate_pepdb_con.user.add_project_to_favorites(
- "random_namespace", project.namespace, project.name, "default"
+ def test_add_projects_to_favorites(self):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.annotation.get(
+ namespace="namespace1",
)
- fav_results = initiate_pepdb_con.user.get_favorites("random_namespace")
+ for project in result.results:
+ agent.user.add_project_to_favorites(
+ "random_namespace", project.namespace, project.name, "default"
+ )
+ fav_results = agent.user.get_favorites("random_namespace")
- assert fav_results.count == len(result.results)
+ assert fav_results.count == len(result.results)
# This can fail if the order of the results is different
assert fav_results.results[0].namespace == result.results[0].namespace
- def test_count_project_none(self, initiate_pepdb_con):
- result = initiate_pepdb_con.user.get_favorites("private_test")
- assert result.count == 0
+ def test_count_project_none(self):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ result = agent.user.get_favorites("private_test")
+ assert result.count == 0
@pytest.mark.parametrize(
"namespace, name",
@@ -97,12 +77,13 @@ def test_count_project_none(self, initiate_pepdb_con):
["namespace1", "amendments1"],
],
)
- def test_count_project_one(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.user.add_project_to_favorites(namespace, namespace, name, "default")
- result = initiate_pepdb_con.user.get_favorites("namespace1")
- assert result.count == 1
- result1 = initiate_pepdb_con.user.get_favorites("private_test")
- assert result1.count == 0
+ def test_count_project_one(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.user.add_project_to_favorites(namespace, namespace, name, "default")
+ result = agent.user.get_favorites("namespace1")
+ assert result.count == 1
+ result1 = agent.user.get_favorites("private_test")
+ assert result1.count == 0
@pytest.mark.parametrize(
"namespace, name",
@@ -110,18 +91,19 @@ def test_count_project_one(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments1"],
],
)
- def test_remove_from_favorite(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.user.add_project_to_favorites("namespace1", namespace, name, "default")
- initiate_pepdb_con.user.add_project_to_favorites(
- "namespace1", namespace, "amendments2", "default"
- )
- result = initiate_pepdb_con.user.get_favorites("namespace1")
- assert result.count == len(result.results) == 2
- initiate_pepdb_con.user.remove_project_from_favorites(
- "namespace1", namespace, name, "default"
- )
- result = initiate_pepdb_con.user.get_favorites("namespace1")
- assert result.count == len(result.results) == 1
+ def test_remove_from_favorite(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.user.add_project_to_favorites("namespace1", namespace, name, "default")
+ agent.user.add_project_to_favorites(
+ "namespace1", namespace, "amendments2", "default"
+ )
+ result = agent.user.get_favorites("namespace1")
+ assert result.count == len(result.results) == 2
+ agent.user.remove_project_from_favorites(
+ "namespace1", namespace, name, "default"
+ )
+ result = agent.user.get_favorites("namespace1")
+ assert result.count == len(result.results) == 1
@pytest.mark.parametrize(
"namespace, name",
@@ -129,11 +111,12 @@ def test_remove_from_favorite(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments1"],
],
)
- def test_remove_from_favorite_error(self, initiate_pepdb_con, namespace, name):
- with pytest.raises(ProjectNotInFavorites):
- initiate_pepdb_con.user.remove_project_from_favorites(
- "namespace1", namespace, name, "default"
- )
+ def test_remove_from_favorite_error(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ with pytest.raises(ProjectNotInFavorites):
+ agent.user.remove_project_from_favorites(
+ "namespace1", namespace, name, "default"
+ )
@pytest.mark.parametrize(
"namespace, name",
@@ -141,12 +124,13 @@ def test_remove_from_favorite_error(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments1"],
],
)
- def test_favorites_duplication_error(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.user.add_project_to_favorites("namespace1", namespace, name, "default")
- with pytest.raises(ProjectAlreadyInFavorites):
- initiate_pepdb_con.user.add_project_to_favorites(
- "namespace1", namespace, name, "default"
- )
+ def test_favorites_duplication_error(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.user.add_project_to_favorites("namespace1", namespace, name, "default")
+ with pytest.raises(ProjectAlreadyInFavorites):
+ agent.user.add_project_to_favorites(
+ "namespace1", namespace, name, "default"
+ )
@pytest.mark.parametrize(
"namespace, name",
@@ -154,12 +138,13 @@ def test_favorites_duplication_error(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments1"],
],
)
- def test_annotation_favorite_number(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.user.add_project_to_favorites("namespace1", namespace, name, "default")
- annotations_in_namespace = initiate_pepdb_con.annotation.get("namespace1")
-
- for prj_annot in annotations_in_namespace.results:
- if prj_annot.name == name:
- assert prj_annot.stars_number == 1
- else:
- assert prj_annot.stars_number == 0
+ def test_annotation_favorite_number(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.user.add_project_to_favorites("namespace1", namespace, name, "default")
+ annotations_in_namespace = agent.annotation.get("namespace1")
+
+ for prj_annot in annotations_in_namespace.results:
+ if prj_annot.name == name:
+ assert prj_annot.stars_number == 1
+ else:
+ assert prj_annot.stars_number == 0
diff --git a/tests/test_project.py b/tests/test_project.py
index bc3c84e..29b08cb 100644
--- a/tests/test_project.py
+++ b/tests/test_project.py
@@ -1,43 +1,14 @@
-import os
-import warnings
-
import numpy as np
import peppy
import pytest
-from sqlalchemy.exc import OperationalError
-import pepdbagent
from pepdbagent.exceptions import ProjectNotFoundError
-from .conftest import DNS
-
-DATA_PATH = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "tests",
- "data",
-)
-
-
-def get_path_to_example_file(namespace, project_name):
- return os.path.join(DATA_PATH, namespace, project_name, "project_config.yaml")
-
-
-def db_setup():
- # Check if the database is setup
- try:
- pepdbagent.PEPDatabaseAgent(dsn=DNS)
- except OperationalError:
- warnings.warn(
- UserWarning(
- f"Skipping tests, because DB is not setup. {DNS}. To setup DB go to README.md"
- )
- )
- return False
- return True
+from .utils import PEPDBAgentContextManager, get_path_to_example_file, list_of_available_peps
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestProject:
@@ -45,22 +16,24 @@ class TestProject:
Test project methods
"""
- def test_create_project(self, initiate_empty_pepdb_con, list_of_available_peps):
- prj = peppy.Project(list_of_available_peps["namespace3"]["subtables"])
- initiate_empty_pepdb_con.project.create(
- prj, namespace="test", name="imply", overwrite=True
- )
- assert True
-
- def test_create_project_from_dict(self, initiate_empty_pepdb_con, list_of_available_peps):
- prj = peppy.Project(list_of_available_peps["namespace3"]["subtables"])
- initiate_empty_pepdb_con.project.create(
- prj.to_dict(extended=True, orient="records"),
- namespace="test",
- name="imply",
- overwrite=True,
- )
- assert True
+ def test_create_project(self):
+ with PEPDBAgentContextManager(add_data=False) as agent:
+ prj = peppy.Project(list_of_available_peps()["namespace3"]["subtables"])
+ agent.project.create(
+ prj, namespace="test", name="imply", overwrite=True
+ )
+ assert True
+
+ def test_create_project_from_dict(self):
+ with PEPDBAgentContextManager(add_data=False) as agent:
+ prj = peppy.Project(list_of_available_peps()["namespace3"]["subtables"])
+ agent.project.create(
+ prj.to_dict(extended=True, orient="records"),
+ namespace="test",
+ name="imply",
+ overwrite=True,
+ )
+ assert True
@pytest.mark.parametrize(
"namespace, name",
@@ -74,12 +47,13 @@ def test_create_project_from_dict(self, initiate_empty_pepdb_con, list_of_availa
["namespace3", "subtable2"],
],
)
- def test_get_project(self, initiate_pepdb_con, namespace, name):
- kk = initiate_pepdb_con.project.get(
- namespace=namespace, name=name, tag="default", raw=False
- )
- ff = peppy.Project(get_path_to_example_file(namespace, name))
- assert kk == ff
+ def test_get_project(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ kk = agent.project.get(
+ namespace=namespace, name=name, tag="default", raw=False
+ )
+ ff = peppy.Project(get_path_to_example_file(namespace, name))
+ assert kk == ff
@pytest.mark.parametrize(
"namespace, name",
@@ -87,17 +61,18 @@ def test_get_project(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments1"],
],
)
- def test_get_config(self, initiate_pepdb_con, namespace, name):
- description = ""
- kk = initiate_pepdb_con.project.get_config(
- namespace=namespace,
- name=name,
- tag="default",
- )
- ff = peppy.Project(get_path_to_example_file(namespace, name))
- ff.description = description
- ff.name = name
- assert kk == ff.config
+ def test_get_config(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ description = ""
+ kk = agent.project.get_config(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ )
+ ff = peppy.Project(get_path_to_example_file(namespace, name))
+ ff.description = description
+ ff.name = name
+ assert kk == ff.config
@pytest.mark.parametrize(
"namespace, name",
@@ -105,18 +80,19 @@ def test_get_config(self, initiate_pepdb_con, namespace, name):
["namespace3", "subtables"],
],
)
- def test_get_subsamples(self, initiate_pepdb_con, namespace, name):
- prj_subtables = initiate_pepdb_con.project.get_subsamples(
- namespace=namespace,
- name=name,
- tag="default",
- )
- orgiginal_prj = peppy.Project(get_path_to_example_file(namespace, name))
+ def test_get_subsamples(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj_subtables = agent.project.get_subsamples(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ )
+ orgiginal_prj = peppy.Project(get_path_to_example_file(namespace, name))
- assert (
- prj_subtables
- == orgiginal_prj.to_dict(extended=True, orient="records")["_subsample_list"]
- )
+ assert (
+ prj_subtables
+ == orgiginal_prj.to_dict(extended=True, orient="records")["_subsample_list"]
+ )
@pytest.mark.parametrize(
"namespace, name",
@@ -124,15 +100,16 @@ def test_get_subsamples(self, initiate_pepdb_con, namespace, name):
["namespace3", "subtables"],
],
)
- def test_get_samples_raw(self, initiate_pepdb_con, namespace, name):
- prj_samples = initiate_pepdb_con.project.get_samples(
- namespace=namespace, name=name, tag="default", raw=True
- )
- orgiginal_prj = peppy.Project(get_path_to_example_file(namespace, name))
+ def test_get_samples_raw(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj_samples = agent.project.get_samples(
+ namespace=namespace, name=name, tag="default", raw=True
+ )
+ orgiginal_prj = peppy.Project(get_path_to_example_file(namespace, name))
- assert (
- prj_samples == orgiginal_prj.to_dict(extended=True, orient="records")["_sample_dict"]
- )
+ assert (
+ prj_samples == orgiginal_prj.to_dict(extended=True, orient="records")["_sample_dict"]
+ )
@pytest.mark.parametrize(
"namespace, name",
@@ -140,18 +117,19 @@ def test_get_samples_raw(self, initiate_pepdb_con, namespace, name):
["namespace3", "subtables"],
],
)
- def test_get_samples_processed(self, initiate_pepdb_con, namespace, name):
- prj_samples = initiate_pepdb_con.project.get_samples(
- namespace=namespace,
- name=name,
- tag="default",
- raw=False,
- )
- orgiginal_prj = peppy.Project(get_path_to_example_file(namespace, name))
+ def test_get_samples_processed(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj_samples = agent.project.get_samples(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ raw=False,
+ )
+ orgiginal_prj = peppy.Project(get_path_to_example_file(namespace, name))
- assert prj_samples == orgiginal_prj.sample_table.replace({np.nan: None}).to_dict(
- orient="records"
- )
+ assert prj_samples == orgiginal_prj.sample_table.replace({np.nan: None}).to_dict(
+ orient="records"
+ )
@pytest.mark.parametrize(
"namespace, name,tag",
@@ -163,9 +141,10 @@ def test_get_samples_processed(self, initiate_pepdb_con, namespace, name):
["namespace1", "incorrect_name", "default"],
],
)
- def test_get_project_error(self, initiate_pepdb_con, namespace, name, tag):
- with pytest.raises(ProjectNotFoundError, match="Project does not exist."):
- initiate_pepdb_con.project.get(namespace=namespace, name=name, tag=tag)
+ def test_get_project_error(self, namespace, name, tag):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ with pytest.raises(ProjectNotFoundError, match="Project does not exist."):
+ agent.project.get(namespace=namespace, name=name, tag=tag)
@pytest.mark.parametrize(
"namespace, name",
@@ -176,18 +155,19 @@ def test_get_project_error(self, initiate_pepdb_con, namespace, name, tag):
["namespace2", "imply"],
],
)
- def test_overwrite_project(self, initiate_pepdb_con, namespace, name):
- new_prj = initiate_pepdb_con.project.get(namespace="namespace1", name="basic", raw=False)
-
- initiate_pepdb_con.project.create(
- project=new_prj,
- namespace=namespace,
- name=name,
- tag="default",
- overwrite=True,
- )
+ def test_overwrite_project(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ new_prj = agent.project.get(namespace="namespace1", name="basic", raw=False)
+
+ agent.project.create(
+ project=new_prj,
+ namespace=namespace,
+ name=name,
+ tag="default",
+ overwrite=True,
+ )
- assert initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False) == new_prj
+ assert agent.project.get(namespace=namespace, name=name, raw=False) == new_prj
@pytest.mark.parametrize(
"namespace, name",
@@ -198,11 +178,12 @@ def test_overwrite_project(self, initiate_pepdb_con, namespace, name):
["namespace2", "imply"],
],
)
- def test_delete_project(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.project.delete(namespace=namespace, name=name, tag="default")
+ def test_delete_project(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.delete(namespace=namespace, name=name, tag="default")
- with pytest.raises(ProjectNotFoundError, match="Project does not exist."):
- initiate_pepdb_con.project.get(namespace=namespace, name=name, tag="default")
+ with pytest.raises(ProjectNotFoundError, match="Project does not exist."):
+ agent.project.get(namespace=namespace, name=name, tag="default")
@pytest.mark.parametrize(
"namespace, name",
@@ -213,23 +194,24 @@ def test_delete_project(self, initiate_pepdb_con, namespace, name):
["namespace2", "imply"],
],
)
- def test_fork_projects(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.project.fork(
- original_namespace=namespace,
- original_name=name,
- original_tag="default",
- fork_namespace="new_namespace",
- fork_name="new_name",
- fork_tag="new_tag",
- )
+ def test_fork_projects(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.fork(
+ original_namespace=namespace,
+ original_name=name,
+ original_tag="default",
+ fork_namespace="new_namespace",
+ fork_name="new_name",
+ fork_tag="new_tag",
+ )
- assert initiate_pepdb_con.project.exists(
- namespace="new_namespace", name="new_name", tag="new_tag"
- )
- result = initiate_pepdb_con.annotation.get(
- namespace="new_namespace", name="new_name", tag="new_tag"
- )
- assert result.results[0].forked_from == f"{namespace}/{name}:default"
+ assert agent.project.exists(
+ namespace="new_namespace", name="new_name", tag="new_tag"
+ )
+ result = agent.annotation.get(
+ namespace="new_namespace", name="new_name", tag="new_tag"
+ )
+ assert result.results[0].forked_from == f"{namespace}/{name}:default"
@pytest.mark.parametrize(
"namespace, name",
@@ -238,26 +220,27 @@ def test_fork_projects(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments2"],
],
)
- def test_parent_project_delete(self, initiate_pepdb_con, namespace, name):
+ def test_parent_project_delete(self, namespace, name):
"""
Test if parent project is deleted, forked project is not deleted
"""
- initiate_pepdb_con.project.fork(
- original_namespace=namespace,
- original_name=name,
- original_tag="default",
- fork_namespace="new_namespace",
- fork_name="new_name",
- fork_tag="new_tag",
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.fork(
+ original_namespace=namespace,
+ original_name=name,
+ original_tag="default",
+ fork_namespace="new_namespace",
+ fork_name="new_name",
+ fork_tag="new_tag",
+ )
- assert initiate_pepdb_con.project.exists(
- namespace="new_namespace", name="new_name", tag="new_tag"
- )
- initiate_pepdb_con.project.delete(namespace=namespace, name=name, tag="default")
- assert initiate_pepdb_con.project.exists(
- namespace="new_namespace", name="new_name", tag="new_tag"
- )
+ assert agent.project.exists(
+ namespace="new_namespace", name="new_name", tag="new_tag"
+ )
+ agent.project.delete(namespace=namespace, name=name, tag="default")
+ assert agent.project.exists(
+ namespace="new_namespace", name="new_name", tag="new_tag"
+ )
@pytest.mark.parametrize(
"namespace, name",
@@ -266,27 +249,28 @@ def test_parent_project_delete(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments2"],
],
)
- def test_child_project_delete(self, initiate_pepdb_con, namespace, name):
+ def test_child_project_delete(self, namespace, name):
"""
Test if child project is deleted, parent project is not deleted
"""
- initiate_pepdb_con.project.fork(
- original_namespace=namespace,
- original_name=name,
- original_tag="default",
- fork_namespace="new_namespace",
- fork_name="new_name",
- fork_tag="new_tag",
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.fork(
+ original_namespace=namespace,
+ original_name=name,
+ original_tag="default",
+ fork_namespace="new_namespace",
+ fork_name="new_name",
+ fork_tag="new_tag",
+ )
- assert initiate_pepdb_con.project.exists(
- namespace="new_namespace", name="new_name", tag="new_tag"
- )
- assert initiate_pepdb_con.project.exists(namespace=namespace, name=name, tag="default")
- initiate_pepdb_con.project.delete(
- namespace="new_namespace", name="new_name", tag="new_tag"
- )
- assert initiate_pepdb_con.project.exists(namespace=namespace, name=name, tag="default")
+ assert agent.project.exists(
+ namespace="new_namespace", name="new_name", tag="new_tag"
+ )
+ assert agent.project.exists(namespace=namespace, name=name, tag="default")
+ agent.project.delete(
+ namespace="new_namespace", name="new_name", tag="new_tag"
+ )
+ assert agent.project.exists(namespace=namespace, name=name, tag="default")
@pytest.mark.parametrize(
"namespace, name",
@@ -295,33 +279,34 @@ def test_child_project_delete(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments2"],
],
)
- def test_project_can_be_forked_twice(self, initiate_pepdb_con, namespace, name):
+ def test_project_can_be_forked_twice(self, namespace, name):
"""
Test if project can be forked twice
"""
- initiate_pepdb_con.project.fork(
- original_namespace=namespace,
- original_name=name,
- original_tag="default",
- fork_namespace="new_namespace",
- fork_name="new_name",
- fork_tag="new_tag",
- )
- initiate_pepdb_con.project.fork(
- original_namespace=namespace,
- original_name=name,
- original_tag="default",
- fork_namespace="new_namespace2",
- fork_name="new_name2",
- fork_tag="new_tag2",
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.fork(
+ original_namespace=namespace,
+ original_name=name,
+ original_tag="default",
+ fork_namespace="new_namespace",
+ fork_name="new_name",
+ fork_tag="new_tag",
+ )
+ agent.project.fork(
+ original_namespace=namespace,
+ original_name=name,
+ original_tag="default",
+ fork_namespace="new_namespace2",
+ fork_name="new_name2",
+ fork_tag="new_tag2",
+ )
- result = initiate_pepdb_con.annotation.get(
- namespace="new_namespace", name="new_name", tag="new_tag"
- )
- assert result.results[0].forked_from == f"{namespace}/{name}:default"
+ result = agent.annotation.get(
+ namespace="new_namespace", name="new_name", tag="new_tag"
+ )
+ assert result.results[0].forked_from == f"{namespace}/{name}:default"
- result = initiate_pepdb_con.annotation.get(
- namespace="new_namespace2", name="new_name2", tag="new_tag2"
- )
- assert result.results[0].forked_from == f"{namespace}/{name}:default"
+ result = agent.annotation.get(
+ namespace="new_namespace2", name="new_name2", tag="new_tag2"
+ )
+ assert result.results[0].forked_from == f"{namespace}/{name}:default"
diff --git a/tests/test_samples.py b/tests/test_samples.py
index 99c73b6..f1bfb3a 100644
--- a/tests/test_samples.py
+++ b/tests/test_samples.py
@@ -1,38 +1,11 @@
-import os
-import warnings
-
import peppy
import pytest
-from sqlalchemy.exc import OperationalError
-
-import pepdbagent
from pepdbagent.exceptions import SampleNotFoundError
-from .conftest import DNS
-
-DATA_PATH = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "tests",
- "data",
-)
-
-
-def db_setup():
- # Check if the database is setup
- try:
- pepdbagent.PEPDatabaseAgent(dsn=DNS)
- except OperationalError:
- warnings.warn(
- UserWarning(
- f"Skipping tests, because DB is not setup. {DNS}. To setup DB go to README.md"
- )
- )
- return False
- return True
-
+from .utils import PEPDBAgentContextManager
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestSamples:
@@ -42,10 +15,11 @@ class TestSamples:
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_retrieve_one_sample(self, initiate_pepdb_con, namespace, name, sample_name):
- one_sample = initiate_pepdb_con.sample.get(namespace, name, sample_name, raw=False)
- assert isinstance(one_sample, peppy.Sample)
- assert one_sample.sample_name == sample_name
+ def test_retrieve_one_sample(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ one_sample = agent.sample.get(namespace, name, sample_name, raw=False)
+ assert isinstance(one_sample, peppy.Sample)
+ assert one_sample.sample_name == sample_name
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -53,10 +27,11 @@ def test_retrieve_one_sample(self, initiate_pepdb_con, namespace, name, sample_n
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_retrieve_raw_sample(self, initiate_pepdb_con, namespace, name, sample_name):
- one_sample = initiate_pepdb_con.sample.get(namespace, name, sample_name)
- assert isinstance(one_sample, dict)
- assert one_sample["sample_name"] == sample_name
+ def test_retrieve_raw_sample(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ one_sample = agent.sample.get(namespace, name, sample_name)
+ assert isinstance(one_sample, dict)
+ assert one_sample["sample_name"] == sample_name
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -65,11 +40,12 @@ def test_retrieve_raw_sample(self, initiate_pepdb_con, namespace, name, sample_n
],
)
def test_retrieve_sample_with_modified_sample_id(
- self, initiate_pepdb_con, namespace, name, sample_name
+ self, namespace, name, sample_name
):
- one_sample = initiate_pepdb_con.sample.get(namespace, name, sample_name, raw=False)
- assert isinstance(one_sample, peppy.Sample)
- assert one_sample.sample_id == "frog_1"
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ one_sample = agent.sample.get(namespace, name, sample_name, raw=False)
+ assert isinstance(one_sample, peppy.Sample)
+ assert one_sample.sample_id == "frog_1"
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -77,16 +53,17 @@ def test_retrieve_sample_with_modified_sample_id(
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_update(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.sample.update(
- namespace=namespace,
- name=name,
- tag="default",
- sample_name=sample_name,
- update_dict={"organism": "butterfly"},
- )
- one_sample = initiate_pepdb_con.sample.get(namespace, name, sample_name, raw=False)
- assert one_sample.organism == "butterfly"
+ def test_update(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.sample.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ sample_name=sample_name,
+ update_dict={"organism": "butterfly"},
+ )
+ one_sample = agent.sample.get(namespace, name, sample_name, raw=False)
+ assert one_sample.organism == "butterfly"
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -94,16 +71,17 @@ def test_update(self, initiate_pepdb_con, namespace, name, sample_name):
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_update_sample_name(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.sample.update(
- namespace=namespace,
- name=name,
- tag="default",
- sample_name=sample_name,
- update_dict={"sample_name": "butterfly"},
- )
- one_sample = initiate_pepdb_con.sample.get(namespace, name, "butterfly", raw=False)
- assert one_sample.sample_name == "butterfly"
+ def test_update_sample_name(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.sample.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ sample_name=sample_name,
+ update_dict={"sample_name": "butterfly"},
+ )
+ one_sample = agent.sample.get(namespace, name, "butterfly", raw=False)
+ assert one_sample.sample_name == "butterfly"
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -111,16 +89,17 @@ def test_update_sample_name(self, initiate_pepdb_con, namespace, name, sample_na
["namespace2", "custom_index", "frog_1"],
],
)
- def test_update_custom_sample_id(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.sample.update(
- namespace=namespace,
- name=name,
- tag="default",
- sample_name=sample_name,
- update_dict={"sample_id": "butterfly"},
- )
- one_sample = initiate_pepdb_con.sample.get(namespace, name, "butterfly", raw=False)
- assert one_sample.sample_id == "butterfly"
+ def test_update_custom_sample_id(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.sample.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ sample_name=sample_name,
+ update_dict={"sample_id": "butterfly"},
+ )
+ one_sample = agent.sample.get(namespace, name, "butterfly", raw=False)
+ assert one_sample.sample_id == "butterfly"
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -128,17 +107,18 @@ def test_update_custom_sample_id(self, initiate_pepdb_con, namespace, name, samp
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_add_new_attributes(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.sample.update(
- namespace=namespace,
- name=name,
- tag="default",
- sample_name=sample_name,
- update_dict={"new_attr": "butterfly"},
- )
- prj = initiate_pepdb_con.project.get(namespace, name, raw=False)
-
- assert prj.get_sample(sample_name).new_attr == "butterfly"
+ def test_add_new_attributes(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.sample.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ sample_name=sample_name,
+ update_dict={"new_attr": "butterfly"},
+ )
+ prj = agent.project.get(namespace, name, raw=False)
+
+ assert prj.get_sample(sample_name).new_attr == "butterfly"
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -146,21 +126,22 @@ def test_add_new_attributes(self, initiate_pepdb_con, namespace, name, sample_na
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_project_timestamp_was_changed(self, initiate_pepdb_con, namespace, name, sample_name):
- annotation1 = initiate_pepdb_con.annotation.get(namespace, name, "default")
- import time
-
- time.sleep(0.2)
- initiate_pepdb_con.sample.update(
- namespace=namespace,
- name=name,
- tag="default",
- sample_name=sample_name,
- update_dict={"new_attr": "butterfly"},
- )
- annotation2 = initiate_pepdb_con.annotation.get(namespace, name, "default")
-
- assert annotation1.results[0].last_update_date != annotation2.results[0].last_update_date
+ def test_project_timestamp_was_changed(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ annotation1 = agent.annotation.get(namespace, name, "default")
+ import time
+
+ time.sleep(0.2)
+ agent.sample.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ sample_name=sample_name,
+ update_dict={"new_attr": "butterfly"},
+ )
+ annotation2 = agent.annotation.get(namespace, name, "default")
+
+ assert annotation1.results[0].last_update_date != annotation2.results[0].last_update_date
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -168,16 +149,17 @@ def test_project_timestamp_was_changed(self, initiate_pepdb_con, namespace, name
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_delete_sample(self, initiate_pepdb_con, namespace, name, sample_name):
- one_sample = initiate_pepdb_con.sample.get(namespace, name, sample_name, raw=False)
- assert isinstance(one_sample, peppy.Sample)
+ def test_delete_sample(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ one_sample = agent.sample.get(namespace, name, sample_name, raw=False)
+ assert isinstance(one_sample, peppy.Sample)
- initiate_pepdb_con.sample.delete(namespace, name, tag="default", sample_name=sample_name)
+ agent.sample.delete(namespace, name, tag="default", sample_name=sample_name)
- with pytest.raises(SampleNotFoundError):
- initiate_pepdb_con.sample.get(
- namespace, name, tag="default", sample_name=sample_name, raw=False
- )
+ with pytest.raises(SampleNotFoundError):
+ agent.sample.get(
+ namespace, name, tag="default", sample_name=sample_name, raw=False
+ )
@pytest.mark.parametrize(
"namespace, name, tag, sample_dict",
@@ -193,14 +175,15 @@ def test_delete_sample(self, initiate_pepdb_con, namespace, name, sample_name):
],
],
)
- def test_add_sample(self, initiate_pepdb_con, namespace, name, tag, sample_dict):
- prj = initiate_pepdb_con.project.get(namespace, name, raw=False)
- initiate_pepdb_con.sample.add(namespace, name, tag, sample_dict)
+ def test_add_sample(self, namespace, name, tag, sample_dict):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj = agent.project.get(namespace, name, raw=False)
+ agent.sample.add(namespace, name, tag, sample_dict)
- prj2 = initiate_pepdb_con.project.get(namespace, name, raw=False)
+ prj2 = agent.project.get(namespace, name, raw=False)
- assert len(prj.samples) + 1 == len(prj2.samples)
- assert prj2.samples[-1].sample_name == sample_dict["sample_name"]
+ assert len(prj.samples) + 1 == len(prj2.samples)
+ assert prj2.samples[-1].sample_name == sample_dict["sample_name"]
@pytest.mark.parametrize(
"namespace, name, tag, sample_dict",
@@ -216,17 +199,18 @@ def test_add_sample(self, initiate_pepdb_con, namespace, name, tag, sample_dict)
],
],
)
- def test_overwrite_sample(self, initiate_pepdb_con, namespace, name, tag, sample_dict):
- assert (
- initiate_pepdb_con.project.get(namespace, name, raw=False).get_sample("pig_0h").time
- == "0"
- )
- initiate_pepdb_con.sample.add(namespace, name, tag, sample_dict, overwrite=True)
-
- assert (
- initiate_pepdb_con.project.get(namespace, name, raw=False).get_sample("pig_0h").time
- == "new_time"
- )
+ def test_overwrite_sample(self, namespace, name, tag, sample_dict):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ assert (
+ agent.project.get(namespace, name, raw=False).get_sample("pig_0h").time
+ == "0"
+ )
+ agent.sample.add(namespace, name, tag, sample_dict, overwrite=True)
+
+ assert (
+ agent.project.get(namespace, name, raw=False).get_sample("pig_0h").time
+ == "new_time"
+ )
@pytest.mark.parametrize(
"namespace, name, tag, sample_dict",
@@ -242,10 +226,11 @@ def test_overwrite_sample(self, initiate_pepdb_con, namespace, name, tag, sample
],
],
)
- def test_delete_and_add(self, initiate_pepdb_con, namespace, name, tag, sample_dict):
- prj = initiate_pepdb_con.project.get(namespace, name, raw=False)
- sample_dict = initiate_pepdb_con.sample.get(namespace, name, "pig_0h", raw=True)
- initiate_pepdb_con.sample.delete(namespace, name, tag, "pig_0h")
- initiate_pepdb_con.sample.add(namespace, name, tag, sample_dict)
- prj2 = initiate_pepdb_con.project.get(namespace, name, raw=False)
- assert prj.get_sample("pig_0h").to_dict() == prj2.get_sample("pig_0h").to_dict()
+ def test_delete_and_add(self, namespace, name, tag, sample_dict):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj = agent.project.get(namespace, name, raw=False)
+ sample_dict = agent.sample.get(namespace, name, "pig_0h", raw=True)
+ agent.sample.delete(namespace, name, tag, "pig_0h")
+ agent.sample.add(namespace, name, tag, sample_dict)
+ prj2 = agent.project.get(namespace, name, raw=False)
+ assert prj.get_sample("pig_0h").to_dict() == prj2.get_sample("pig_0h").to_dict()
diff --git a/tests/test_updates.py b/tests/test_updates.py
index 6fb4e02..d6155e0 100644
--- a/tests/test_updates.py
+++ b/tests/test_updates.py
@@ -1,38 +1,11 @@
-import os
-import warnings
-
import peppy
import pytest
-from sqlalchemy.exc import OperationalError
-
-import pepdbagent
from pepdbagent.exceptions import ProjectDuplicatedSampleGUIDsError, SampleTableUpdateError
-from .conftest import DNS
-
-DATA_PATH = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "tests",
- "data",
-)
-
-
-def db_setup():
- # Check if the database is setup
- try:
- pepdbagent.PEPDatabaseAgent(dsn=DNS)
- except OperationalError:
- warnings.warn(
- UserWarning(
- f"Skipping tests, because DB is not setup. {DNS}. To setup DB go to README.md"
- )
- )
- return False
- return True
-
+from .utils import PEPDBAgentContextManager
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestProjectUpdate:
@@ -43,14 +16,15 @@ class TestProjectUpdate:
["namespace1", "amendments2", "name2"],
],
)
- def test_update_project_name(self, initiate_pepdb_con, namespace, name, new_name):
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"name": new_name},
- )
- assert initiate_pepdb_con.project.exists(namespace=namespace, name=new_name, tag="default")
+ def test_update_project_name(self, namespace, name, new_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"name": new_name},
+ )
+ assert agent.project.exists(namespace=namespace, name=new_name, tag="default")
@pytest.mark.parametrize(
"namespace, name,new_name",
@@ -59,18 +33,19 @@ def test_update_project_name(self, initiate_pepdb_con, namespace, name, new_name
["namespace1", "amendments2", "name2"],
],
)
- def test_update_project_name_in_config(self, initiate_pepdb_con, namespace, name, new_name):
- prj = initiate_pepdb_con.project.get(
- namespace=namespace, name=name, raw=False, with_id=True
- )
- prj.name = new_name
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"project": prj},
- )
- assert initiate_pepdb_con.project.exists(namespace=namespace, name=new_name, tag="default")
+ def test_update_project_name_in_config(self, namespace, name, new_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj = agent.project.get(
+ namespace=namespace, name=name, raw=False, with_id=True
+ )
+ prj.name = new_name
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"project": prj},
+ )
+ assert agent.project.exists(namespace=namespace, name=new_name, tag="default")
@pytest.mark.parametrize(
"namespace, name, new_tag",
@@ -79,14 +54,15 @@ def test_update_project_name_in_config(self, initiate_pepdb_con, namespace, name
["namespace1", "amendments2", "tag2"],
],
)
- def test_update_project_tag(self, initiate_pepdb_con, namespace, name, new_tag):
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"tag": new_tag},
- )
- assert initiate_pepdb_con.project.exists(namespace=namespace, name=name, tag=new_tag)
+ def test_update_project_tag(self, namespace, name, new_tag):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"tag": new_tag},
+ )
+ assert agent.project.exists(namespace=namespace, name=name, tag=new_tag)
@pytest.mark.parametrize(
"namespace, name, new_description",
@@ -96,21 +72,22 @@ def test_update_project_tag(self, initiate_pepdb_con, namespace, name, new_tag):
],
)
def test_update_project_description(
- self, initiate_pepdb_con, namespace, name, new_description
+ self, namespace, name, new_description
):
- prj = initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False)
- prj.description = new_description
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"description": new_description},
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj = agent.project.get(namespace=namespace, name=name, raw=False)
+ prj.description = new_description
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"description": new_description},
+ )
- assert (
- initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False).description
- == new_description
- )
+ assert (
+ agent.project.get(namespace=namespace, name=name, raw=False).description
+ == new_description
+ )
@pytest.mark.parametrize(
"namespace, name, new_description",
@@ -119,23 +96,24 @@ def test_update_project_description(
],
)
def test_update_project_description_in_config(
- self, initiate_pepdb_con, namespace, name, new_description
+ self, namespace, name, new_description
):
- prj = initiate_pepdb_con.project.get(
- namespace=namespace, name=name, raw=False, with_id=True
- )
- prj.description = new_description
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"project": prj},
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ prj = agent.project.get(
+ namespace=namespace, name=name, raw=False, with_id=True
+ )
+ prj.description = new_description
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"project": prj},
+ )
- assert (
- initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False).description
- == new_description
- )
+ assert (
+ agent.project.get(namespace=namespace, name=name, raw=False).description
+ == new_description
+ )
@pytest.mark.parametrize(
"namespace, name",
@@ -144,36 +122,36 @@ def test_update_project_description_in_config(
["namespace3", "subtable1"],
],
)
- def test_update_whole_project(self, initiate_pepdb_con, namespace, name):
- new_prj = initiate_pepdb_con.project.get(namespace="namespace1", name="basic", raw=False)
- # update name. If name is different, it will update name too
- new_prj.name = name
- with pytest.raises(SampleTableUpdateError):
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"project": new_prj},
- )
+ def test_update_whole_project(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ new_prj = agent.project.get(namespace="namespace1", name="basic", raw=False)
+ # update name. If name is different, it will update name too
+ new_prj.name = name
+ with pytest.raises(SampleTableUpdateError):
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"project": new_prj},
+ )
@pytest.mark.parametrize(
"namespace, name, pep_schema",
[
["namespace1", "amendments1", "schema1"],
["namespace2", "derive", "schema3"],
- ["namespace1", "basic", "schema4"],
- ["namespace2", "derive", "schema5"],
],
)
- def test_update_pep_schema(self, initiate_pepdb_con, namespace, name, pep_schema):
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"pep_schema": pep_schema},
- )
- res = initiate_pepdb_con.annotation.get(namespace, name, "default")
- assert res.results[0].pep_schema == pep_schema
+ def test_update_pep_schema(self, namespace, name, pep_schema):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"pep_schema": pep_schema},
+ )
+ res = agent.annotation.get(namespace, name, "default")
+ assert res.results[0].pep_schema == pep_schema
@pytest.mark.parametrize(
"namespace, name",
@@ -181,22 +159,23 @@ def test_update_pep_schema(self, initiate_pepdb_con, namespace, name, pep_schema
["namespace1", "amendments1"],
],
)
- def test_update_project_private(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"is_private": True},
- )
+ def test_update_project_private(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"is_private": True},
+ )
- is_private = (
- initiate_pepdb_con.annotation.get(
- namespace=namespace, name=name, tag="default", admin=[namespace]
+ is_private = (
+ agent.annotation.get(
+ namespace=namespace, name=name, tag="default", admin=[namespace]
+ )
+ .results[0]
+ .is_private
)
- .results[0]
- .is_private
- )
- assert is_private is True
+ assert is_private is True
@pytest.mark.parametrize(
"namespace, name",
@@ -204,39 +183,40 @@ def test_update_project_private(self, initiate_pepdb_con, namespace, name):
["namespace1", "amendments1"],
],
)
- def test_update_project_pop(self, initiate_pepdb_con, namespace, name):
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"pop": True},
- )
+ def test_update_project_pop(self, namespace, name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"pop": True},
+ )
- pop = (
- initiate_pepdb_con.annotation.get(
- namespace=namespace, name=name, tag="default", admin=[namespace]
+ pop = (
+ agent.annotation.get(
+ namespace=namespace, name=name, tag="default", admin=[namespace]
+ )
+ .results[0]
+ .pop
)
- .results[0]
- .pop
- )
- assert pop is True
+ assert pop is True
- # Update to pop = False and check if it is updated
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"pop": False},
- )
+ # Update to pop = False and check if it is updated
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"pop": False},
+ )
- pop = (
- initiate_pepdb_con.annotation.get(
- namespace=namespace, name=name, tag="default", admin=[namespace]
+ pop = (
+ agent.annotation.get(
+ namespace=namespace, name=name, tag="default", admin=[namespace]
+ )
+ .results[0]
+ .pop
)
- .results[0]
- .pop
- )
- assert pop is False
+ assert pop is False
@pytest.mark.parametrize(
"namespace, name",
@@ -244,42 +224,43 @@ def test_update_project_pop(self, initiate_pepdb_con, namespace, name):
["namespace1", "basic"],
],
)
- def test_project_can_have_2_sample_names(self, initiate_pepdb_con, namespace, name):
+ def test_project_can_have_2_sample_names(self, namespace, name):
"""
In PEP 2.1.0 project can have 2 rows with same sample name,
ensure that update works correctly
"""
- new_prj = initiate_pepdb_con.project.get(
- namespace=namespace, name=name, raw=False, with_id=True
- )
- prj_dict = new_prj.to_dict(extended=True, orient="records")
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ new_prj = agent.project.get(
+ namespace=namespace, name=name, raw=False, with_id=True
+ )
+ prj_dict = new_prj.to_dict(extended=True, orient="records")
- prj_dict["_sample_dict"].append(
- {"file": "data/frog23_data.txt", "protocol": "anySample3Type", "sample_name": "frog_2"}
- )
- prj_dict["_sample_dict"].append(
- {
- "file": "data/frog23_data.txt4",
- "protocol": "anySample3Type4",
- "sample_name": "frog_2",
- }
- )
+ prj_dict["_sample_dict"].append(
+ {"file": "data/frog23_data.txt", "protocol": "anySample3Type", "sample_name": "frog_2"}
+ )
+ prj_dict["_sample_dict"].append(
+ {
+ "file": "data/frog23_data.txt4",
+ "protocol": "anySample3Type4",
+ "sample_name": "frog_2",
+ }
+ )
- new_prj.description = "new_description"
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"project": peppy.Project.from_dict(prj_dict)},
- )
+ new_prj.description = "new_description"
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"project": peppy.Project.from_dict(prj_dict)},
+ )
- prj = initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=True)
+ prj = agent.project.get(namespace=namespace, name=name, raw=True)
- assert len(prj["_sample_dict"]) == 4
+ assert len(prj["_sample_dict"]) == 4
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestUpdateProjectWithId:
@@ -290,7 +271,7 @@ class TestUpdateProjectWithId:
["namespace3", "subtable1"],
],
)
- def test_update_whole_project_with_id(self, initiate_pepdb_con, namespace, name):
+ def test_update_whole_project_with_id(self, namespace, name):
pass
# TODO: write more tests
@@ -299,26 +280,21 @@ def test_update_whole_project_with_id(self, initiate_pepdb_con, namespace, name)
"namespace, name",
[
["namespace1", "amendments1"],
- # ["namespace3", "subtable1"],
],
)
def test_update_project_with_duplicated_sample_guids(
- self, initiate_pepdb_con, namespace, name
+ self, namespace, name
):
- new_prj = initiate_pepdb_con.project.get(
- namespace=namespace, name=name, raw=True, with_id=True
- )
- new_prj["_sample_dict"].append(new_prj["_sample_dict"][0])
-
- with pytest.raises(ProjectDuplicatedSampleGUIDsError):
- initiate_pepdb_con.project.update(
- namespace=namespace,
- name=name,
- tag="default",
- update_dict={"project": peppy.Project.from_dict(new_prj)},
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ new_prj = agent.project.get(
+ namespace=namespace, name=name, raw=True, with_id=True
)
-
-
-class TestProjectSamplesUpdates:
- ...
- # TODO: write tests
+ new_prj["_sample_dict"].append(new_prj["_sample_dict"][0])
+
+ with pytest.raises(ProjectDuplicatedSampleGUIDsError):
+ agent.project.update(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ update_dict={"project": peppy.Project.from_dict(new_prj)},
+ )
diff --git a/tests/test_views.py b/tests/test_views.py
index f7b0590..159efda 100644
--- a/tests/test_views.py
+++ b/tests/test_views.py
@@ -1,10 +1,5 @@
-import os
-import warnings
-
import pytest
-from sqlalchemy.exc import OperationalError
-import pepdbagent
from pepdbagent.exceptions import (
SampleAlreadyInView,
SampleNotFoundError,
@@ -12,35 +7,11 @@
ViewNotFoundError,
)
-from .conftest import DNS
-
-DATA_PATH = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "tests",
- "data",
-)
-
-
-def get_path_to_example_file(namespace, project_name):
- return os.path.join(DATA_PATH, namespace, project_name, "project_config.yaml")
-
-
-def db_setup():
- # Check if the database is setup
- try:
- pepdbagent.PEPDatabaseAgent(dsn=DNS)
- except OperationalError:
- warnings.warn(
- UserWarning(
- f"Skipping tests, because DB is not setup. {DNS}. To setup DB go to README.md"
- )
- )
- return False
- return True
+from .utils import PEPDBAgentContextManager
@pytest.mark.skipif(
- not db_setup(),
+ not PEPDBAgentContextManager().db_setup(),
reason="DB is not setup",
)
class TestViews:
@@ -54,23 +25,24 @@ class TestViews:
["namespace1", "amendments1", "pig_0h", "view1"],
],
)
- def test_create_view(self, initiate_pepdb_con, namespace, name, sample_name, view_name):
- initiate_pepdb_con.view.create(
- view_name,
- {
- "project_namespace": namespace,
- "project_name": name,
- "project_tag": "default",
- "sample_list": [sample_name, "pig_1h"],
- },
- )
+ def test_create_view(self, namespace, name, sample_name, view_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.view.create(
+ view_name,
+ {
+ "project_namespace": namespace,
+ "project_name": name,
+ "project_tag": "default",
+ "sample_list": [sample_name, "pig_1h"],
+ },
+ )
- project = initiate_pepdb_con.project.get(namespace, name, raw=False)
- view_project = initiate_pepdb_con.view.get(
- namespace, name, "default", view_name, raw=False
- )
- assert len(view_project.samples) == 2
- assert view_project != project
+ project = agent.project.get(namespace, name, raw=False)
+ view_project = agent.view.get(
+ namespace, name, "default", view_name, raw=False
+ )
+ assert len(view_project.samples) == 2
+ assert view_project != project
@pytest.mark.parametrize(
"namespace, name, sample_name, view_name",
@@ -79,18 +51,19 @@ def test_create_view(self, initiate_pepdb_con, namespace, name, sample_name, vie
],
)
def test_create_view_with_incorrect_sample(
- self, initiate_pepdb_con, namespace, name, sample_name, view_name
+ self, namespace, name, sample_name, view_name
):
- with pytest.raises(SampleNotFoundError):
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": "namespace1",
- "project_name": "amendments1",
- "project_tag": "default",
- "sample_list": ["pig_0h", "pig_1h", "pig_2h"],
- },
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ with pytest.raises(SampleNotFoundError):
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": "namespace1",
+ "project_name": "amendments1",
+ "project_tag": "default",
+ "sample_list": ["pig_0h", "pig_1h", "pig_2h"],
+ },
+ )
@pytest.mark.parametrize(
"namespace, name, sample_name, view_name",
@@ -99,24 +72,25 @@ def test_create_view_with_incorrect_sample(
],
)
def test_create_view_with_incorrect_sample_no_fail(
- self, initiate_pepdb_con, namespace, name, sample_name, view_name
+ self, namespace, name, sample_name, view_name
):
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": "namespace1",
- "project_name": "amendments1",
- "project_tag": "default",
- "sample_list": ["pig_0h", "pig_1h", "pig_2h"],
- },
- no_fail=True,
- )
- project = initiate_pepdb_con.project.get(namespace, name, raw=False)
- view_project = initiate_pepdb_con.view.get(
- namespace, name, "default", view_name, raw=False
- )
- assert len(view_project.samples) == 2
- assert view_project != project
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": "namespace1",
+ "project_name": "amendments1",
+ "project_tag": "default",
+ "sample_list": ["pig_0h", "pig_1h", "pig_2h"],
+ },
+ no_fail=True,
+ )
+ project = agent.project.get(namespace, name, raw=False)
+ view_project = agent.view.get(
+ namespace, name, "default", view_name, raw=False
+ )
+ assert len(view_project.samples) == 2
+ assert view_project != project
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -124,26 +98,27 @@ def test_create_view_with_incorrect_sample_no_fail(
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_delete_view(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": namespace,
- "project_name": name,
- "project_tag": "default",
- "sample_list": [sample_name, "pig_1h"],
- },
- )
- assert (
- len(
- initiate_pepdb_con.view.get(namespace, name, "default", "view1", raw=False).samples
+ def test_delete_view(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": namespace,
+ "project_name": name,
+ "project_tag": "default",
+ "sample_list": [sample_name, "pig_1h"],
+ },
)
- == 2
- )
- initiate_pepdb_con.view.delete(namespace, name, "default", "view1")
- with pytest.raises(ViewNotFoundError):
- initiate_pepdb_con.view.get(namespace, name, "default", "view1", raw=False)
- assert len(initiate_pepdb_con.project.get(namespace, name, raw=False).samples) == 4
+ assert (
+ len(
+ agent.view.get(namespace, name, "default", "view1", raw=False).samples
+ )
+ == 2
+ )
+ agent.view.delete(namespace, name, "default", "view1")
+ with pytest.raises(ViewNotFoundError):
+ agent.view.get(namespace, name, "default", "view1", raw=False)
+ assert len(agent.project.get(namespace, name, raw=False).samples) == 4
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -151,23 +126,24 @@ def test_delete_view(self, initiate_pepdb_con, namespace, name, sample_name):
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_add_sample_to_view(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": namespace,
- "project_name": name,
- "project_tag": "default",
- "sample_list": [sample_name],
- },
- )
- initiate_pepdb_con.view.add_sample(namespace, name, "default", "view1", "pig_1h")
- assert (
- len(
- initiate_pepdb_con.view.get(namespace, name, "default", "view1", raw=False).samples
+ def test_add_sample_to_view(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": namespace,
+ "project_name": name,
+ "project_tag": "default",
+ "sample_list": [sample_name],
+ },
+ )
+ agent.view.add_sample(namespace, name, "default", "view1", "pig_1h")
+ assert (
+ len(
+ agent.view.get(namespace, name, "default", "view1", raw=False).samples
+ )
+ == 2
)
- == 2
- )
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -175,25 +151,26 @@ def test_add_sample_to_view(self, initiate_pepdb_con, namespace, name, sample_na
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_add_multiple_samples_to_view(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": namespace,
- "project_name": name,
- "project_tag": "default",
- "sample_list": [sample_name],
- },
- )
- initiate_pepdb_con.view.add_sample(
- namespace, name, "default", "view1", ["pig_1h", "frog_0h"]
- )
- assert (
- len(
- initiate_pepdb_con.view.get(namespace, name, "default", "view1", raw=False).samples
+ def test_add_multiple_samples_to_view(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": namespace,
+ "project_name": name,
+ "project_tag": "default",
+ "sample_list": [sample_name],
+ },
+ )
+ agent.view.add_sample(
+ namespace, name, "default", "view1", ["pig_1h", "frog_0h"]
+ )
+ assert (
+ len(
+ agent.view.get(namespace, name, "default", "view1", raw=False).samples
+ )
+ == 3
)
- == 3
- )
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -201,27 +178,28 @@ def test_add_multiple_samples_to_view(self, initiate_pepdb_con, namespace, name,
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_remove_sample_from_view(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": namespace,
- "project_name": name,
- "project_tag": "default",
- "sample_list": [sample_name, "pig_1h"],
- },
- )
- initiate_pepdb_con.view.remove_sample(namespace, name, "default", "view1", sample_name)
- assert (
- len(
- initiate_pepdb_con.view.get(namespace, name, "default", "view1", raw=False).samples
+ def test_remove_sample_from_view(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": namespace,
+ "project_name": name,
+ "project_tag": "default",
+ "sample_list": [sample_name, "pig_1h"],
+ },
)
- == 1
- )
- assert len(initiate_pepdb_con.project.get(namespace, name, raw=False).samples) == 4
+ agent.view.remove_sample(namespace, name, "default", "view1", sample_name)
+ assert (
+ len(
+ agent.view.get(namespace, name, "default", "view1", raw=False).samples
+ )
+ == 1
+ )
+ assert len(agent.project.get(namespace, name, raw=False).samples) == 4
- with pytest.raises(SampleNotInViewError):
- initiate_pepdb_con.view.remove_sample(namespace, name, "default", "view1", sample_name)
+ with pytest.raises(SampleNotInViewError):
+ agent.view.remove_sample(namespace, name, "default", "view1", sample_name)
@pytest.mark.parametrize(
"namespace, name, sample_name",
@@ -229,18 +207,19 @@ def test_remove_sample_from_view(self, initiate_pepdb_con, namespace, name, samp
["namespace1", "amendments1", "pig_0h"],
],
)
- def test_add_existing_sample_in_view(self, initiate_pepdb_con, namespace, name, sample_name):
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": namespace,
- "project_name": name,
- "project_tag": "default",
- "sample_list": [sample_name, "pig_1h"],
- },
- )
- with pytest.raises(SampleAlreadyInView):
- initiate_pepdb_con.view.add_sample(namespace, name, "default", "view1", sample_name)
+ def test_add_existing_sample_in_view(self, namespace, name, sample_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": namespace,
+ "project_name": name,
+ "project_tag": "default",
+ "sample_list": [sample_name, "pig_1h"],
+ },
+ )
+ with pytest.raises(SampleAlreadyInView):
+ agent.view.add_sample(namespace, name, "default", "view1", sample_name)
@pytest.mark.parametrize(
"namespace, name, sample_name, view_name",
@@ -248,15 +227,16 @@ def test_add_existing_sample_in_view(self, initiate_pepdb_con, namespace, name,
["namespace1", "amendments1", "pig_0h", "view1"],
],
)
- def test_get_snap_view(self, initiate_pepdb_con, namespace, name, sample_name, view_name):
- snap_project = initiate_pepdb_con.view.get_snap_view(
- namespace=namespace,
- name=name,
- tag="default",
- sample_name_list=[sample_name, "pig_1h"],
- )
+ def test_get_snap_view(self, namespace, name, sample_name, view_name):
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ snap_project = agent.view.get_snap_view(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ sample_name_list=[sample_name, "pig_1h"],
+ )
- assert len(snap_project.samples) == 2
+ assert len(snap_project.samples) == 2
@pytest.mark.parametrize(
"namespace, name, sample_name, view_name",
@@ -265,22 +245,23 @@ def test_get_snap_view(self, initiate_pepdb_con, namespace, name, sample_name, v
],
)
def test_get_view_list_from_project(
- self, initiate_pepdb_con, namespace, name, sample_name, view_name
+ self, namespace, name, sample_name, view_name
):
- assert (
- len(initiate_pepdb_con.view.get_views_annotation(namespace, name, "default").views)
- == 0
- )
- initiate_pepdb_con.view.create(
- "view1",
- {
- "project_namespace": namespace,
- "project_name": name,
- "project_tag": "default",
- "sample_list": [sample_name, "pig_1h"],
- },
- )
- assert (
- len(initiate_pepdb_con.view.get_views_annotation(namespace, name, "default").views)
- == 1
- )
+ with PEPDBAgentContextManager(add_data=True) as agent:
+ assert (
+ len(agent.view.get_views_annotation(namespace, name, "default").views)
+ == 0
+ )
+ agent.view.create(
+ "view1",
+ {
+ "project_namespace": namespace,
+ "project_name": name,
+ "project_tag": "default",
+ "sample_list": [sample_name, "pig_1h"],
+ },
+ )
+ assert (
+ len(agent.view.get_views_annotation(namespace, name, "default").views)
+ == 1
+ )
diff --git a/tests/utils.py b/tests/utils.py
new file mode 100644
index 0000000..0ee217f
--- /dev/null
+++ b/tests/utils.py
@@ -0,0 +1,98 @@
+import os
+import peppy
+import warnings
+from sqlalchemy.exc import OperationalError
+from typing import Union
+
+from pepdbagent.db_utils import BaseEngine
+from pepdbagent import PEPDatabaseAgent
+
+DSN = "postgresql+psycopg://postgres:docker@localhost:5432/pep-db"
+
+DATA_PATH = os.path.join(
+ os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+ "tests",
+ "data",
+)
+
+
+def get_path_to_example_file(namespace: str, project_name: str) -> str:
+ """
+ Get path to example file
+ """
+ return os.path.join(DATA_PATH, namespace, project_name, "project_config.yaml")
+
+
+def list_of_available_peps() -> dict:
+ pep_namespaces = os.listdir(DATA_PATH)
+ projects = {}
+ for np in pep_namespaces:
+ pep_name = os.listdir(os.path.join(DATA_PATH, np))
+ projects[np] = {p: get_path_to_example_file(np, p) for p in pep_name}
+ return projects
+
+
+class PEPDBAgentContextManager:
+ """
+ Class with context manager to connect to database. Adds data and drops everything from the database upon exit to ensure.
+ """
+
+ def __init__(self, url: str = DSN, add_data: bool = False):
+ """
+ :param url: database url e.g. "postgresql+psycopg://postgres:docker@localhost:5432/pep-db"
+ :param add_data: add data to the database
+ """
+
+ self.url = url
+ self._agent = None
+ self.add_data = add_data
+
+ def __enter__(self):
+ self._agent = PEPDatabaseAgent(dsn=self.url, echo=False)
+ self.db_engine = self._agent.pep_db_engine
+ self.db_engine.create_schema()
+ if self.add_data:
+ self._insert_data()
+ return self._agent
+
+ def __exit__(self, exc_type, exc_value, exc_traceback):
+ self.db_engine.delete_schema()
+
+ def _insert_data(self):
+ pepdb_con = PEPDatabaseAgent(dsn=self.url, echo=True)
+ for namespace, item in list_of_available_peps().items():
+ if namespace == "private_test":
+ private = True
+ else:
+ private = False
+ for name, path in item.items():
+ prj = peppy.Project(path)
+ pepdb_con.project.create(
+ namespace=namespace,
+ name=name,
+ tag="default",
+ is_private=private,
+ project=prj,
+ overwrite=True,
+ pep_schema="random_schema_name",
+ )
+
+ @property
+ def agent(self) -> PEPDatabaseAgent:
+ return self._agent
+
+ def db_setup(self):
+ # Check if the database is setup
+ try:
+ PEPDatabaseAgent(dsn=self.url)
+ except OperationalError:
+ warnings.warn(
+ UserWarning(
+ f"Skipping tests, because DB is not setup. {self.url}. To setup DB go to README.md"
+ )
+ )
+ return False
+ return True
+
+
+