diff --git a/pepdbagent/db_utils.py b/pepdbagent/db_utils.py index fd5c7cd..c9f93b3 100644 --- a/pepdbagent/db_utils.py +++ b/pepdbagent/db_utils.py @@ -1,32 +1,26 @@ import datetime import logging -from typing import Optional, List +from typing import List, Optional from sqlalchemy import ( + TIMESTAMP, BigInteger, FetchedValue, + ForeignKey, Result, Select, String, + UniqueConstraint, event, select, - TIMESTAMP, - ForeignKey, - UniqueConstraint, ) from sqlalchemy.dialects.postgresql import JSON from sqlalchemy.engine import URL, create_engine from sqlalchemy.exc import ProgrammingError from sqlalchemy.ext.compiler import compiles -from sqlalchemy.orm import ( - DeclarativeBase, - Mapped, - Session, - mapped_column, - relationship, -) +from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship -from pepdbagent.const import POSTGRES_DIALECT, PKG_NAME +from pepdbagent.const import PKG_NAME, POSTGRES_DIALECT from pepdbagent.exceptions import SchemaError _LOGGER = logging.getLogger(PKG_NAME) diff --git a/pepdbagent/models.py b/pepdbagent/models.py index a0b2c45..52b8fa6 100644 --- a/pepdbagent/models.py +++ b/pepdbagent/models.py @@ -1,7 +1,8 @@ # file with pydantic models -from typing import List, Optional, Union, Dict -from pydantic import BaseModel, Field, ConfigDict, field_validator -from peppy.const import CONFIG_KEY, SUBSAMPLE_RAW_LIST_KEY, SAMPLE_RAW_DICT_KEY +from typing import Dict, List, Optional, Union + +from peppy.const import CONFIG_KEY, SAMPLE_RAW_DICT_KEY, SUBSAMPLE_RAW_LIST_KEY +from pydantic import BaseModel, ConfigDict, Field, field_validator from pepdbagent.const import DEFAULT_TAG diff --git a/pepdbagent/modules/annotation.py b/pepdbagent/modules/annotation.py index bd9f511..2b30ccb 100644 --- a/pepdbagent/modules/annotation.py +++ b/pepdbagent/modules/annotation.py @@ -3,16 +3,16 @@ from typing import List, Literal, Optional, Union from sqlalchemy import and_, func, or_, select -from sqlalchemy.sql.selectable import Select from sqlalchemy.orm import Session +from sqlalchemy.sql.selectable import Select from pepdbagent.const import ( DEFAULT_LIMIT, DEFAULT_OFFSET, DEFAULT_TAG, + LAST_UPDATE_DATE_KEY, PKG_NAME, SUBMISSION_DATE_KEY, - LAST_UPDATE_DATE_KEY, ) from pepdbagent.db_utils import BaseEngine, Projects from pepdbagent.exceptions import FilterError, ProjectNotFoundError, RegistryPathError diff --git a/pepdbagent/modules/namespace.py b/pepdbagent/modules/namespace.py index 5af92db..1abf470 100644 --- a/pepdbagent/modules/namespace.py +++ b/pepdbagent/modules/namespace.py @@ -1,20 +1,20 @@ import logging -from typing import List, Union, Tuple from collections import Counter from datetime import datetime, timedelta +from typing import List, Tuple, Union from sqlalchemy import distinct, func, or_, select, text -from sqlalchemy.sql.selectable import Select from sqlalchemy.orm import Session +from sqlalchemy.sql.selectable import Select -from pepdbagent.const import DEFAULT_LIMIT, DEFAULT_OFFSET, PKG_NAME, DEFAULT_LIMIT_INFO +from pepdbagent.const import DEFAULT_LIMIT, DEFAULT_LIMIT_INFO, DEFAULT_OFFSET, PKG_NAME +from pepdbagent.db_utils import BaseEngine, Projects from pepdbagent.exceptions import NamespaceNotFoundError -from pepdbagent.db_utils import Projects, BaseEngine from pepdbagent.models import ( + ListOfNamespaceInfo, Namespace, - NamespaceList, NamespaceInfo, - ListOfNamespaceInfo, + NamespaceList, NamespaceStats, ) from pepdbagent.utils import tuple_converter diff --git a/pepdbagent/modules/project.py b/pepdbagent/modules/project.py index b7fe98c..b4e658d 100644 --- a/pepdbagent/modules/project.py +++ b/pepdbagent/modules/project.py @@ -1,43 +1,33 @@ import datetime import json import logging -from typing import Union, List, NoReturn, Dict +from typing import Dict, List, NoReturn, Union -import peppy -from sqlalchemy import and_, delete, select -from sqlalchemy.exc import IntegrityError, NoResultFound -from sqlalchemy.orm import Session -from sqlalchemy import Select -from sqlalchemy.orm.attributes import flag_modified import numpy as np - +import peppy from peppy.const import ( - SAMPLE_RAW_DICT_KEY, - SUBSAMPLE_RAW_LIST_KEY, CONFIG_KEY, - SAMPLE_TABLE_INDEX_KEY, SAMPLE_NAME_ATTR, + SAMPLE_RAW_DICT_KEY, + SAMPLE_TABLE_INDEX_KEY, + SUBSAMPLE_RAW_LIST_KEY, ) +from sqlalchemy import Select, and_, delete, select +from sqlalchemy.exc import IntegrityError, NoResultFound +from sqlalchemy.orm import Session +from sqlalchemy.orm.attributes import flag_modified -from pepdbagent.const import ( - DEFAULT_TAG, - DESCRIPTION_KEY, - NAME_KEY, - PKG_NAME, - PEPHUB_SAMPLE_ID_KEY, -) - -from pepdbagent.db_utils import Projects, Samples, Subsamples, BaseEngine +from pepdbagent.const import DEFAULT_TAG, DESCRIPTION_KEY, NAME_KEY, PEPHUB_SAMPLE_ID_KEY, PKG_NAME +from pepdbagent.db_utils import BaseEngine, Projects, Samples, Subsamples from pepdbagent.exceptions import ( + PEPDatabaseAgentError, + ProjectDuplicatedSampleGUIDsError, ProjectNotFoundError, ProjectUniqueNameError, - PEPDatabaseAgentError, SampleTableUpdateError, - ProjectDuplicatedSampleGUIDsError, ) -from pepdbagent.models import UpdateItems, UpdateModel, ProjectDict -from pepdbagent.utils import create_digest, registry_path_converter, order_samples, generate_guid - +from pepdbagent.models import ProjectDict, UpdateItems, UpdateModel +from pepdbagent.utils import create_digest, generate_guid, order_samples, registry_path_converter _LOGGER = logging.getLogger(PKG_NAME) diff --git a/pepdbagent/modules/sample.py b/pepdbagent/modules/sample.py index c1df859..7e8d89a 100644 --- a/pepdbagent/modules/sample.py +++ b/pepdbagent/modules/sample.py @@ -1,20 +1,16 @@ +import datetime import logging from typing import Union -import datetime import peppy from peppy.const import SAMPLE_TABLE_INDEX_KEY -from sqlalchemy import select, and_ +from sqlalchemy import and_, select from sqlalchemy.orm import Session from sqlalchemy.orm.attributes import flag_modified -from pepdbagent.const import ( - DEFAULT_TAG, - PKG_NAME, -) -from pepdbagent.exceptions import SampleNotFoundError, SampleAlreadyExistsError - -from pepdbagent.db_utils import BaseEngine, Samples, Projects +from pepdbagent.const import DEFAULT_TAG, PKG_NAME +from pepdbagent.db_utils import BaseEngine, Projects, Samples +from pepdbagent.exceptions import SampleAlreadyExistsError, SampleNotFoundError from pepdbagent.utils import generate_guid, order_samples _LOGGER = logging.getLogger(PKG_NAME) diff --git a/pepdbagent/modules/user.py b/pepdbagent/modules/user.py index 7ae4705..eea8f84 100644 --- a/pepdbagent/modules/user.py +++ b/pepdbagent/modules/user.py @@ -2,17 +2,13 @@ from typing import Union from sqlalchemy import and_, delete, select -from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session -from pepdbagent.const import ( - PKG_NAME, -) - -from pepdbagent.db_utils import BaseEngine, User, Stars, Projects +from pepdbagent.const import PKG_NAME +from pepdbagent.db_utils import BaseEngine, Projects, Stars, User +from pepdbagent.exceptions import ProjectAlreadyInFavorites, ProjectNotInFavorites from pepdbagent.models import AnnotationList, AnnotationModel -from pepdbagent.exceptions import ProjectNotInFavorites, ProjectAlreadyInFavorites - _LOGGER = logging.getLogger(PKG_NAME) diff --git a/pepdbagent/modules/view.py b/pepdbagent/modules/view.py index 01c7ed3..8704c97 100644 --- a/pepdbagent/modules/view.py +++ b/pepdbagent/modules/view.py @@ -1,29 +1,24 @@ # View of the PEP. In other words, it is a part of the PEP, or subset of the samples in the PEP. import logging -from typing import Union, List +from typing import List, Union import peppy -from sqlalchemy import select, and_, delete -from sqlalchemy.orm import Session +from sqlalchemy import and_, delete, select from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session - -from pepdbagent.const import ( - DEFAULT_TAG, - PKG_NAME, -) +from pepdbagent.const import DEFAULT_TAG, PKG_NAME +from pepdbagent.db_utils import BaseEngine, Projects, Samples, Views, ViewSampleAssociation from pepdbagent.exceptions import ( - ViewNotFoundError, - SampleAlreadyInView, ProjectNotFoundError, + SampleAlreadyInView, SampleNotFoundError, - ViewAlreadyExistsError, SampleNotInViewError, + ViewAlreadyExistsError, + ViewNotFoundError, ) - -from pepdbagent.db_utils import BaseEngine, Samples, Projects, Views, ViewSampleAssociation -from pepdbagent.models import ViewAnnotation, CreateViewDictModel, ProjectViews +from pepdbagent.models import CreateViewDictModel, ProjectViews, ViewAnnotation _LOGGER = logging.getLogger(PKG_NAME) diff --git a/pepdbagent/pepdbagent.py b/pepdbagent/pepdbagent.py index 19995e5..fc1a386 100644 --- a/pepdbagent/pepdbagent.py +++ b/pepdbagent/pepdbagent.py @@ -3,8 +3,8 @@ from pepdbagent.modules.annotation import PEPDatabaseAnnotation from pepdbagent.modules.namespace import PEPDatabaseNamespace from pepdbagent.modules.project import PEPDatabaseProject -from pepdbagent.modules.user import PEPDatabaseUser from pepdbagent.modules.sample import PEPDatabaseSample +from pepdbagent.modules.user import PEPDatabaseUser from pepdbagent.modules.view import PEPDatabaseView diff --git a/pepdbagent/utils.py b/pepdbagent/utils.py index 4bb7103..7f86ca4 100644 --- a/pepdbagent/utils.py +++ b/pepdbagent/utils.py @@ -1,9 +1,9 @@ import datetime import json +import uuid from collections.abc import Iterable from hashlib import md5 -from typing import Tuple, Union, List -import uuid +from typing import List, Tuple, Union import ubiquerg from peppy.const import SAMPLE_RAW_DICT_KEY diff --git a/tests/conftest.py b/tests/conftest.py index 32d3dbe..19ec436 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,9 +1,8 @@ -import peppy -import pytest import os -from sqlalchemy import create_engine -from sqlalchemy import text +import peppy +import pytest +from sqlalchemy import create_engine, text from pepdbagent import PEPDatabaseAgent diff --git a/tests/test_annotation.py b/tests/test_annotation.py new file mode 100644 index 0000000..f70addb --- /dev/null +++ b/tests/test_annotation.py @@ -0,0 +1,340 @@ +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 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +class TestAnnotation: + """ + Test function within annotation class + """ + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ["namespace2", "derive"], + ["namespace2", "imply"], + ["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 + + @pytest.mark.parametrize( + "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", + ) + + @pytest.mark.parametrize( + "namespace, n_projects", + [ + ["namespace1", 6], + ["namespace2", 8], + ["namespace3", 10], + ["private", 0], + ["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 + + @pytest.mark.parametrize( + "namespace, n_projects", + [ + ["namespace1", 6], + ["namespace2", 8], + ["namespace3", 10], + ["private", 0], + ["private_test", 6], + ], + ) + @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 + + @pytest.mark.parametrize( + "namespace, limit, n_projects", + [ + ["namespace1", 3, 6], + ["namespace2", 2, 8], + ["namespace3", 8, 10], + ["private", 0, 0], + ["private_test", 5, 6], + ], + ) + @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 + + @pytest.mark.parametrize( + "namespace, order_by, first_name", + [ + ["namespace1", "name", "amendments1"], + ["namespace2", "name", "biocproject_exceptions"], + ["namespace3", "name", "node_alias"], + ["private_test", "name", "amendments1"], + ], + ) + @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 + + @pytest.mark.parametrize( + "namespace, order_by, last_name", + [ + ["namespace1", "name", "biocproject"], + ["namespace2", "name", "imports"], + ["namespace3", "name", "subtables"], + ["private_test", "name", "subtable3"], + ], + ) + @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 + + @pytest.mark.parametrize( + "namespace, query, found_number", + [ + ["namespace1", "ame", 2], + ["namespace2", "proj", 2], + ["namespace3", "ABLE", 6], + ["private_test", "a", 0], + [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 + + @pytest.mark.parametrize( + "namespace, query, found_number", + [ + ["namespace1", "ame", 2], + ["namespace2", "proj", 2], + ["namespace3", "ABLE", 6], + ["private_test", "b", 2], + [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 + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ["namespace2", "derive"], + ["namespace2", "imply"], + ["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", + } + + @pytest.mark.parametrize( + "namespace, query, found_number", + [ + ["namespace1", "ame", 2], + [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 + + @pytest.mark.parametrize( + "namespace, query, found_number", + [ + ["namespace1", "ame", 0], + [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 + + @pytest.mark.parametrize( + "namespace, query, found_number", + [ + ["namespace1", "ame", 2], + ], + ) + def test_search_incorrect_filter_by_string( + 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) + 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"), + ) + + @pytest.mark.parametrize( + "rp_list, admin, found_number", + [ + [ + [ + "namespace1/amendments1:default", + "namespace1/amendments2:default", + "namespace2/derive:default", + "private_test/amendments1:default", + ], + "namespace1", + 4, + ], + [ + [ + "namespace1/amendments1:default", + "namespace1/amendments2:default", + "namespace2/derive:default", + "private_test/amendments1:default", + ], + "private_test", + 4, + ], + ], + ) + 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_enpty_list(self, initiate_pepdb_con): + result = initiate_pepdb_con.annotation.get_by_rp_list([]) + assert len(result.results) == 0 + + @pytest.mark.parametrize( + "namespace, query, found_number", + [ + ["namespace1", "ame", 2], + ], + ) + def test_search_incorrect_incorrect_pep_type( + self, initiate_pepdb_con, namespace, query, found_number + ): + with pytest.raises(ValueError): + initiate_pepdb_con.annotation.get(namespace=namespace, pep_type="incorrect") + + @pytest.mark.parametrize( + "namespace, query, found_number", + [ + ["namespace1", "ame", 2], + ], + ) + def test_project_list_without_annotation( + self, initiate_pepdb_con, namespace, query, found_number + ): + result = initiate_pepdb_con.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 new file mode 100644 index 0000000..aa49889 --- /dev/null +++ b/tests/test_namespace.py @@ -0,0 +1,165 @@ +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 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +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_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 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +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" + ) + fav_results = initiate_pepdb_con.user.get_favorites("random_namespace") + + 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 + + @pytest.mark.parametrize( + "namespace, name", + [ + ["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 + + @pytest.mark.parametrize( + "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 + + @pytest.mark.parametrize( + "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" + ) + + @pytest.mark.parametrize( + "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" + ) + + @pytest.mark.parametrize( + "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 diff --git a/tests/test_pepagent.py b/tests/test_pepagent.py deleted file mode 100644 index 6c966b3..0000000 --- a/tests/test_pepagent.py +++ /dev/null @@ -1,1542 +0,0 @@ -import datetime -import os -import warnings - -import peppy -import pytest -from sqlalchemy.exc import OperationalError -import numpy as np - -import pepdbagent -from pepdbagent.exceptions import ( - FilterError, - ProjectNotFoundError, - ProjectNotInFavorites, - ProjectAlreadyInFavorites, - SampleNotFoundError, - ViewNotFoundError, - SampleAlreadyInView, - SampleNotInViewError, - SampleTableUpdateError, - ProjectDuplicatedSampleGUIDsError, -) -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 - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -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 - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ["namespace1", "basic"], - ["namespace2", "derive"], - ["namespace2", "imply"], - ["namespace3", "piface"], - ["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 - - @pytest.mark.parametrize( - "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 - - @pytest.mark.parametrize( - "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)) - - assert ( - prj_subtables - == orgiginal_prj.to_dict(extended=True, orient="records")["_subsample_list"] - ) - - @pytest.mark.parametrize( - "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)) - - assert ( - prj_samples == orgiginal_prj.to_dict(extended=True, orient="records")["_sample_dict"] - ) - - @pytest.mark.parametrize( - "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)) - - assert prj_samples == orgiginal_prj.sample_table.replace({np.nan: None}).to_dict( - orient="records" - ) - - @pytest.mark.parametrize( - "namespace, name,tag", - [ - ["incorrect_namespace", "amendments1", "default"], - ["namespace1", "subtable2", "default"], - ["namespace3", "basic", "default"], - ["namespace3", "subtable2", "incorrect_tag"], - ["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) - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ["namespace2", "derive"], - ["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, - ) - - assert initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False) == new_prj - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ["namespace2", "derive"], - ["namespace2", "imply"], - ], - ) - def test_delete_project(self, initiate_pepdb_con, namespace, name): - initiate_pepdb_con.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") - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ["namespace2", "derive"], - ["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", - ) - - 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" - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ], - ) - def test_parent_project_delete(self, initiate_pepdb_con, 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", - ) - - 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" - ) - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ], - ) - def test_child_project_delete(self, initiate_pepdb_con, 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", - ) - - 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") - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ], - ) - def test_project_can_be_forked_twice(self, initiate_pepdb_con, 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", - ) - - 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 = initiate_pepdb_con.annotation.get( - namespace="new_namespace2", name="new_name2", tag="new_tag2" - ) - assert result.results[0].forked_from == f"{namespace}/{name}:default" - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -class TestProjectUpdate: - @pytest.mark.parametrize( - "namespace, name,new_name", - [ - ["namespace1", "amendments1", "name1"], - ["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") - - @pytest.mark.parametrize( - "namespace, name,new_name", - [ - ["namespace1", "amendments1", "name1"], - ["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") - - @pytest.mark.parametrize( - "namespace, name, new_tag", - [ - ["namespace1", "amendments1", "tag1"], - ["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) - - @pytest.mark.parametrize( - "namespace, name, new_description", - [ - ["namespace1", "amendments1", "desc1 f"], - ["namespace2", "derive", "desc5 f"], - ], - ) - def test_update_project_description( - self, initiate_pepdb_con, 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}, - ) - - assert ( - initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False).description - == new_description - ) - - @pytest.mark.parametrize( - "namespace, name, new_description", - [ - ["namespace1", "amendments1", "desc1 f"], - ], - ) - def test_update_project_description_in_config( - self, initiate_pepdb_con, 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}, - ) - - assert ( - initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False).description - == new_description - ) - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["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}, - ) - - @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 - - @pytest.mark.parametrize( - "namespace, name", - [ - ["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}, - ) - - is_private = ( - initiate_pepdb_con.annotation.get( - namespace=namespace, name=name, tag="default", admin=[namespace] - ) - .results[0] - .is_private - ) - assert is_private is True - - @pytest.mark.parametrize( - "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}, - ) - - pop = ( - initiate_pepdb_con.annotation.get( - namespace=namespace, name=name, tag="default", admin=[namespace] - ) - .results[0] - .pop - ) - 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}, - ) - - pop = ( - initiate_pepdb_con.annotation.get( - namespace=namespace, name=name, tag="default", admin=[namespace] - ) - .results[0] - .pop - ) - assert pop is False - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "basic"], - ], - ) - def test_project_can_have_2_sample_names(self, initiate_pepdb_con, 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") - - 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)}, - ) - - prj = initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=True) - - assert len(prj["_sample_dict"]) == 4 - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -class TestUpdateProjectWithId: - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace3", "subtable1"], - ], - ) - def test_update_whole_project_with_id(self, initiate_pepdb_con, namespace, name): - pass - - # TODO: write more tests - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - # ["namespace3", "subtable1"], - ], - ) - def test_update_project_with_duplicated_sample_guids( - self, initiate_pepdb_con, 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)}, - ) - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -class TestAnnotation: - """ - Test function within annotation class - """ - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ["namespace2", "derive"], - ["namespace2", "imply"], - ["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 - - @pytest.mark.parametrize( - "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", - ) - - @pytest.mark.parametrize( - "namespace, n_projects", - [ - ["namespace1", 6], - ["namespace2", 8], - ["namespace3", 10], - ["private", 0], - ["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 - - @pytest.mark.parametrize( - "namespace, n_projects", - [ - ["namespace1", 6], - ["namespace2", 8], - ["namespace3", 10], - ["private", 0], - ["private_test", 6], - ], - ) - @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 - - @pytest.mark.parametrize( - "namespace, limit, n_projects", - [ - ["namespace1", 3, 6], - ["namespace2", 2, 8], - ["namespace3", 8, 10], - ["private", 0, 0], - ["private_test", 5, 6], - ], - ) - @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 - - @pytest.mark.parametrize( - "namespace, order_by, first_name", - [ - ["namespace1", "name", "amendments1"], - ["namespace2", "name", "biocproject_exceptions"], - ["namespace3", "name", "node_alias"], - ["private_test", "name", "amendments1"], - ], - ) - @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 - - @pytest.mark.parametrize( - "namespace, order_by, last_name", - [ - ["namespace1", "name", "biocproject"], - ["namespace2", "name", "imports"], - ["namespace3", "name", "subtables"], - ["private_test", "name", "subtable3"], - ], - ) - @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 - - @pytest.mark.parametrize( - "namespace, query, found_number", - [ - ["namespace1", "ame", 2], - ["namespace2", "proj", 2], - ["namespace3", "ABLE", 6], - ["private_test", "a", 0], - [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 - - @pytest.mark.parametrize( - "namespace, query, found_number", - [ - ["namespace1", "ame", 2], - ["namespace2", "proj", 2], - ["namespace3", "ABLE", 6], - ["private_test", "b", 2], - [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 - - @pytest.mark.parametrize( - "namespace, name", - [ - ["namespace1", "amendments1"], - ["namespace1", "amendments2"], - ["namespace2", "derive"], - ["namespace2", "imply"], - ["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", - } - - @pytest.mark.parametrize( - "namespace, query, found_number", - [ - ["namespace1", "ame", 2], - [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 - - @pytest.mark.parametrize( - "namespace, query, found_number", - [ - ["namespace1", "ame", 0], - [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 - - @pytest.mark.parametrize( - "namespace, query, found_number", - [ - ["namespace1", "ame", 2], - ], - ) - def test_search_incorrect_filter_by_string( - 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) - 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"), - ) - - @pytest.mark.parametrize( - "rp_list, admin, found_number", - [ - [ - [ - "namespace1/amendments1:default", - "namespace1/amendments2:default", - "namespace2/derive:default", - "private_test/amendments1:default", - ], - "namespace1", - 4, - ], - [ - [ - "namespace1/amendments1:default", - "namespace1/amendments2:default", - "namespace2/derive:default", - "private_test/amendments1:default", - ], - "private_test", - 4, - ], - ], - ) - 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_enpty_list(self, initiate_pepdb_con): - result = initiate_pepdb_con.annotation.get_by_rp_list([]) - assert len(result.results) == 0 - - @pytest.mark.parametrize( - "namespace, query, found_number", - [ - ["namespace1", "ame", 2], - ], - ) - def test_search_incorrect_incorrect_pep_type( - self, initiate_pepdb_con, namespace, query, found_number - ): - with pytest.raises(ValueError): - initiate_pepdb_con.annotation.get(namespace=namespace, pep_type="incorrect") - - @pytest.mark.parametrize( - "namespace, query, found_number", - [ - ["namespace1", "ame", 2], - ], - ) - def test_project_list_without_annotation( - self, initiate_pepdb_con, namespace, query, found_number - ): - result = initiate_pepdb_con.annotation.get_projects_list( - namespace=namespace, - search_str=query, - ) - assert len(result) == found_number - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -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_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 - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -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" - ) - fav_results = initiate_pepdb_con.user.get_favorites("random_namespace") - - 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 - - @pytest.mark.parametrize( - "namespace, name", - [ - ["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 - - @pytest.mark.parametrize( - "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 - - @pytest.mark.parametrize( - "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" - ) - - @pytest.mark.parametrize( - "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" - ) - - @pytest.mark.parametrize( - "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 - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -class TestSamples: - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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 - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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 - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["namespace2", "custom_index", "frog_1"], - ], - ) - def test_retrieve_sample_with_modified_sample_id( - 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_id == "frog_1" - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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" - - @pytest.mark.parametrize( - "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" - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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" - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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" - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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 - - @pytest.mark.parametrize( - "namespace, name, sample_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) - - initiate_pepdb_con.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 - ) - - @pytest.mark.parametrize( - "namespace, name, tag, sample_dict", - [ - [ - "namespace1", - "amendments1", - "default", - { - "sample_name": "new_sample", - "time": "new_time", - }, - ], - ], - ) - 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) - - prj2 = initiate_pepdb_con.project.get(namespace, name, raw=False) - - 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", - [ - [ - "namespace1", - "amendments1", - "default", - { - "sample_name": "pig_0h", - "time": "new_time", - }, - ], - ], - ) - 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" - ) - - @pytest.mark.parametrize( - "namespace, name, tag, sample_dict", - [ - [ - "namespace1", - "amendments1", - "default", - { - "sample_name": "new_sample", - "time": "new_time", - }, - ], - ], - ) - 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() - - -@pytest.mark.skipif( - not db_setup(), - reason="DB is not setup", -) -class TestViews: - """ - Test function within view class - """ - - @pytest.mark.parametrize( - "namespace, name, sample_name, view_name", - [ - ["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"], - }, - ) - - 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 - - @pytest.mark.parametrize( - "namespace, name, sample_name, view_name", - [ - ["namespace1", "amendments1", "pig_0h", "view1"], - ], - ) - def test_create_view_with_incorrect_sample( - self, initiate_pepdb_con, 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"], - }, - ) - - @pytest.mark.parametrize( - "namespace, name, sample_name, view_name", - [ - ["namespace1", "amendments1", "pig_0h", "view1"], - ], - ) - def test_create_view_with_incorrect_sample_no_fail( - self, initiate_pepdb_con, 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 - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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 - ) - == 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 - - @pytest.mark.parametrize( - "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 - ) - == 2 - ) - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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 - ) - == 3 - ) - - @pytest.mark.parametrize( - "namespace, name, sample_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 - ) - == 1 - ) - assert len(initiate_pepdb_con.project.get(namespace, name, raw=False).samples) == 4 - - with pytest.raises(SampleNotInViewError): - initiate_pepdb_con.view.remove_sample(namespace, name, "default", "view1", sample_name) - - @pytest.mark.parametrize( - "namespace, name, sample_name", - [ - ["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) - - @pytest.mark.parametrize( - "namespace, name, sample_name, view_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"], - ) - - assert len(snap_project.samples) == 2 - - @pytest.mark.parametrize( - "namespace, name, sample_name, view_name", - [ - ["namespace1", "amendments1", "pig_0h", "view1"], - ], - ) - def test_get_view_list_from_project( - self, initiate_pepdb_con, 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 - ) - - -class TestProjectSamplesUpdates: - ... - # TODO: write tests diff --git a/tests/test_project.py b/tests/test_project.py new file mode 100644 index 0000000..bc3c84e --- /dev/null +++ b/tests/test_project.py @@ -0,0 +1,327 @@ +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 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +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 + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ["namespace1", "basic"], + ["namespace2", "derive"], + ["namespace2", "imply"], + ["namespace3", "piface"], + ["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 + + @pytest.mark.parametrize( + "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 + + @pytest.mark.parametrize( + "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)) + + assert ( + prj_subtables + == orgiginal_prj.to_dict(extended=True, orient="records")["_subsample_list"] + ) + + @pytest.mark.parametrize( + "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)) + + assert ( + prj_samples == orgiginal_prj.to_dict(extended=True, orient="records")["_sample_dict"] + ) + + @pytest.mark.parametrize( + "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)) + + assert prj_samples == orgiginal_prj.sample_table.replace({np.nan: None}).to_dict( + orient="records" + ) + + @pytest.mark.parametrize( + "namespace, name,tag", + [ + ["incorrect_namespace", "amendments1", "default"], + ["namespace1", "subtable2", "default"], + ["namespace3", "basic", "default"], + ["namespace3", "subtable2", "incorrect_tag"], + ["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) + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ["namespace2", "derive"], + ["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, + ) + + assert initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False) == new_prj + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ["namespace2", "derive"], + ["namespace2", "imply"], + ], + ) + def test_delete_project(self, initiate_pepdb_con, namespace, name): + initiate_pepdb_con.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") + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ["namespace2", "derive"], + ["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", + ) + + 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" + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ], + ) + def test_parent_project_delete(self, initiate_pepdb_con, 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", + ) + + 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" + ) + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ], + ) + def test_child_project_delete(self, initiate_pepdb_con, 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", + ) + + 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") + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace1", "amendments2"], + ], + ) + def test_project_can_be_forked_twice(self, initiate_pepdb_con, 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", + ) + + 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 = initiate_pepdb_con.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 new file mode 100644 index 0000000..99c73b6 --- /dev/null +++ b/tests/test_samples.py @@ -0,0 +1,251 @@ +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 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +class TestSamples: + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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 + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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 + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["namespace2", "custom_index", "frog_1"], + ], + ) + def test_retrieve_sample_with_modified_sample_id( + 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_id == "frog_1" + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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" + + @pytest.mark.parametrize( + "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" + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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" + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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" + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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 + + @pytest.mark.parametrize( + "namespace, name, sample_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) + + initiate_pepdb_con.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 + ) + + @pytest.mark.parametrize( + "namespace, name, tag, sample_dict", + [ + [ + "namespace1", + "amendments1", + "default", + { + "sample_name": "new_sample", + "time": "new_time", + }, + ], + ], + ) + 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) + + prj2 = initiate_pepdb_con.project.get(namespace, name, raw=False) + + 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", + [ + [ + "namespace1", + "amendments1", + "default", + { + "sample_name": "pig_0h", + "time": "new_time", + }, + ], + ], + ) + 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" + ) + + @pytest.mark.parametrize( + "namespace, name, tag, sample_dict", + [ + [ + "namespace1", + "amendments1", + "default", + { + "sample_name": "new_sample", + "time": "new_time", + }, + ], + ], + ) + 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() diff --git a/tests/test_updates.py b/tests/test_updates.py new file mode 100644 index 0000000..6fb4e02 --- /dev/null +++ b/tests/test_updates.py @@ -0,0 +1,324 @@ +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 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +class TestProjectUpdate: + @pytest.mark.parametrize( + "namespace, name,new_name", + [ + ["namespace1", "amendments1", "name1"], + ["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") + + @pytest.mark.parametrize( + "namespace, name,new_name", + [ + ["namespace1", "amendments1", "name1"], + ["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") + + @pytest.mark.parametrize( + "namespace, name, new_tag", + [ + ["namespace1", "amendments1", "tag1"], + ["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) + + @pytest.mark.parametrize( + "namespace, name, new_description", + [ + ["namespace1", "amendments1", "desc1 f"], + ["namespace2", "derive", "desc5 f"], + ], + ) + def test_update_project_description( + self, initiate_pepdb_con, 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}, + ) + + assert ( + initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False).description + == new_description + ) + + @pytest.mark.parametrize( + "namespace, name, new_description", + [ + ["namespace1", "amendments1", "desc1 f"], + ], + ) + def test_update_project_description_in_config( + self, initiate_pepdb_con, 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}, + ) + + assert ( + initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=False).description + == new_description + ) + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["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}, + ) + + @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 + + @pytest.mark.parametrize( + "namespace, name", + [ + ["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}, + ) + + is_private = ( + initiate_pepdb_con.annotation.get( + namespace=namespace, name=name, tag="default", admin=[namespace] + ) + .results[0] + .is_private + ) + assert is_private is True + + @pytest.mark.parametrize( + "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}, + ) + + pop = ( + initiate_pepdb_con.annotation.get( + namespace=namespace, name=name, tag="default", admin=[namespace] + ) + .results[0] + .pop + ) + 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}, + ) + + pop = ( + initiate_pepdb_con.annotation.get( + namespace=namespace, name=name, tag="default", admin=[namespace] + ) + .results[0] + .pop + ) + assert pop is False + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "basic"], + ], + ) + def test_project_can_have_2_sample_names(self, initiate_pepdb_con, 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") + + 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)}, + ) + + prj = initiate_pepdb_con.project.get(namespace=namespace, name=name, raw=True) + + assert len(prj["_sample_dict"]) == 4 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +class TestUpdateProjectWithId: + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + ["namespace3", "subtable1"], + ], + ) + def test_update_whole_project_with_id(self, initiate_pepdb_con, namespace, name): + pass + + # TODO: write more tests + + @pytest.mark.parametrize( + "namespace, name", + [ + ["namespace1", "amendments1"], + # ["namespace3", "subtable1"], + ], + ) + def test_update_project_with_duplicated_sample_guids( + self, initiate_pepdb_con, 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)}, + ) + + +class TestProjectSamplesUpdates: + ... + # TODO: write tests diff --git a/tests/test_views.py b/tests/test_views.py new file mode 100644 index 0000000..f7b0590 --- /dev/null +++ b/tests/test_views.py @@ -0,0 +1,286 @@ +import os +import warnings + +import pytest +from sqlalchemy.exc import OperationalError + +import pepdbagent +from pepdbagent.exceptions import ( + SampleAlreadyInView, + SampleNotFoundError, + SampleNotInViewError, + 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 + + +@pytest.mark.skipif( + not db_setup(), + reason="DB is not setup", +) +class TestViews: + """ + Test function within view class + """ + + @pytest.mark.parametrize( + "namespace, name, sample_name, view_name", + [ + ["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"], + }, + ) + + 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 + + @pytest.mark.parametrize( + "namespace, name, sample_name, view_name", + [ + ["namespace1", "amendments1", "pig_0h", "view1"], + ], + ) + def test_create_view_with_incorrect_sample( + self, initiate_pepdb_con, 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"], + }, + ) + + @pytest.mark.parametrize( + "namespace, name, sample_name, view_name", + [ + ["namespace1", "amendments1", "pig_0h", "view1"], + ], + ) + def test_create_view_with_incorrect_sample_no_fail( + self, initiate_pepdb_con, 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 + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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 + ) + == 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 + + @pytest.mark.parametrize( + "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 + ) + == 2 + ) + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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 + ) + == 3 + ) + + @pytest.mark.parametrize( + "namespace, name, sample_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 + ) + == 1 + ) + assert len(initiate_pepdb_con.project.get(namespace, name, raw=False).samples) == 4 + + with pytest.raises(SampleNotInViewError): + initiate_pepdb_con.view.remove_sample(namespace, name, "default", "view1", sample_name) + + @pytest.mark.parametrize( + "namespace, name, sample_name", + [ + ["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) + + @pytest.mark.parametrize( + "namespace, name, sample_name, view_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"], + ) + + assert len(snap_project.samples) == 2 + + @pytest.mark.parametrize( + "namespace, name, sample_name, view_name", + [ + ["namespace1", "amendments1", "pig_0h", "view1"], + ], + ) + def test_get_view_list_from_project( + self, initiate_pepdb_con, 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 + )