Skip to content
Snippets Groups Projects
Commit 1281606e authored by AndiMajore's avatar AndiMajore
Browse files

python_nedrex code (for now)

parent 214e6c6b
Branches
No related tags found
No related merge requests found
from typing import Any as _Any
from typing import Dict as _Dict
from typing import Optional as _Optional
from python_nedrex import config as _config
from python_nedrex.common import check_response as _check_response
from python_nedrex.common import download_file as _download_file
from python_nedrex.common import http as _http
def get_metadata() -> _Dict[str, _Any]:
url = f"{_config.url_base}/static/metadata"
resp = _http.get(url, headers={"x-api-key": _config.api_key})
result: _Dict[str, _Any] = _check_response(resp)
return result
def get_license() -> str:
url = f"{_config.url_base}/static/license"
resp = _http.get(url)
result: str = _check_response(resp)
return result
def download_lengths_map(target: _Optional[str] = None) -> None:
if target is None:
target = "lengths.map"
url = f"{_config.url_base}/static/lengths.map"
_download_file(url, target)
from typing import List as _List
from typing import Optional as _Optional
from python_nedrex import config as _config
from python_nedrex.common import check_response as _check_response
from python_nedrex.common import check_status_factory as _check_status_factory
from python_nedrex.common import http as _http
def trustrank_submit(
seeds: _List[str],
damping_factor: float = 0.85,
only_direct_drugs: bool = True,
only_approved_drugs: bool = True,
n: _Optional[int] = None, # pylint: disable=C0103
) -> str:
url = f"{_config.url_base}/trustrank/submit"
body = {
"seeds": seeds,
"damping_factor": damping_factor,
"only_direct_drugs": only_direct_drugs,
"only_approved_drugs": only_approved_drugs,
"N": n,
}
resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
result: str = _check_response(resp)
return result
check_trustrank_status = _check_status_factory("/trustrank/status")
def download_trustrank_results(uid: str) -> str:
url = f"{_config.url_base}/trustrank/download"
params = {"uid": uid}
resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
result: str = _check_response(resp, return_type="text")
return result
from typing import List as _List
from python_nedrex import config as _config
from python_nedrex.common import check_response as _check_response
from python_nedrex.common import check_status_factory as _check_status_factory
from python_nedrex.common import http as _http
check_validation_status = _check_status_factory("/validation/status")
def check_module_member_type(mmt: str) -> None:
if mmt not in {"gene", "protein"}:
raise ValueError(f"module_member_type {mmt!r} is invalid (should be 'gene' or 'protein'")
# pylint: disable=R0913
def joint_validation_submit(
module_members: _List[str],
module_member_type: str,
test_drugs: _List[str],
true_drugs: _List[str],
permutations: int,
only_approved_drugs: bool = True,
) -> str:
check_module_member_type(module_member_type)
url = f"{_config.url_base}/validation/joint"
body = {
"module_members": module_members,
"module_member_type": module_member_type,
"test_drugs": test_drugs,
"true_drugs": true_drugs,
"permutations": permutations,
"only_approved_drugs": only_approved_drugs,
}
resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
result: str = _check_response(resp)
return result
# pylint: enable=R0913
def module_validation_submit(
module_members: _List[str],
module_member_type: str,
true_drugs: _List[str],
permutations: int,
only_approved_drugs: bool = True,
) -> str:
check_module_member_type(module_member_type)
url = f"{_config.url_base}/validation/module"
body = {
"module_members": module_members,
"module_member_type": module_member_type,
"true_drugs": true_drugs,
"permutations": permutations,
"only_approved_drugs": only_approved_drugs,
}
resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
result: str = _check_response(resp)
return result
def drug_validation_submit(
test_drugs: _List[str], true_drugs: _List[str], permutations: int, only_approved_drugs: bool = True
) -> str:
url = f"{_config.url_base}/validation/drug"
body = {
"test_drugs": test_drugs,
"true_drugs": true_drugs,
"permutations": permutations,
"only_approved_drugs": only_approved_drugs,
}
resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
result: str = _check_response(resp)
return result
from typing import Any as _Any
from typing import Dict as _Dict
from typing import Generator as _Generator
from typing import List as _List
from typing import Optional as _Optional
from python_nedrex import config as _config
from python_nedrex.common import check_pagination_limit as _check_pagination_limit
from python_nedrex.common import check_response as _check_response
from python_nedrex.common import get_pagination_limit as _get_pagination_limit
from python_nedrex.common import http as _http
def get_effect_choices() -> _List[str]:
url = f"{_config.url_base}/variants/get_effect_choices"
resp = _http.get(url, headers={"x-api-key": _config.api_key})
result: _List[str] = _check_response(resp)
return result
def get_review_status_choices() -> _List[str]:
url = f"{_config.url_base}/variants/get_review_choices"
resp = _http.get(url, headers={"x-api-key": _config.api_key})
result: _List[str] = _check_response(resp)
return result
# pylint: disable=R0913
def get_variant_disorder_associations(
variant_ids: _Optional[_List[str]] = None,
disorder_ids: _Optional[_List[str]] = None,
review_status: _Optional[_List[str]] = None,
effect: _Optional[_List[str]] = None,
limit: _Optional[int] = None,
offset: int = 0,
) -> _List[_Dict[str, _Any]]:
max_limit = _get_pagination_limit()
if isinstance(limit, int):
_check_pagination_limit(limit, max_limit)
else:
limit = max_limit
params = {
"variant_id": variant_ids,
"disorder_id": disorder_ids,
"review_status": review_status,
"effect": effect,
"limit": limit,
"offset": offset,
}
url = f"{_config.url_base}/variants/get_variant_disorder_associations"
resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
result: _List[_Dict[str, _Any]] = _check_response(resp)
return result
# pylint: enable=R0913
def iter_variant_disorder_associations(
variant_ids: _Optional[_List[str]] = None,
disorder_ids: _Optional[_List[str]] = None,
review_status: _Optional[_List[str]] = None,
effect: _Optional[_List[str]] = None,
) -> _Generator[_Dict[str, _Any], None, None]:
max_limit = _get_pagination_limit()
offset = 0
kwargs = {
"variant_ids": variant_ids,
"disorder_ids": disorder_ids,
"review_status": review_status,
"effect": effect,
"limit": max_limit,
"offset": offset,
}
while True:
results = get_variant_disorder_associations(**kwargs)
if len(results) == 0:
return
yield from results
offset += max_limit
kwargs["offset"] = offset
def get_variant_gene_associations(
variant_ids: _Optional[_List[str]] = None,
gene_ids: _Optional[_List[str]] = None,
limit: _Optional[int] = None,
offset: int = 0,
) -> _List[_Dict[str, _Any]]:
max_limit = _get_pagination_limit()
if isinstance(limit, int):
_check_pagination_limit(limit, max_limit)
else:
limit = max_limit
params = {
"variant_id": variant_ids,
"gene_id": gene_ids,
"offset": offset,
"limit": limit,
}
url = f"{_config.url_base}/variants/get_variant_gene_associations"
resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
result: _List[_Dict[str, _Any]] = _check_response(resp)
return result
def iter_variant_gene_associations(
variant_ids: _Optional[_List[str]] = None,
gene_ids: _Optional[_List[str]] = None,
) -> _Generator[_Dict[str, _Any], None, None]:
max_limit = _get_pagination_limit()
offset = 0
kwargs = {
"variant_ids": variant_ids,
"gene_ids": gene_ids,
"limit": max_limit,
"offset": offset,
}
while True:
results = get_variant_gene_associations(**kwargs)
if len(results) == 0:
return
yield from results
offset += max_limit
kwargs["offset"] = offset
def get_variant_based_disorder_associated_genes(
disorder_id: str, review_status: _Optional[_List[str]] = None, effect: _Optional[_List[str]] = None
) -> _List[_Dict[str, _Any]]:
params = {"disorder_id": disorder_id, "review_status": review_status, "effect": effect}
url = f"{_config.url_base}/variants/variant_based_disorder_associated_genes"
resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
result: _List[_Dict[str, _Any]] = _check_response(resp)
return result
def get_variant_based_gene_associated_disorders(
gene_id: str, review_status: _Optional[_List[str]] = None, effect: _Optional[_List[str]] = None
) -> _List[_Dict[str, _Any]]:
params = {"gene_id": gene_id, "review_status": review_status, "effect": effect}
url = f"{_config.url_base}/variants/variant_based_gene_associated_disorders"
resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
result: _List[_Dict[str, _Any]] = _check_response(resp)
return result
import os as _os
from typing import Optional as _Optional
from python_nedrex import config as _config
from python_nedrex.common import http as _http
from python_nedrex.decorators import check_url_vpd as _check_url_vpd
@_check_url_vpd
def get_vpd(disorder: str, number_of_patients: int, out_dir: str) -> _Optional[str]:
"""
Downloads a .zip archive with the requested virtual patient data to the given directory.
Parameters:
disorder (str): The disorder mondoID (e.g mondo.0000090) for which the virtual patient should be retrieved.
number_of_patients (int): The number of simulated patients in the dataset. Can be 1, 10 or 100.
out_dir (str): The absolute path of a directory where the virtual patient data should be stored.
Returns:
archive (str): Absolute path of the downloaded zip archive or None if the requested resource does not exist.
"""
archive_name: str = f"{disorder}_1000GP_{number_of_patients}VPSim.zip"
url: str = f"{_config.url_vpd}/vpd/{disorder}/{archive_name}"
archive: str = _os.path.join(out_dir, archive_name)
data = _http.get(url)
if data.status_code != 200:
return None
with open(archive, "wb") as arch:
arch.write(data.content)
return archive
attrs==21.4.0
cachetools==4.2.4
more-itertools==8.13.0
pytest==7.0.1
requests==2.27.1
pip==19.2.3
bump2version==0.5.11
wheel==0.33.6
watchdog==0.9.0
flake8==3.7.8
tox==3.14.0
coverage==4.5.4
Sphinx==1.8.5
twine==1.14.0
pytest==6.2.4
black==21.7b0
[bumpversion]
current_version = 0.1.1
commit = True
tag = True
[bumpversion:file:setup.py]
search = version='{current_version}'
replace = version='{new_version}'
[bumpversion:file:python_nedrex/__init__.py]
search = __version__ = '{current_version}'
replace = __version__ = '{new_version}'
[bdist_wheel]
universal = 1
[flake8]
exclude = docs
[tool:pytest]
#!/usr/bin/env python
"""The setup script."""
from setuptools import setup, find_packages
with open("README.rst") as readme_file:
readme = readme_file.read()
with open("HISTORY.rst") as history_file:
history = history_file.read()
requirements = [
"attrs",
"requests",
"cachetools",
]
test_requirements = [
"pytest>=3",
]
setup(
author="David James Skelton",
author_email="james.skelton@newcastle.ac.uk",
python_requires=">=3.6",
classifiers=[
"Development Status :: 2 - Pre-Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Natural Language :: English",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
],
description="A Python library for interfacing with the PNeDRex API",
install_requires=requirements,
license="MIT license",
long_description=readme + "\n\n" + history,
include_package_data=True,
keywords="python_nedrex",
name="python_nedrex",
packages=find_packages(include=["python_nedrex", "python_nedrex.*"]),
test_suite="tests",
tests_require=test_requirements,
url="https://github.com/james-skelton/python_nedrex",
version="0.1.1",
zip_safe=False,
)
"""Unit test package for python_nedrex."""
#!/usr/bin/env python
"""Tests for `python_nedrex` package."""
import os
import re
from pathlib import Path
import random
import tempfile
import time
from contextlib import contextmanager
from functools import lru_cache
from more_itertools import take
import pytest
import requests
import python_nedrex
from python_nedrex.common import get_pagination_limit
from python_nedrex.core import (
get_edges,
iter_edges,
iter_nodes,
get_node_types,
get_edge_types,
get_collection_attributes,
get_node_ids,
get_nodes,
api_keys_active,
)
from python_nedrex.diamond import diamond_submit, check_diamond_status, download_diamond_results
from python_nedrex.disorder import (
get_disorder_ancestors,
get_disorder_children,
get_disorder_descendants,
get_disorder_parents,
search_by_icd10,
)
from python_nedrex.domino import (
domino_submit,
check_domino_status
)
from python_nedrex.exceptions import ConfigError, NeDRexError
from python_nedrex.graph import (
build_request,
check_build_status,
download_graph,
)
from python_nedrex.must import must_request, check_must_status
from python_nedrex.ppi import ppis
from python_nedrex.relations import (
get_encoded_proteins,
get_drugs_indicated_for_disorders,
get_drugs_targetting_proteins,
get_drugs_targetting_gene_products,
)
API_URL = "http://82.148.225.92:8123/"
API_KEY = requests.post(f"{API_URL}admin/api_key/generate", json={"accept_eula": True}).json()
SEEDS = [
"P43121",
"P01589",
"P30203",
"P21554",
"P01579",
"O43557",
"Q99572",
"P01920",
"P25942",
"P01189",
"P21580",
"Q02556",
"P01584",
"P01574",
"P02649",
"P29466",
"P22301",
"P16581",
"P06276",
"P11473",
"O60333",
"P19256",
"Q96P20",
"P01911",
"Q2KHT3",
"P18510",
"P05362",
"P01903",
"P29597",
"P13232",
"Q13191",
"Q06330",
"P04440",
"P78508",
"P19320",
"P19438",
"P02774",
"O75508",
"P29459",
"P16871",
"Q14765",
"Q16552",
]
UID_REGEX = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
@contextmanager
def url_base():
python_nedrex.config.set_url_base(API_URL)
yield
python_nedrex.config._url_base = None
@contextmanager
def api_key():
python_nedrex.config.set_api_key(API_KEY)
yield
python_nedrex.config._api_key = None
@lru_cache(maxsize=10)
def get_node_collections():
with api_key(), url_base():
collections = get_node_types()
return collections
@lru_cache(maxsize=10)
def get_edge_collections():
with api_key(), url_base():
collections = get_edge_types()
return collections
def get_random_disorder_selection(n, skip_root=True):
random.seed(20220621)
with api_key(), url_base():
disorder_ids = set(get_node_ids("disorder"))
disorder_ids.remove("mondo.0000001")
return random.sample(sorted(disorder_ids), n)
@pytest.fixture
def config():
return {"api_url": API_URL, "api_key": API_KEY}
@pytest.fixture
def set_api_key(config):
with api_key():
yield
@pytest.fixture
def set_base_url(config):
with url_base():
yield
def test_set_api_base(set_base_url):
assert python_nedrex.config._url_base == API_URL.rstrip("/")
class TestGetNodeTypes:
@pytest.fixture
def result(self, set_base_url, set_api_key):
result = get_node_types()
return result
def test_return_type(self, result):
assert isinstance(result, list)
assert all(isinstance(item, str) for item in result)
def test_ordering(self, result):
assert result == sorted(result)
def test_content(self, result):
assert "protein" in result
class TestGetEdgeTypes:
@pytest.fixture
def result(self, set_base_url, set_api_key):
result = get_edge_types()
return result
def test_return_type(self, result):
assert isinstance(result, list)
assert all(isinstance(item, str) for item in result)
def test_ordering(self, result):
assert result == sorted(result)
def test_content(self, result):
assert "protein_encoded_by_gene" in result
class TestGetCollectionAttributes:
@pytest.mark.parametrize("collection", get_node_collections())
def test_get_node_collection_attributes(self, set_base_url, set_api_key, collection):
expected_attributes = ("primaryDomainId", "domainIds", "type")
coll_attributes = get_collection_attributes(collection)
assert all(attr in coll_attributes for attr in expected_attributes)
@pytest.mark.parametrize("collection", get_edge_collections())
def test_get_edge_collection_attributes(self, set_base_url, set_api_key, collection):
# NOTE: Exclude the protein_interacts_with_protein collection because of its size.
coll_attributes = get_collection_attributes(collection)
assert "type" in coll_attributes
assert all(attr in coll_attributes for attr in ("memberOne", "memberTwo")) or all(
attr in coll_attributes for attr in ("sourceDomainId", "targetDomainId")
)
class TestGetNodeIds:
@pytest.mark.parametrize("collection", get_node_collections())
def test_get_node_ids(self, set_base_url, set_api_key, collection):
assert get_node_ids(collection)
@pytest.mark.parametrize("collection", get_edge_collections())
def test_get_node_ids_fails_for_edges(self, set_base_url, set_api_key, collection):
with pytest.raises(NeDRexError):
get_node_ids(collection)
class TestGetEdgeRoutes:
@pytest.mark.parametrize("collection", get_edge_collections())
def test_return_type_get_edges(self, set_base_url, set_api_key, collection):
edges = get_edges(collection, limit=1_000)
assert isinstance(edges, list)
@pytest.mark.parametrize("collection", get_edge_collections())
def test_edge_attributes(self, set_base_url, set_api_key, collection):
result = get_collection_attributes(collection, include_counts=True)
total = result["document_count"]
attr_counts = result["attribute_counts"]
assert attr_counts["type"] == total
assert (
attr_counts.get("memberOne", 0) == attr_counts.get("memberTwo", 0) == total
and attr_counts.get("sourceDomainId", 0) == attr_counts.get("targetDomainId", 0) == 0
) ^ (
attr_counts.get("memberOne", 0) == attr_counts.get("memberTwo", 0) == 0
and attr_counts.get("sourceDomainId", 0) == attr_counts.get("targetDomainId", 0) == total
)
class TestGetNodeRoutes:
@pytest.mark.parametrize("collection", get_node_collections())
def test_get_all_nodes(self, set_base_url, set_api_key, collection):
assert isinstance(get_nodes(collection), list)
def test_get_specific_nodes(self, set_base_url, set_api_key):
nodes = get_nodes("disorder", node_ids=["mondo.0000001"])
assert isinstance(nodes, list)
assert len(nodes) == 1
assert nodes[0]["primaryDomainId"] == "mondo.0000001"
def test_get_drugs_with_api_key(self, set_base_url, set_api_key):
nodes = get_nodes("drug")
assert isinstance(nodes, list)
def test_get_specific_attributes(self, set_base_url, set_api_key):
nodes = get_nodes("disorder", attributes=["displayName"])
assert isinstance(nodes, list)
assert [set(i.keys()) == {"primaryDomainId", "displayName"} for i in nodes]
def test_get_specific_attribute_and_nodes(self, set_base_url, set_api_key):
nodes = get_nodes("disorder", attributes=["displayName"], node_ids=["mondo.0000001"])
assert isinstance(nodes, list)
assert len(nodes) == 1
assert nodes[0] == {
"displayName": "disease or disorder",
"primaryDomainId": "mondo.0000001",
}
def test_pagination(self, set_base_url, set_api_key):
nodes = get_nodes("genomic_variant", limit=1000, offset=1000)
assert isinstance(nodes, list)
assert len(nodes) == 1000
def test_consistent_pagination(self, set_base_url, set_api_key):
offset = 1234
limit = 69
nodes = get_nodes("genomic_variant", limit=limit, offset=offset)
for _ in range(10):
nodes_repeat = get_nodes("genomic_variant", limit=limit, offset=offset)
assert nodes_repeat == nodes
class TestDisorderRoutes:
def test_search_by_icd10(self, set_base_url, set_api_key):
# NOTE: There is currently an ICD-10 mapping issue due to MONDO
search_by_icd10("I52")
def test_get_disorder_ancestors(self, set_base_url, set_api_key):
# Check that `disease or disorder`is an ancestor of `lupus nephritis`
# `disease or disorder` is not a parent of `lupus neprhitis`
lupus_nephritis = "mondo.0005556"
disease_or_disorder = "mondo.0000001"
result = get_disorder_ancestors(lupus_nephritis)
assert disease_or_disorder in result[lupus_nephritis]
def test_get_disorder_descendants(self, set_base_url, set_api_key):
# Check that `lupus nephritis` is a descendant of `inflammatory disease`
# `lupus nephritis` is not a child of `inflammatory disease`
inflam_disease = "mondo.0021166"
lupus_nephritis = "mondo.0005556"
result = get_disorder_descendants(inflam_disease)
assert lupus_nephritis in result[inflam_disease]
def test_get_disorder_parents(self, set_base_url, set_api_key):
# Check that `glomerulonephritis` is a parent of `lupus nephritis`
lupus_nephritis = "mondo.0005556"
glomerulonephritis = "mondo.0002462"
result = get_disorder_parents("mondo.0005556")
assert glomerulonephritis in result[lupus_nephritis]
def test_get_disorder_children(self, set_base_url, set_api_key):
# Check that `lupus nephritis` is a child of `glomerulonephritis`
glomerulonephritis = "mondo.0002462"
lupus_nephritis = "mondo.0005556"
result = get_disorder_children(glomerulonephritis)
assert lupus_nephritis in result[glomerulonephritis]
@pytest.mark.parametrize("chosen_id", get_random_disorder_selection(20))
def test_parent_child_reciprocity(self, set_base_url, set_api_key, chosen_id):
parents = get_disorder_parents(chosen_id)
children_of_parents = get_disorder_children(parents[chosen_id])
assert all(chosen_id in value for value in children_of_parents.values())
@pytest.mark.parametrize("chosen_id", get_random_disorder_selection(20))
def test_ancestor_descendant_reciprocity(self, set_base_url, set_api_key, chosen_id):
parents = get_disorder_ancestors(chosen_id)
descendants_of_parents = get_disorder_descendants(parents[chosen_id])
assert all(chosen_id in value for value in descendants_of_parents.values())
class TestRoutesFailWithoutAPIUrl:
def test_get_node_type(self, set_api_key):
with pytest.raises(ConfigError) as excinfo:
get_node_types()
assert "API URL is not set in the config" == str(excinfo.value)
def test_get_edge_type(self, set_api_key):
with pytest.raises(ConfigError) as excinfo:
get_edge_types()
assert "API URL is not set in the config" == str(excinfo.value)
class TestRoutesFailWithoutAPIKey:
def test_get_node_type(self, set_base_url):
if not api_keys_active():
return
with pytest.raises(ConfigError) as excinfo:
get_node_types()
assert "no API key set in the configuration" == str(excinfo.value)
if not api_keys_active():
return
with pytest.raises(ConfigError) as excinfo:
get_edge_types()
assert "no API key set in the configuration" == str(excinfo.value)
@pytest.mark.parametrize("collection", get_node_collections())
def test_node_routes_fail(self, set_base_url, collection):
if not api_keys_active():
return
with pytest.raises(ConfigError) as excinfo:
get_collection_attributes(collection)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
get_node_ids(collection)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
get_nodes(collection)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
for _ in take(1, iter_nodes(collection)):
pass
assert "no API key set in the configuration" == str(excinfo.value)
@pytest.mark.parametrize("collection", get_edge_collections())
def test_edge_routes_fail(self, set_base_url, collection):
if not api_keys_active():
return
with pytest.raises(ConfigError) as excinfo:
get_collection_attributes(collection)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
get_edges(collection)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
for _ in take(1, iter_edges(collection)):
pass
assert "no API key set in the configuration" == str(excinfo.value)
def test_disorder_routes_fail(self, set_base_url):
disorder_id = "mondo.0000001" # root of the MONDO tree
icd10_id = "I59.1" # Heart disease, unspecified
with pytest.raises(ConfigError) as excinfo:
get_disorder_children(disorder_id)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
get_disorder_parents(disorder_id)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
get_disorder_ancestors(disorder_id)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
get_disorder_descendants(disorder_id)
assert "no API key set in the configuration" == str(excinfo.value)
with pytest.raises(ConfigError) as excinfo:
search_by_icd10(icd10_id)
assert "no API key set in the configuration" == str(excinfo.value)
class TestPPIRoute:
def test_ppi_route(self, set_base_url, set_api_key):
ppis(["exp"], 0, get_pagination_limit())
def test_overlap_with_pagination(self, set_base_url, set_api_key):
page_limit = 1_000
delta = page_limit // 2
skip = delta
previous = ppis(["exp"], 0, page_limit)
for _ in range(100):
current = ppis(["exp"], skip, page_limit)
assert previous[-delta:] == current[:delta]
previous = current
skip += delta
def test_each_evidence_type_works(self, set_base_url, set_api_key):
for evidence_type in ["exp", "pred", "ortho"]:
results = ppis([evidence_type], 0, get_pagination_limit())
assert all(evidence_type in doc["evidenceTypes"] for doc in results)
def test_fails_with_invalid_type(self, set_base_url, set_api_key):
for evidence_type in ["exps", "pr3d", "orth"]:
with pytest.raises(NeDRexError) as excinfo:
ppis([evidence_type])
err_val = {evidence_type}
assert str(excinfo.value) == f"unexpected evidence types: {err_val}"
def test_fails_with_large_limit(self, set_base_url, set_api_key):
page_limit = get_pagination_limit()
with pytest.raises(NeDRexError) as excinfo:
ppis(["exp"], limit=page_limit + 1)
assert str(excinfo.value) == f"limit={page_limit + 1:,} is too great (maximum is {page_limit:,})"
class TestRelationshipRoutes:
def test_get_encoded_proteins(self, set_base_url, set_api_key):
# NOTE: If result changes, check these examples are still accurate.
histamine_receptor_genes = ["3269", 3274, "entrez.11255"] # HRH1, as str # HRH2, as int # HRH3, as prefix
results = get_encoded_proteins(histamine_receptor_genes)
assert "P35367" in results["3269"]
assert "P25021" in results["3274"]
assert "Q9Y5N1" in results["11255"]
def test_get_drugs_indicated_for_disorders(self, set_base_url, set_api_key):
# NOTE: If result changes, check these examples are still accurate.
disorders = [
"mondo.0005393", # Gout
"0005362", # ED
]
results = get_drugs_indicated_for_disorders(disorders)
assert "DB00437" in results["0005393"] # Allopurinol for gout
assert "DB00203" in results["0005362"] # Sildenafil for ED
def test_get_drugs_targetting_proteins(self, set_base_url, set_api_key):
# NOTE: If result changes, check these examples are still accurate.
proteins = [
"P35367", # Histamine H1 receptor, targetted by antihistamines
"uniprot.P03372", # Estrogen receptor α, targetted by ethinylestradiol
]
results = get_drugs_targetting_proteins(proteins)
assert "DB00341" in results["P35367"]
assert "DB00977" in results["P03372"]
def test_get_drugs_targetting_gene_products(self, set_base_url, set_api_key):
genes = [
"entrez.3269", # HRH1 gene (product targetted by antihistamines)
2099, # Estrogen receptor α gene (product targetted by ethinylestradiol)
"6532", # SLC6A4, encodes Sodium-dependent serotonin transporter, targetted by SSRIs
]
results = get_drugs_targetting_gene_products(genes)
assert "DB00341" in results["3269"]
assert "DB00977" in results["2099"]
assert "DB00215" in results["6532"]
class TestGraphRoutes:
def test_default_build(self, set_base_url, set_api_key):
build_request()
@pytest.mark.parametrize(
"kwargs",
[
{"nodes": ["this_is_not_a_node"]},
{"edges": ["this_is_not_an_edge"]},
{"ppi_evidence": ["made_up"]},
{"taxid": ["human"]},
],
)
def test_build_fails_with_invalid_params(self, kwargs, set_base_url, set_api_key):
with pytest.raises(NeDRexError):
build_request(**kwargs)
def test_get_uid(self, set_base_url, set_api_key):
uid = build_request()
assert UID_REGEX.match(uid)
check_build_status(uid)
def test_fails_with_invalid_uid(self, set_base_url, set_api_key):
uid = "this-is-not-a-valid-uid!"
with pytest.raises(NeDRexError):
check_build_status(uid)
def test_download_graph(self, set_base_url, set_api_key):
uid = build_request()
while True:
status = check_build_status(uid)
if status["status"] == "completed":
break
time.sleep(10)
download_graph(uid)
p = Path(f"{uid}.graphml")
assert p.exists()
p.unlink()
def test_download_graph_different_dir(self, set_base_url, set_api_key):
with tempfile.TemporaryDirectory() as tmpdir:
uid = build_request()
while True:
status = check_build_status(uid)
if status["status"] == "completed":
break
time.sleep(10)
target = os.path.join(tmpdir, "mygraph.graphml")
download_graph(uid, target)
p = Path(target)
assert p.exists()
p.unlink()
def test_download_fails_with_invalid_uid(self, set_base_url, set_api_key):
uid = "this-is-not-a-valid-uid!"
with pytest.raises(NeDRexError):
download_graph(uid)
class TestMustRoutes:
def test_simple_request(self, set_base_url, set_api_key):
uid = must_request(SEEDS, 0.5, True, 10, 2)
assert UID_REGEX.match(uid)
def test_must_status(self, set_base_url, set_api_key):
uid = must_request(SEEDS, 0.5, True, 10, 2)
status = check_must_status(uid)
assert isinstance(status, dict)
assert "status" in status.keys()
@pytest.mark.parametrize(
"update",
[
{"hubpenalty": 1.01},
{"hubpenalty": -0.01},
{"hubpenalty": None},
{"multiple": None},
{"trees": -1},
{"trees": None},
{"maxit": -1},
{"maxit": None},
],
)
def test_must_fails_with_invalid_arguments(self, set_base_url, set_api_key, update):
kwargs = {"seeds": SEEDS, "hubpenalty": 0.5, "multiple": True, "trees": 10, "maxit": 2, "network": "DEFAULT"}
with pytest.raises(NeDRexError):
kwargs = {**kwargs, **update}
must_request(**kwargs)
class TestDiamondRoutes:
def test_simple_request(self, set_base_url, set_api_key):
uid = diamond_submit(SEEDS, 10)
assert UID_REGEX.match(uid)
def test_diamond_status(self, set_base_url, set_api_key):
uid = diamond_submit(SEEDS, 10)
status = check_diamond_status(uid)
assert isinstance(status, dict)
assert "status" in status.keys()
def test_diamond_download(self, set_base_url, set_api_key):
uid = diamond_submit(SEEDS, 10)
while True:
status = check_diamond_status(uid)
if status["status"] == "completed":
break
time.sleep(10)
download_diamond_results(uid)
def test_diamond_fails_with_invalid_arguments(self, set_base_url, set_api_key):
with pytest.raises(ValueError):
diamond_submit(SEEDS, n=10, edges="some")
class TestDominoRoutes:
def test_simple_request(self, set_base_url, set_api_key):
uid = domino_submit(SEEDS)
assert UID_REGEX.match(uid)
def test_check_domino_status(self, set_base_url, set_api_key):
uid = domino_submit(SEEDS)
status = check_domino_status(uid)
assert isinstance(status, dict)
assert "status" in status.keys()
\ No newline at end of file
[tox]
envlist = py36, py37, py38, flake8
[travis]
python =
3.8: py38
3.7: py37
3.6: py36
[testenv:flake8]
basepython = python
deps = flake8
commands = flake8 python_nedrex tests
[testenv]
setenv =
PYTHONPATH = {toxinidir}
deps =
-r{toxinidir}/requirements_dev.txt
; If you want to make tox run the tests with the same versions, create a
; requirements.txt with the pinned versions and uncomment the following line:
; -r{toxinidir}/requirements.txt
commands =
pip install -U pip
pytest --basetemp={envtmpdir}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment