fixed test context manager

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

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

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