Split out the local backend (#2052)

This splits the backend into two parts: the local backend for desktop cellxgene and the AWS backend for hosted cellxgene. The local backend is in local_server while the hosted remains in server. The general idea is to copy everything from server to local_server, pull unneeded stuff out of local_server, and keep server as-is for this PR. Not touching server means all the infra and deployment code will continue working just as it did before so we can make those changes incrementally.
This commit is contained in:
Marcus Kinsella
2021-02-18 12:58:22 -08:00
committed by GitHub
parent 036b5f8c0f
commit fb61bd6e9c
153 changed files with 14027 additions and 46 deletions
View File
+89
View File
@@ -0,0 +1,89 @@
import unittest
import requests
from local_server.common.config.app_config import AppConfig
from local_server.test import H5AD_FIXTURE, test_server
class AuthTest(unittest.TestCase):
def setUp(self):
self.dataset_datapath = H5AD_FIXTURE
def test_auth_none(self):
app_config = AppConfig()
app_config.update_server_config(app__flask_secret_key="secret")
app_config.update_server_config(authentication__type=None, single_dataset__datapath=self.dataset_datapath)
app_config.update_dataset_config(user_annotations__enable=False)
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertNotIn("authentication", config["config"])
self.assertIsNone(userinfo)
def test_auth_session(self):
app_config = AppConfig()
app_config.update_server_config(app__flask_secret_key="secret")
app_config.update_server_config(authentication__type="session", single_dataset__datapath=self.dataset_datapath)
app_config.update_dataset_config(user_annotations__enable=True)
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertFalse(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "anonymous")
def test_auth_test_single(self):
app_config = AppConfig()
app_config.update_server_config(app__flask_secret_key="secret")
app_config.update_server_config(
authentication__type="test",
single_dataset__datapath=self.dataset_datapath,
authentication__insecure_test_environment=True,
)
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(config["config"]["parameters"]["annotations"])
login_uri = config["config"]["authentication"]["login"]
logout_uri = config["config"]["authentication"]["logout"]
self.assertEqual(login_uri, "/login")
self.assertEqual(logout_uri, "/logout")
response = session.get(f"{server}/{login_uri}")
# check that the login redirect worked
self.assertEqual(response.history[0].status_code, 302)
self.assertEqual(response.url, f"{server}/")
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "test_account")
self.assertTrue(config["config"]["parameters"]["annotations"])
response = session.get(f"{server}/{logout_uri}")
# check that the logout redirect worked
self.assertEqual(response.history[0].status_code, 302)
self.assertEqual(response.url, f"{server}/")
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["parameters"]["annotations"])
+27
View File
@@ -0,0 +1,27 @@
import filecmp
import os
import shutil
import unittest
import yaml
from local_server.default_config import default_config
from local_server.test import FIXTURES_ROOT
class CLIPLaunchTests(unittest.TestCase):
tmp_dir = os.path.join(FIXTURES_ROOT, "dump_configs")
@classmethod
def setUpClass(cls) -> None:
os.mkdir(cls.tmp_dir)
@classmethod
def tearDownClass(cls) -> None:
shutil.rmtree(cls.tmp_dir)
def test_dump_default_config(self):
os.system(f"cellxgene launch --dump-default-config > {self.tmp_dir}/test_config_dump.txt")
with open(f"{self.tmp_dir}/expected_config_dump.txt", "w") as expected_config:
expected_config.write(yaml.dump(default_config))
filecmp.cmp(f"{self.tmp_dir}/expected_config_dump.txt", f"{self.tmp_dir}/test_config_dump.txt")
@@ -0,0 +1,15 @@
import unittest
import pandas as pd
from local_server.cli.prepare import make_index_unique
class CLIPrepareTests(unittest.TestCase):
""" Test cases for CLI prepare logic """
def test_make_index_unique(self):
index = pd.Index(["SNORD113", "SNORD113", "SNORD113-1"])
result = make_index_unique(index)
expected = pd.Index(["SNORD113", "SNORD113-2", "SNORD113-1"])
self.assertTrue(all(left == right for left, right in zip(result.values, expected.values)))
@@ -0,0 +1,28 @@
import unittest
from local_server.cli.upgrade import validate_version_str, split_version, version_gt
class CLIUpgradeTests(unittest.TestCase):
""" Test cases for CLI logic """
def test_validate_version_str(self):
self.assertTrue(validate_version_str("0.1.2"))
self.assertTrue(validate_version_str("0.1.2-RC", release_only=False))
self.assertFalse(validate_version_str("0.1"))
self.assertFalse(validate_version_str("0.1.2.3"))
self.assertFalse(validate_version_str("0.1.2-RC"))
def test_split_version_str(self):
self.assertEqual(split_version("0.1.2"), [0, 1, 2])
with self.assertRaises(AttributeError):
split_version("0.1")
def test_assert_verstion_gt(self):
self.assertTrue(version_gt("1.0.0", "0.1.1"))
self.assertTrue(version_gt("0.1.0", "0.0.1"))
self.assertTrue(version_gt("0.0.1", "0.0.0"))
self.assertFalse(version_gt("0.0.0", "0.0.0"))
self.assertFalse(version_gt("0.0.0", "0.0.1"))
self.assertFalse(version_gt("0.0.1", "0.1.0"))
self.assertFalse(version_gt("0.1.1", "1.0.0"))
@@ -0,0 +1,222 @@
import os
import shutil
import unittest
import random
from unittest import mock
import yaml
from local_server.test import FIXTURES_ROOT
def mockenv(**envvars):
return mock.patch.dict(os.environ, envvars)
class ConfigTests(unittest.TestCase):
tmp_fixtures_directory = os.path.join(FIXTURES_ROOT, "tmp_dir")
@classmethod
def tearDownClass(cls) -> None:
shutil.rmtree(cls.tmp_fixtures_directory)
@classmethod
def setUpClass(cls) -> None:
os.makedirs(cls.tmp_fixtures_directory)
def custom_server_config(
self,
verbose="false",
debug="false",
host="localhost",
port="null",
open_browser="false",
force_https="false",
flask_secret_key="secret",
auth_type="session",
insecure_test_environment="false",
index="false",
allowed_matrix_types=[],
max_cached_datasets=5,
timelimit_s=5,
dataset_datapath="null",
obs_names="null",
var_names="null",
about="null",
title="null",
data_locater_region_name="us-east-1",
anndata_backed="false",
column_request_max=32,
diffexp_cellcount_max="null",
config_file_name="server_config.yaml",
):
configfile = os.path.join(self.tmp_fixtures_directory, config_file_name)
server_config_outline_path = os.path.join(FIXTURES_ROOT, "server_config_outline.py")
with open(server_config_outline_path, "r") as config_skeleton:
config = config_skeleton.read()
server_config = eval(config)
with open(configfile, "w") as server_config_file:
server_config_file.write(server_config)
return configfile
def custom_app_config(
self,
verbose="false",
debug="false",
host="localhost",
port="null",
open_browser="false",
force_https="false",
flask_secret_key="secret",
auth_type="session",
index="false",
allowed_matrix_types=[],
max_cached_datasets=5,
timelimit_s=5,
dataset_datapath="null",
obs_names="null",
var_names="null",
about="null",
title="null",
data_locater_region_name="us-east-1",
anndata_backed="false",
column_request_max=32,
diffexp_cellcount_max="null",
scripts=[],
inline_scripts=[],
authentication_enable="true",
max_categories=1000,
custom_colors="true",
enable_users_annotations="true",
annotation_type="local_file_csv",
db_uri="null",
hosted_file_directory="null",
local_file_csv_directory="null",
local_file_csv_file="null",
ontology_enabled="false",
obo_location="null",
embedding_names=[],
enable_reembedding="false",
enable_difexp="true",
lfc_cutoff=0.01,
top_n=10,
environment=None,
aws_secrets_manager_region=None,
aws_secrets_manager_secrets=[],
config_file_name="app_config.yml",
):
random_num = random.randrange(999999)
configfile = os.path.join(self.tmp_fixtures_directory, config_file_name)
server_config = self.custom_server_config(
verbose=verbose,
debug=debug,
host=host,
port=port,
open_browser=open_browser,
force_https=force_https,
flask_secret_key=flask_secret_key,
auth_type=auth_type,
index=index,
allowed_matrix_types=allowed_matrix_types,
max_cached_datasets=max_cached_datasets,
timelimit_s=timelimit_s,
dataset_datapath=dataset_datapath,
obs_names=obs_names,
var_names=var_names,
about=about,
title=title,
data_locater_region_name=data_locater_region_name,
anndata_backed=anndata_backed,
column_request_max=column_request_max,
diffexp_cellcount_max=diffexp_cellcount_max,
config_file_name=f"temp_server_config_{random_num}.yml",
)
dataset_config = self.custom_dataset_config(
scripts=scripts,
inline_scripts=inline_scripts,
authentication_enable=authentication_enable,
max_categories=max_categories,
custom_colors=custom_colors,
enable_users_annotations=enable_users_annotations,
annotation_type=annotation_type,
db_uri=db_uri,
hosted_file_directory=hosted_file_directory,
local_file_csv_directory=local_file_csv_directory,
local_file_csv_file=local_file_csv_file,
ontology_enabled=ontology_enabled,
obo_location=obo_location,
embedding_names=embedding_names,
enable_reembedding=enable_reembedding,
enable_difexp=enable_difexp,
lfc_cutoff=lfc_cutoff,
top_n=top_n,
config_file_name=f"temp_dataset_config_{random_num}.yml",
)
external_config = self.custom_external_config(
environment=environment,
aws_secrets_manager_region=aws_secrets_manager_region,
aws_secrets_manager_secrets=aws_secrets_manager_secrets,
config_file_name=f"temp_external_config_{random_num}.yml",
)
with open(configfile, "w") as app_config_file:
app_config_file.write(open(server_config).read())
app_config_file.write(open(dataset_config).read())
app_config_file.write(open(external_config).read())
return configfile
def custom_dataset_config(
self,
scripts=[],
inline_scripts=[],
authentication_enable="true",
max_categories=1000,
custom_colors="true",
enable_users_annotations="true",
annotation_type="local_file_csv",
db_uri="null",
hosted_file_directory="null",
local_file_csv_directory="null",
local_file_csv_file="null",
ontology_enabled="false",
obo_location="null",
embedding_names=[],
enable_reembedding="false",
enable_difexp="true",
lfc_cutoff=0.01,
top_n=10,
config_file_name="dataset_config.yml",
):
configfile = os.path.join(self.tmp_fixtures_directory, config_file_name)
dataset_config_outline_path = os.path.join(FIXTURES_ROOT, "dataset_config_outline.py")
with open(dataset_config_outline_path, "r") as config_skeleton:
config = config_skeleton.read()
dataset_config = eval(config)
with open(configfile, "w") as dataset_config_file:
dataset_config_file.write(dataset_config)
return configfile
def custom_external_config(
self,
environment=None,
aws_secrets_manager_region=None,
aws_secrets_manager_secrets=[],
config_file_name="external_config.yaml",
):
# set to the default if environment is None
if environment is None:
environment = [
dict(name="CXG_SECRET_KEY", path=["server", "app", "flask_secret_key"], required=False),
]
external_config = {
"external": {
"environment": environment,
"aws_secrets_manager": {"region": aws_secrets_manager_region, "secrets": aws_secrets_manager_secrets},
}
}
configfile = os.path.join(self.tmp_fixtures_directory, config_file_name)
with open(configfile, "w") as external_config_file:
yaml.dump(external_config, external_config_file)
return configfile
@@ -0,0 +1,153 @@
import os
import tempfile
import unittest
import yaml
from local_server.default_config import default_config
from local_server.common.config.app_config import AppConfig
from local_server.test.unit.common.config import ConfigTests
from local_server.common.errors import ConfigurationError
from local_server.test import FIXTURES_ROOT, H5AD_FIXTURE
class AppConfigTest(ConfigTests):
def setUp(self):
self.config_file_name = f"{unittest.TestCase.id(self).split('.')[-1]}.yml"
self.config = AppConfig()
self.config.update_server_config(app__flask_secret_key="secret")
self.config.update_server_config(single_dataset__datapath=H5AD_FIXTURE)
self.server_config = self.config.server_config
self.config.complete_config()
message_list = []
def noop(message):
message_list.append(message)
messagefn = noop
self.context = dict(messagefn=messagefn, messages=message_list)
def get_config(self, **kwargs):
file_name = self.custom_app_config(
dataset_datapath=H5AD_FIXTURE, config_file_name=self.config_file_name, **kwargs
)
config = AppConfig()
config.update_from_config_file(file_name)
return config
def test_get_default_config_correctly_reads_default_config_file(self):
app_default_config = AppConfig().default_config
expected_config = yaml.load(default_config, Loader=yaml.Loader)
server_config = app_default_config["server"]
dataset_config = app_default_config["dataset"]
expected_server_config = expected_config["server"]
expected_dataset_config = expected_config["dataset"]
self.assertDictEqual(app_default_config, expected_config)
self.assertDictEqual(server_config, expected_server_config)
self.assertDictEqual(dataset_config, expected_dataset_config)
def test_get_dataset_config_returns_dataset_config_for_single_datasets(self):
datapath = f"{FIXTURES_ROOT}/1e4dfec4-c0b2-46ad-a04e-ff3ffb3c0a8f.h5ad"
file_name = self.custom_app_config(dataset_datapath=datapath, config_file_name=self.config_file_name)
config = AppConfig()
config.update_from_config_file(file_name)
self.assertEqual(config.get_dataset_config(), config.dataset_config)
def test_update_server_config_updates_server_config_and_config_status(self):
config = self.get_config()
config.complete_config()
config.check_config()
config.update_server_config(single_dataset__datapath=H5AD_FIXTURE)
with self.assertRaises(ConfigurationError):
config.server_config.check_config()
def test_write_config_outputs_yaml_with_all_config_vars(self):
config = self.get_config()
config.write_config(f"{FIXTURES_ROOT}/tmp_dir/write_config.yml")
with open(f"{FIXTURES_ROOT}/tmp_dir/{self.config_file_name}", "r") as default_config:
default_config_yml = yaml.safe_load(default_config)
with open(f"{FIXTURES_ROOT}/tmp_dir/write_config.yml", "r") as output_config:
output_config_yml = yaml.safe_load(output_config)
self.maxDiff = None
self.assertEqual(default_config_yml, output_config_yml)
def test_update_app_config(self):
config = AppConfig()
config.update_server_config(app__verbose=True, single_dataset__datapath="datapath")
vars = config.server_config.changes_from_default()
self.assertCountEqual(vars, [("app__verbose", True, False), ("single_dataset__datapath", "datapath", None)])
config = AppConfig()
config.update_dataset_config(app__scripts=(), app__inline_scripts=())
vars = config.server_config.changes_from_default()
self.assertCountEqual(vars, [])
config = AppConfig()
config.update_dataset_config(app__scripts=[], app__inline_scripts=[])
vars = config.dataset_config.changes_from_default()
self.assertCountEqual(vars, [])
config = AppConfig()
config.update_dataset_config(app__scripts=("a", "b"), app__inline_scripts=["c", "d"])
vars = config.dataset_config.changes_from_default()
self.assertCountEqual(vars, [("app__scripts", ["a", "b"], []), ("app__inline_scripts", ["c", "d"], [])])
def test_configfile_no_server_section(self):
# test a config file without a dataset section
with tempfile.TemporaryDirectory() as tempdir:
configfile = os.path.join(tempdir, "config.yaml")
with open(configfile, "w") as fconfig:
config = """
dataset:
user_annotations:
enable: false
"""
fconfig.write(config)
app_config = AppConfig()
app_config.update_from_config_file(configfile)
server_changes = app_config.server_config.changes_from_default()
dataset_changes = app_config.dataset_config.changes_from_default()
self.assertEqual(server_changes, [])
self.assertEqual(dataset_changes, [("user_annotations__enable", False, True)])
def test_simple_update_single_config_from_path_and_value(self):
"""Update a simple config parameter"""
config = AppConfig()
config.server_config.single_dataset__datapath = "my/data/path"
# test simple value in server
config.update_single_config_from_path_and_value(["server", "app", "flask_secret_key"], "mysecret")
self.assertEqual(config.server_config.app__flask_secret_key, "mysecret")
# test simple value in default dataset
config.update_single_config_from_path_and_value(
["dataset", "user_annotations", "ontology", "obo_location"], "dummy_location",
)
self.assertEqual(config.dataset_config.user_annotations__ontology__obo_location, "dummy_location")
# error checking
bad_paths = [
(
["dataset", "does", "not", "exist"],
"unknown config parameter at path: '['dataset', 'does', 'not', 'exist']'",
),
(["does", "not", "exist"], "path must start with 'server', or 'dataset'"),
([], "path must start with 'server', or 'dataset'"),
([1, 2, 3], "path must be a list of strings, got '[1, 2, 3]'"),
("string", "path must be a list of strings, got 'string'"),
]
for bad_path, error_message in bad_paths:
with self.assertRaises(ConfigurationError) as config_error:
config.update_single_config_from_path_and_value(bad_path, "value")
self.assertEqual(config_error.exception.message, error_message)
@@ -0,0 +1,63 @@
import unittest
from local_server.common.config.app_config import AppConfig
from local_server.test import H5AD_FIXTURE
from local_server.test.unit.common.config import ConfigTests
from local_server.common.errors import ConfigurationError
class BaseConfigTest(ConfigTests):
def setUp(self):
self.config_file_name = f"{unittest.TestCase.id(self).split('.')[-1]}.yml"
self.config = AppConfig()
self.config.update_server_config(app__flask_secret_key="secret")
self.config.update_server_config(single_dataset__datapath=H5AD_FIXTURE)
self.server_config = self.config.server_config
self.config.complete_config()
message_list = []
def noop(message):
message_list.append(message)
messagefn = noop
self.context = dict(messagefn=messagefn, messages=message_list)
def get_config(self, **kwargs):
file_name = self.custom_app_config(
dataset_datapath=f"{H5AD_FIXTURE}", config_file_name=self.config_file_name, **kwargs
)
config = AppConfig()
config.update_from_config_file(file_name)
return config
def test_mapping_creation_returns_map_of_server_and_dataset_config(self):
config = AppConfig()
mapping = config.dataset_config.create_mapping(config.default_config)
self.assertIsNotNone(mapping["server__app__verbose"])
self.assertIsNotNone(mapping["dataset__presentation__max_categories"])
self.assertIsNotNone(mapping["dataset__user_annotations__ontology__obo_location"])
def test_changes_from_default_returns_list_of_nondefault_config_values(self):
config = self.get_config(verbose="true", lfc_cutoff=0.05)
server_changes = config.server_config.changes_from_default()
dataset_changes = config.dataset_config.changes_from_default()
self.assertEqual(
server_changes,
[
("app__verbose", True, False),
("app__flask_secret_key", "secret", None),
("single_dataset__datapath", H5AD_FIXTURE, None),
('data_locator__s3__region_name', 'us-east-1', True)
],
)
self.assertEqual(dataset_changes, [("diffexp__lfc_cutoff", 0.05, 0.01)])
def test_check_config_throws_error_if_attr_has_not_been_checked(self):
config = self.get_config(verbose="true")
config.complete_config()
config.check_config()
config.update_server_config(app__verbose=False)
with self.assertRaises(ConfigurationError):
config.check_config()
@@ -0,0 +1,157 @@
import os
import tempfile
import unittest
from unittest.mock import patch
from local_server.common.annotations.local_file_csv import AnnotationsLocalFile
from local_server.common.config.app_config import AppConfig
from local_server.common.config.base_config import BaseConfig
from local_server.test import FIXTURES_ROOT, H5AD_FIXTURE
from local_server.common.errors import ConfigurationError
from local_server.test.unit.common.config import ConfigTests
class TestDatasetConfig(ConfigTests):
def setUp(self):
self.config_file_name = f"{unittest.TestCase.id(self).split('.')[-1]}.yml"
self.config = AppConfig()
self.config.update_server_config(app__flask_secret_key="secret")
self.config.update_server_config(single_dataset__datapath=H5AD_FIXTURE)
self.dataset_config = self.config.dataset_config
self.config.complete_config()
message_list = []
def noop(message):
message_list.append(message)
messagefn = noop
self.context = dict(messagefn=messagefn, messages=message_list)
def get_config(self, **kwargs):
file_name = self.custom_app_config(dataset_datapath=H5AD_FIXTURE, **kwargs)
config = AppConfig()
config.update_from_config_file(file_name)
return config
def test_init_datatset_config_sets_vars_from_config(self):
config = AppConfig()
self.assertEqual(config.dataset_config.presentation__max_categories, 1000)
self.assertEqual(config.dataset_config.user_annotations__type, "local_file_csv")
self.assertEqual(config.dataset_config.diffexp__lfc_cutoff, 0.01)
self.assertIsNone(config.dataset_config.user_annotations__ontology__obo_location)
@patch("local_server.common.config.dataset_config.BaseConfig.validate_correct_type_of_configuration_attribute")
def test_complete_config_checks_all_attr(self, mock_check_attrs):
mock_check_attrs.side_effect = BaseConfig.validate_correct_type_of_configuration_attribute()
self.dataset_config.complete_config(self.context)
self.assertIsNotNone(self.config.server_config.data_adaptor)
self.assertEqual(mock_check_attrs.call_count, 17)
def test_app_sets_script_vars(self):
config = self.get_config(scripts=["path/to/script"])
config.dataset_config.handle_app()
self.assertEqual(config.dataset_config.app__scripts, [{"src": "path/to/script"}])
config = self.get_config(scripts=[{"src": "path/to/script", "more": "different/script/path"}])
config.dataset_config.handle_app()
self.assertEqual(
config.dataset_config.app__scripts, [{"src": "path/to/script", "more": "different/script/path"}]
)
config = self.get_config(scripts=["path/to/script", "different/script/path"])
config.dataset_config.handle_app()
# TODO @madison -- is this the desired functionality?
self.assertEqual(
config.dataset_config.app__scripts, [{"src": "path/to/script"}, {"src": "different/script/path"}]
)
config = self.get_config(scripts=[{"more": "different/script/path"}])
with self.assertRaises(ConfigurationError):
config.dataset_config.handle_app()
def test_handle_user_annotations_ensures_auth_is_enabled_with_valid_auth_type(self):
config = self.get_config(enable_users_annotations="true", authentication_enable="false")
config.server_config.complete_config(self.context)
with self.assertRaises(ConfigurationError):
config.dataset_config.handle_user_annotations(self.context)
config = self.get_config(enable_users_annotations="true", authentication_enable="true", auth_type="pretend")
with self.assertRaises(ConfigurationError):
config.server_config.complete_config(self.context)
def test_handle_user_annotations__instantiates_user_annotations_class_correctly(self):
config = self.get_config(
enable_users_annotations="true", authentication_enable="true", annotation_type="local_file_csv"
)
config.server_config.complete_config(self.context)
config.dataset_config.handle_user_annotations(self.context)
self.assertIsInstance(config.dataset_config.user_annotations, AnnotationsLocalFile)
config = self.get_config(
enable_users_annotations="true", authentication_enable="true", annotation_type="NOT_REAL"
)
config.server_config.complete_config(self.context)
with self.assertRaises(ConfigurationError):
config.dataset_config.handle_user_annotations(self.context)
def test_handle_local_file_csv_annotations__sets_dir_if_not_passed_in(self):
config = self.get_config(
enable_users_annotations="true", authentication_enable="true", annotation_type="local_file_csv"
)
config.server_config.complete_config(self.context)
config.dataset_config.handle_local_file_csv_annotations()
self.assertIsInstance(config.dataset_config.user_annotations, AnnotationsLocalFile)
cwd = os.getcwd()
self.assertEqual(config.dataset_config.user_annotations._get_output_dir(), cwd)
def test_handle_embeddings__checks_data_file_types(self):
file_name = self.custom_app_config(
embedding_names=["name1", "name2"],
enable_reembedding="true",
dataset_datapath=f"{FIXTURES_ROOT}/pbmc3k-CSC-gz.h5ad",
anndata_backed="true",
config_file_name=self.config_file_name,
)
config = AppConfig()
config.update_from_config_file(file_name)
config.server_config.complete_config(self.context)
with self.assertRaises(ConfigurationError):
config.dataset_config.handle_embeddings()
def test_handle_diffexp__raises_warning_for_large_datasets(self):
config = self.get_config(lfc_cutoff=0.02, enable_difexp="true", top_n=15)
config.server_config.complete_config(self.context)
config.dataset_config.handle_diffexp(self.context)
self.assertEqual(len(self.context["messages"]), 1)
def test_configfile_with_specialization(self):
# test that per_dataset_config config load the default config, then the specialized config
with tempfile.TemporaryDirectory() as tempdir:
configfile = os.path.join(tempdir, "config.yaml")
with open(configfile, "w") as fconfig:
config = """
server:
single_dataset:
datapath: fake_datapath
dataset:
user_annotations:
enable: false
type: local_file_csv
local_file_csv:
file: fake_file
directory: fake_dir
"""
fconfig.write(config)
app_config = AppConfig()
app_config.update_from_config_file(configfile)
test_config = app_config.dataset_config
# test config from default
self.assertEqual(test_config.user_annotations__type, "local_file_csv")
self.assertEqual(test_config.user_annotations__local_file_csv__file, "fake_file")
@@ -0,0 +1,215 @@
import os
from unittest.mock import patch
import requests
from local_server.common.errors import ConfigurationError
from local_server.common.config.app_config import AppConfig
from local_server.test import test_server, FIXTURES_ROOT
from local_server.common.utils.type_conversion_utils import convert_string_to_value
from local_server.test.unit.common.config import ConfigTests
class TestExternalConfig(ConfigTests):
def test_type_convert(self):
# The values from environment variables and aws secrets are returned as strings.
# These values need to be converted to the proper types.
self.assertEqual(convert_string_to_value("1"), int(1))
self.assertEqual(convert_string_to_value("1.1"), float(1.1))
self.assertEqual(convert_string_to_value("string"), "string")
self.assertEqual(convert_string_to_value("true"), True)
self.assertEqual(convert_string_to_value("True"), True)
self.assertEqual(convert_string_to_value("false"), False)
self.assertEqual(convert_string_to_value("False"), False)
self.assertEqual(convert_string_to_value("null"), None)
self.assertEqual(convert_string_to_value("None"), None)
self.assertEqual(convert_string_to_value("{'a':10, 'b':'string'}"), dict(a=int(10), b="string"))
def test_environment_variable(self):
configfile = self.custom_external_config(
environment=[
dict(name="DATAPATH", path=["server", "single_dataset", "datapath"], required=True),
dict(name="DIFFEXP", path=["dataset", "diffexp", "enable"], required=True),
],
config_file_name="environment_external_config.yaml",
)
env = os.environ
env["DATAPATH"] = f"{FIXTURES_ROOT}/pbmc3k-CSC-gz.h5ad"
env["DIFFEXP"] = "False"
with test_server(command_line_args=["-c", configfile], env=env) as server:
session = requests.Session()
response = session.get(f"{server}/api/v0.2/config")
data_config = response.json()
self.assertEqual(data_config["config"]["displayNames"]["dataset"], "pbmc3k-CSC-gz")
self.assertTrue(data_config["config"]["parameters"]["disable-diffexp"])
env["DATAPATH"] = f"{FIXTURES_ROOT}/a95c59b4-7f5d-4b80-ad53-a694834ca18b.h5ad"
env["DIFFEXP"] = "True"
with test_server(command_line_args=["-c", configfile], env=env) as server:
session = requests.Session()
response = session.get(f"{server}/api/v0.2/config")
data_config = response.json()
self.assertEqual(data_config["config"]["displayNames"]["dataset"], "a95c59b4-7f5d-4b80-ad53-a694834ca18b")
self.assertFalse(data_config["config"]["parameters"]["disable-diffexp"])
def test_environment_variable_errors(self):
# no name
app_config = AppConfig()
app_config.external_config.environment = [dict(required=True, path=["this", "is", "a", "path"])]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "environment: 'name' is missing")
# required has wrong type
app_config = AppConfig()
app_config.external_config.environment = [
dict(name="myenvar", required="optional", path=["this", "is", "a", "path"])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "environment: 'required' must be a bool")
# no path
app_config = AppConfig()
app_config.external_config.environment = [dict(name="myenvar", required=True)]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "environment: 'path' is missing")
# required environment variable is not set
app_config = AppConfig()
app_config.external_config.environment = [
dict(name="THIS_ENV_IS_NOT_SET", required=True, path=["this", "is", "a", "path"])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "required environment variable 'THIS_ENV_IS_NOT_SET' not set")
@patch("local_server.common.config.external_config.get_secret_key")
def test_aws_secrets_manager(self, mock_get_secret_key):
mock_get_secret_key.return_value = {
"flask_secret_key": "mock_flask_secret_key",
}
configfile = self.custom_external_config(
aws_secrets_manager_region="us-west-2",
aws_secrets_manager_secrets=[
dict(
name="my_secret",
values=[
dict(key="flask_secret_key", path=["server", "app", "flask_secret_key"], required=True),
],
)
],
config_file_name="secret_external_config.yaml",
)
app_config = AppConfig()
app_config.update_from_config_file(configfile)
app_config.server_config.single_dataset__datapath = f"{FIXTURES_ROOT}/pbmc3k-CSC-gz.h5ad"
app_config.complete_config()
self.assertEqual(app_config.server_config.app__flask_secret_key, "mock_flask_secret_key")
@patch("local_server.common.config.external_config.get_secret_key")
def test_aws_secrets_manager_error(self, mock_get_secret_key):
mock_get_secret_key.return_value = {
"db_uri": "mock_db_uri",
}
# no region
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = None
app_config.external_config.aws_secrets_manager__secrets = [
dict(name="secret1", values=[dict(key="key1", required=True, path=["this", "is", "my", "path"])])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(
config_error.exception.message,
"Invalid type for attribute: aws_secrets_manager__region, expected type str, got NoneType",
)
# missing secret name
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [
dict(values=[dict(key="db_uri", required=True, path=["this", "is", "my", "path"])])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "aws_secrets_manager: 'name' is missing")
# secret name wrong type
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [
dict(name=1, values=[dict(key="db_uri", required=True, path=["this", "is", "my", "path"])])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "aws_secrets_manager: 'name' must be a string")
# missing values name
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [dict(name="mysecret")]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "aws_secrets_manager: 'values' is missing")
# values wrong type
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [
dict(name="mysecret", values=dict(key="db_uri", required=True, path=["this", "is", "my", "path"]))
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "aws_secrets_manager: 'values' must be a list")
# entry missing key
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [
dict(name="mysecret", values=[dict(required=True, path=["this", "is", "my", "path"])])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "missing 'key' in secret values: mysecret")
# entry required is wrong type
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [
dict(name="mysecret", values=[dict(key="db_uri", required="optional", path=["this", "is", "my", "path"])])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "wrong type for 'required' in secret values: mysecret")
# entry missing path
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [
dict(name="mysecret", values=[dict(key="db_uri", required=True)])
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "missing 'path' in secret values: mysecret")
# secret missing required key
app_config = AppConfig()
app_config.external_config.aws_secrets_manager__region = "us-west-2"
app_config.external_config.aws_secrets_manager__secrets = [
dict(
name="mysecret",
values=[dict(key="KEY_DOES_NOT_EXIST", required=True, path=["this", "is", "a", "path"])],
)
]
with self.assertRaises(ConfigurationError) as config_error:
app_config.complete_config()
self.assertEqual(config_error.exception.message, "required secret 'mysecret:KEY_DOES_NOT_EXIST' not set")
@@ -0,0 +1,122 @@
import os
import unittest
from unittest import mock
from unittest.mock import patch
from local_server.common.config.base_config import BaseConfig
from local_server.test import H5AD_FIXTURE
from local_server.common.config.app_config import AppConfig
from local_server.common.errors import ConfigurationError
from local_server.test.unit.common.config import ConfigTests
def mockenv(**envvars):
return mock.patch.dict(os.environ, envvars)
class TestServerConfig(ConfigTests):
def setUp(self):
self.config_file_name = f"{unittest.TestCase.id(self).split('.')[-1]}.yml"
self.config = AppConfig()
self.config.update_server_config(app__flask_secret_key="secret")
self.config.update_server_config(single_dataset__datapath=H5AD_FIXTURE)
self.server_config = self.config.server_config
self.config.complete_config()
message_list = []
def noop(message):
message_list.append(message)
messagefn = noop
self.context = dict(messagefn=messagefn, messages=message_list)
def get_config(self, **kwargs):
file_name = self.custom_app_config(
dataset_datapath=f"{H5AD_FIXTURE}", config_file_name=self.config_file_name, **kwargs
)
config = AppConfig()
config.update_from_config_file(file_name)
return config
def test_init_raises_error_if_default_config_is_invalid(self):
invalid_config = self.get_config(port="not_valid")
with self.assertRaises(ConfigurationError):
invalid_config.complete_config()
@patch("local_server.common.config.server_config.BaseConfig.validate_correct_type_of_configuration_attribute")
def test_complete_config_checks_all_attr(self, mock_check_attrs):
mock_check_attrs.side_effect = BaseConfig.validate_correct_type_of_configuration_attribute()
self.server_config.complete_config(self.context)
self.assertEqual(mock_check_attrs.call_count, 20)
def test_handle_app__throws_error_if_port_doesnt_exist(self):
config = self.get_config(port=99999999)
with self.assertRaises(ConfigurationError):
config.server_config.handle_app(self.context)
@patch("local_server.common.config.server_config.discover_s3_region_name")
def test_handle_data_locator_works_for_default_types(self, mock_discover_region_name):
mock_discover_region_name.return_value = None
# Default config
self.assertEqual(self.config.server_config.data_locator__s3__region_name, None)
# hard coded
config = self.get_config()
self.assertEqual(config.server_config.data_locator__s3__region_name, "us-east-1")
# incorrectly formatted
datapath = "s3://shouldnt/work"
file_name = self.custom_app_config(
dataset_datapath=datapath, config_file_name=self.config_file_name, data_locater_region_name="true"
)
config = AppConfig()
config.update_from_config_file(file_name)
with self.assertRaises(ConfigurationError):
config.server_config.handle_data_locator()
def test_handle_app___can_use_envar_port(self):
config = self.get_config(port=24)
self.assertEqual(config.server_config.app__port, 24)
# Note if the port is set in the config file it will NOT be overwritten by a different envvar
os.environ["CXG_SERVER_PORT"] = "4008"
self.config = AppConfig()
self.config.update_server_config(app__flask_secret_key="secret")
self.config.server_config.handle_app(self.context)
self.assertEqual(self.config.server_config.app__port, 4008)
del os.environ["CXG_SERVER_PORT"]
def test_handle_app__can_get_secret_key_from_envvar_or_config_file_with_envvar_given_preference(self):
config = self.get_config(flask_secret_key="KEY_FROM_FILE")
self.assertEqual(config.server_config.app__flask_secret_key, "KEY_FROM_FILE")
os.environ["CXG_SECRET_KEY"] = "KEY_FROM_ENV"
config.external_config.handle_environment(self.context)
self.assertEqual(config.server_config.app__flask_secret_key, "KEY_FROM_ENV")
def test_config_for_single_dataset(self):
file_name = self.custom_app_config(
config_file_name="single_dataset.yml", dataset_datapath=f"{H5AD_FIXTURE}"
)
config = AppConfig()
config.update_from_config_file(file_name)
config.server_config.handle_single_dataset(self.context)
file_name = self.custom_app_config(
config_file_name="single_dataset_with_about.yml",
about="www.cziscience.com",
dataset_datapath=f"{H5AD_FIXTURE}",
)
config = AppConfig()
config.update_from_config_file(file_name)
with self.assertRaises(ConfigurationError):
config.server_config.handle_single_dataset(self.context)
def test_test_auth_only_in_insecure(self):
config = self.get_config(auth_type="test")
with self.assertRaises(ConfigurationError):
config.complete_config()
config.update_server_config(authentication__insecure_test_environment=True)
config.complete_config()
+422
View File
@@ -0,0 +1,422 @@
import shutil
import time
import unittest
import zlib
from http import HTTPStatus
import pandas as pd
import requests
import local_server.test.unit.decode_fbs as decode_fbs
from local_server.data_common.matrix_loader import MatrixDataType
from local_server.test import (
data_with_tmp_annotations,
make_fbs,
PROJECT_ROOT,
start_test_server,
stop_test_server,
)
from local_server.test.fixtures.fixtures import pbmc3k_colors
BAD_FILTER = {"filter": {"obs": {"annotation_value": [{"name": "xyz"}]}}}
# TODO (mweiden): remove ANNOTATIONS_ENABLED and Annotation subclasses when annotations are no longer experimental
class EndPoints(object):
ANNOTATIONS_ENABLED = True
def test_initialize(self):
endpoint = "schema"
url = f"{self.URL_BASE}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
result_data = result.json()
self.assertEqual(result_data["schema"]["dataframe"]["nObs"], 2638)
self.assertEqual(len(result_data["schema"]["annotations"]["obs"]), 2)
self.assertEqual(
len(result_data["schema"]["annotations"]["obs"]["columns"]), 6 if self.ANNOTATIONS_ENABLED else 5
)
def test_config(self):
endpoint = "config"
url = f"{self.URL_BASE}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
result_data = result.json()
self.assertIn("library_versions", result_data["config"])
self.assertEqual(result_data["config"]["displayNames"]["dataset"], "pbmc3k")
def test_get_layout_fbs(self):
endpoint = "layout/obs"
url = f"{self.URL_BASE}{endpoint}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 8)
self.assertIsNotNone(df["columns"])
self.assertSetEqual(
set(df["col_idx"]),
{"pca_0", "pca_1", "tsne_0", "tsne_1", "umap_0", "umap_1", "draw_graph_fr_0", "draw_graph_fr_1"},
)
self.assertIsNone(df["row_idx"])
self.assertEqual(len(df["columns"]), df["n_cols"])
def test_put_layout_fbs(self):
# first check that re-embedding is turned on
result = self.session.get(f"{self.URL_BASE}config")
config_data = result.json()
re_embed = config_data["config"]["parameters"]["enable-reembedding"]
if not re_embed:
return
# attempt to reembed with umap over 100 cells.
endpoint = "layout/obs"
url = f"{self.URL_BASE}{endpoint}"
data = {}
data["filter"] = {}
data["filter"]["obs"] = {}
data["filter"]["obs"]["index"] = list(range(100))
data["method"] = "umap"
result = self.session.put(url, json=data)
self.assertEqual(result.status_code, HTTPStatus.OK)
result_data = result.json()
self.assertIsInstance(result_data, dict)
self.assertEqual(result_data["type"], "float32")
self.assertTrue(result_data["name"].startswith("reembed:umap_"))
self.assertIsInstance(result_data["dims"], list)
self.assertEqual(len(result_data["dims"]), 2)
dims = result_data["dims"]
self.assertTrue(dims[0].startswith("reembed:umap_") and dims[0].endswith("_0"))
self.assertTrue(dims[1].startswith("reembed:umap_") and dims[1].endswith("_1"))
def test_bad_filter(self):
endpoint = "data/var"
url = f"{self.URL_BASE}{endpoint}"
result = self.session.put(url, json=BAD_FILTER)
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
def test_get_annotations_obs_fbs(self):
endpoint = "annotations/obs"
url = f"{self.URL_BASE}{endpoint}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 6 if self.ANNOTATIONS_ENABLED else 5)
self.assertIsNotNone(df["columns"])
self.assertIsNone(df["row_idx"])
self.assertEqual(len(df["columns"]), df["n_cols"])
obs_index_col_name = self.schema["schema"]["annotations"]["obs"]["index"]
self.assertCountEqual(
df["col_idx"],
[obs_index_col_name, "n_genes", "percent_mito", "n_counts", "louvain"]
+ (["cluster-test"] if self.ANNOTATIONS_ENABLED else []),
)
def test_get_annotations_obs_keys_fbs(self):
endpoint = "annotations/obs"
query = "annotation-name=n_genes&annotation-name=percent_mito"
url = f"{self.URL_BASE}{endpoint}?{query}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 2)
self.assertIsNotNone(df["columns"])
self.assertIsNone(df["row_idx"])
self.assertEqual(len(df["columns"]), df["n_cols"])
self.assertCountEqual(df["col_idx"], ["n_genes", "percent_mito"])
def test_get_annotations_obs_error(self):
endpoint = "annotations/obs"
query = "annotation-name=notakey"
url = f"{self.URL_BASE}{endpoint}?{query}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
def test_get_annotations_var_fbs(self):
endpoint = "annotations/var"
url = f"{self.URL_BASE}{endpoint}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 1838)
self.assertEqual(df["n_cols"], 2)
self.assertIsNotNone(df["columns"])
self.assertIsNone(df["row_idx"])
self.assertEqual(len(df["columns"]), df["n_cols"])
var_index_col_name = self.schema["schema"]["annotations"]["var"]["index"]
self.assertCountEqual(df["col_idx"], [var_index_col_name, "n_cells"])
def test_get_annotations_var_keys_fbs(self):
endpoint = "annotations/var"
query = "annotation-name=n_cells"
url = f"{self.URL_BASE}{endpoint}?{query}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 1838)
self.assertEqual(df["n_cols"], 1)
self.assertIsNotNone(df["columns"])
self.assertIsNone(df["row_idx"])
self.assertEqual(len(df["columns"]), df["n_cols"])
self.assertCountEqual(df["col_idx"], ["n_cells"])
def test_get_annotations_var_error(self):
endpoint = "annotations/var"
query = "annotation-name=notakey"
url = f"{self.URL_BASE}{endpoint}?{query}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
def test_data_mimetype_error(self):
endpoint = "data/var"
header = {"Accept": "xxx"}
url = f"{self.URL_BASE}{endpoint}"
result = self.session.put(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.NOT_ACCEPTABLE)
def test_fbs_default(self):
endpoint = "data/var"
url = f"{self.URL_BASE}{endpoint}"
result = self.session.put(url)
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
filter = {"filter": {"var": {"index": [0, 1, 4]}}}
result = self.session.put(url, json=filter)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
def test_data_put_fbs(self):
endpoint = "data/var"
url = f"{self.URL_BASE}{endpoint}"
header = {"Accept": "application/octet-stream"}
result = self.session.put(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
def test_data_get_fbs(self):
endpoint = "data/var"
url = f"{self.URL_BASE}{endpoint}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
def test_data_put_filter_fbs(self):
endpoint = "data/var"
url = f"{self.URL_BASE}{endpoint}"
header = {"Accept": "application/octet-stream"}
filter = {"filter": {"var": {"index": [0, 1, 4]}}}
result = self.session.put(url, headers=header, json=filter)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 3)
self.assertIsNotNone(df["columns"])
self.assertIsNone(df["row_idx"])
self.assertEqual(len(df["columns"]), df["n_cols"])
self.assertListEqual(df["col_idx"].tolist(), [0, 1, 4])
def test_data_get_filter_fbs(self):
index_col_name = self.schema["schema"]["annotations"]["var"]["index"]
endpoint = "data/var"
query = f"var:{index_col_name}=SIK1"
url = f"{self.URL_BASE}{endpoint}?{query}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 1)
def test_data_get_unknown_filter_fbs(self):
index_col_name = self.schema["schema"]["annotations"]["var"]["index"]
endpoint = "data/var"
query = f"var:{index_col_name}=UNKNOWN"
url = f"{self.URL_BASE}{endpoint}?{query}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 0)
def test_data_put_single_var(self):
endpoint = "data/var"
url = f"{self.URL_BASE}{endpoint}"
header = {"Accept": "application/octet-stream"}
index_col_name = self.schema["schema"]["annotations"]["var"]["index"]
var_filter = {"filter": {"var": {"annotation_value": [{"name": index_col_name, "values": ["RER1"]}]}}}
result = self.session.put(url, headers=header, json=var_filter)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 1)
def test_colors(self):
endpoint = "colors"
url = f"{self.URL_BASE}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
result_data = result.json()
self.assertEqual(result_data, pbmc3k_colors)
def test_static(self):
endpoint = "static"
file = "assets/favicon.ico"
url = f"{self.server}/{endpoint}/{file}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
def _setupClass(child_class, command_line):
child_class.ps, child_class.server = start_test_server(command_line)
child_class.URL_BASE = f"{child_class.server}/api/v0.2/"
child_class.session = requests.Session()
for i in range(90):
try:
result = child_class.session.get(f"{child_class.URL_BASE}schema")
child_class.schema = result.json()
except requests.exceptions.ConnectionError:
time.sleep(1)
class EndPointsAnnotations(EndPoints):
def test_get_schema_existing_writable(self):
self._test_get_schema_writable("cluster-test")
def test_get_user_annotations_existing_obs_keys_fbs(self):
self._test_get_user_annotations_obs_keys_fbs(
"cluster-test", {"unassigned", "one", "two", "three", "four", "five", "six", "seven"},
)
def test_put_user_annotations_obs_fbs(self):
endpoint = "annotations/obs"
query = "annotation-collection-name=test_annotations"
url = f"{self.URL_BASE}{endpoint}?{query}"
n_rows = self.data.get_shape()[0]
fbs = make_fbs({"cat_A": pd.Series(["label_A"] * n_rows, dtype="category")})
result = self.session.put(url, data=zlib.compress(fbs))
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
self.assertEqual(result.json(), {"status": "OK"})
self._test_get_schema_writable("cat_A")
self._test_get_user_annotations_obs_keys_fbs("cat_A", {"label_A"})
def _test_get_user_annotations_obs_keys_fbs(self, annotation_name, columns):
endpoint = "annotations/obs"
query = f"annotation-name={annotation_name}"
url = f"{self.URL_BASE}{endpoint}?{query}"
header = {"Accept": "application/octet-stream"}
result = self.session.get(url, headers=header)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertEqual(df["n_rows"], 2638)
self.assertEqual(df["n_cols"], 1)
self.assertListEqual(df["col_idx"], [annotation_name])
self.assertEqual(set(df["columns"][0]), columns)
self.assertIsNone(df["row_idx"])
self.assertEqual(len(df["columns"]), df["n_cols"])
def _test_get_schema_writable(self, cluster_name):
endpoint = "schema"
url = f"{self.URL_BASE}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
result_data = result.json()
columns = result_data["schema"]["annotations"]["obs"]["columns"]
matching_columns = [c for c in columns if c["name"] == cluster_name]
self.assertEqual(len(matching_columns), 1)
self.assertTrue(matching_columns[0]["writable"])
class EndPointsAnndata(unittest.TestCase, EndPoints):
"""Test Case for endpoints"""
ANNOTATIONS_ENABLED = False
@classmethod
def setUpClass(cls):
cls._setupClass(
cls,
[
f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad",
"--disable-annotations",
"--experimental-enable-reembedding",
],
)
@classmethod
def tearDownClass(cls):
stop_test_server(cls.ps)
@property
def annotations_enabled(self):
return False
def test_diff_exp(self):
endpoint = "diffexp/obs"
url = f"{self.URL_BASE}{endpoint}"
params = {
"mode": "topN",
"set1": {"filter": {"obs": {"annotation_value": [{"name": "louvain", "values": ["NK cells"]}]}}},
"set2": {"filter": {"obs": {"annotation_value": [{"name": "louvain", "values": ["CD8 T cells"]}]}}},
"count": 7,
}
result = self.session.post(url, json=params)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
result_data = result.json()
self.assertEqual(len(result_data), 7)
def test_diff_exp_indices(self):
endpoint = "diffexp/obs"
url = f"{self.URL_BASE}{endpoint}"
params = {
"mode": "topN",
"count": 10,
"set1": {"filter": {"obs": {"index": [[0, 500]]}}},
"set2": {"filter": {"obs": {"index": [[500, 1000]]}}},
}
result = self.session.post(url, json=params)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
result_data = result.json()
self.assertEqual(len(result_data), 10)
class EndPointsAnndataAnnotations(unittest.TestCase, EndPointsAnnotations):
"""Test Case for endpoints"""
ANNOTATIONS_ENABLED = True
@classmethod
def setUpClass(cls):
cls.data, cls.tmp_dir, cls.annotations = data_with_tmp_annotations(
MatrixDataType.H5AD, annotations_fixture=True
)
cls._setupClass(cls, ["--annotations-file", cls.annotations.output_file, cls.data.get_location()])
@classmethod
def tearDownClass(cls):
shutil.rmtree(cls.tmp_dir)
stop_test_server(cls.ps)
@@ -0,0 +1,40 @@
import unittest
import anndata
from local_server.common.colors import convert_color_to_hex_format, convert_anndata_category_colors_to_cxg_category_colors
from local_server.common.errors import ColorFormatException
from local_server.test import PROJECT_ROOT
from local_server.test.fixtures.fixtures import pbmc3k_colors
class ColorsTest(unittest.TestCase):
""" Test color helper functions """
def test_convert_color_to_hex_format(self):
self.assertEqual(convert_color_to_hex_format("wheat"), "#f5deb3")
self.assertEqual(convert_color_to_hex_format("WHEAT"), "#f5deb3")
self.assertEqual(convert_color_to_hex_format((245, 222, 179)), "#f5deb3")
self.assertEqual(convert_color_to_hex_format([245, 222, 179]), "#f5deb3")
self.assertEqual(convert_color_to_hex_format("#f5deb3"), "#f5deb3")
self.assertEqual(
convert_color_to_hex_format([0.9607843137254902, 0.8705882352941177, 0.7019607843137254]), "#f5deb3"
)
for bad_input in ["foo", "BAR", "#AABB", "#AABBCCDD", "#AABBGG", (1, 2), [1, 2], (1, 2, 3, 4), [1, 2, 3, 4]]:
with self.assertRaises(ColorFormatException):
convert_color_to_hex_format(bad_input)
def test_anndata_colors_to_cxg_colors(self):
# test standard behavior
adata = self._get_h5ad()
self.assertEqual(convert_anndata_category_colors_to_cxg_category_colors(adata), pbmc3k_colors)
# test that invalid color formats raise an exception
adata.uns["louvain_colors"][0] = "#NOTCOOL"
with self.assertRaises(ColorFormatException):
convert_anndata_category_colors_to_cxg_category_colors(adata)
# test that colors without a matching obs category are skipped
adata = self._get_h5ad()
del adata.obs["louvain"]
self.assertEqual(convert_anndata_category_colors_to_cxg_category_colors(adata), {})
def _get_h5ad(self):
return anndata.read_h5ad(f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad")
@@ -0,0 +1,164 @@
import json
import shutil
import tempfile
import unittest
from http import HTTPStatus
import anndata
import requests
from local_server.common.corpora import (
corpora_get_versions_from_anndata,
corpora_is_version_supported,
corpora_get_props_from_anndata,
)
from local_server.test import PROJECT_ROOT, start_test_server, stop_test_server
VERSION = "v0.2"
class CorporaAPITest(unittest.TestCase):
def test_corpora_get_versions_from_anndata(self):
adata = self._get_h5ad()
if "version" in adata.uns:
del adata.uns["version"]
self.assertIsNone(corpora_get_versions_from_anndata(adata))
# something bogus
adata.uns["version"] = 99
self.assertIsNone(corpora_get_versions_from_anndata(adata))
# something legit
adata.uns["version"] = {"corpora_schema_version": "0.0.0", "corpora_encoding_version": "9.9.9"}
self.assertEqual(corpora_get_versions_from_anndata(adata), ["0.0.0", "9.9.9"])
def test_corpora_is_version_supported(self):
self.assertTrue(corpora_is_version_supported("1.0.0", "0.1.0"))
self.assertFalse(corpora_is_version_supported("0.0.0", "0.1.0"))
self.assertFalse(corpora_is_version_supported("1.0.0", "0.0.0"))
def test_corpora_get_props_from_anndata(self):
adata = self._get_h5ad()
if "version" in adata.uns:
del adata.uns["version"]
self.assertIsNone(corpora_get_props_from_anndata(adata))
# something bogus
adata.uns["version"] = 99
self.assertIsNone(corpora_get_props_from_anndata(adata))
# unsupported version, but missing required values
adata.uns["version"] = {"corpora_schema_version": "99.0.0", "corpora_encoding_version": "32.1.0"}
with self.assertRaises(ValueError):
corpora_get_props_from_anndata(adata)
# legit version, but missing required values
adata.uns["version"] = {"corpora_schema_version": "1.0.0", "corpora_encoding_version": "0.1.0"}
with self.assertRaises(KeyError):
corpora_get_props_from_anndata(adata)
some_fields = {
"version": {"corpora_schema_version": "1.0.0", "corpora_encoding_version": "0.1.0"},
"title": "title",
"layer_descriptions": "layer_descriptions",
"organism": "organism",
"organism_ontology_term_id": "organism_ontology_term_id",
"project_name": "project_name",
"project_description": "project_description",
"contributors": json.dumps([{"contributors": "contributors"}]),
"project_links": json.dumps([{"link_name": "link_name", "link_url": "link_url", "link_type": "SUMMARY"}]),
}
for k in some_fields:
adata.uns[k] = some_fields[k]
some_fields["contributors"] = json.loads(some_fields["contributors"])
some_fields["project_links"] = json.loads(some_fields["project_links"])
self.assertEqual(corpora_get_props_from_anndata(adata), some_fields)
def test_corpora_get_props_from_anndata_v110(self):
adata = self._get_h5ad()
if "version" in adata.uns:
del adata.uns["version"]
self.assertIsNone(corpora_get_props_from_anndata(adata))
# legit version, but missing required values
adata.uns["version"] = {"corpora_schema_version": "1.1.0", "corpora_encoding_version": "0.1.0"}
with self.assertRaises(KeyError):
corpora_get_props_from_anndata(adata)
# Metadata following schema 1.1.0, which removes some fields relative to 1.1.0
some_110_fields = {
"version": {"corpora_schema_version": "1.0.0", "corpora_encoding_version": "0.1.0"},
"title": "title",
"layer_descriptions": "layer_descriptions",
"organism": "organism",
"organism_ontology_term_id": "organism_ontology_term_id",
}
for k in some_110_fields:
adata.uns[k] = some_110_fields[k]
self.assertEqual(corpora_get_props_from_anndata(adata), some_110_fields)
def _get_h5ad(self):
return anndata.read_h5ad(f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad")
class CorporaRESTAPITest(unittest.TestCase):
""" Confirm endpoints reflect Corpora-specific features """
@classmethod
def setCorporaFields(cls, path):
adata = anndata.read_h5ad(path)
corpora_props = {
"version": {"corpora_schema_version": "1.0.0", "corpora_encoding_version": "0.1.0"},
"title": "PBMC3K",
"contributors": json.dumps([{"name": "name"}]),
"layer_descriptions": {"X": "raw counts"},
"organism": "human",
"organism_ontology_term_id": "unknown",
"project_name": "test project",
"project_description": "test description",
"project_links": json.dumps(
[{"link_name": "test link", "link_type": "SUMMARY", "link_url": "https://a.u.r.l/"}]
),
"default_embedding": "X_tsne",
}
adata.uns.update(corpora_props)
adata.write(path)
@classmethod
def setUpClass(cls):
cls.tmp_dir = tempfile.TemporaryDirectory()
src = f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad"
dst = f"{cls.tmp_dir.name}/pbmc3k.h5ad"
shutil.copyfile(src, dst)
cls.setCorporaFields(dst)
cls.ps, cls.server = start_test_server([dst])
@classmethod
def tearDownClass(cls):
stop_test_server(cls.ps)
cls.tmp_dir.cleanup()
def setUp(self):
self.session = requests.Session()
self.url_base = f"{self.server}/api/{VERSION}/"
def test_config(self):
endpoint = "config"
url = f"{self.url_base}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/json")
result_data = result.json()
self.assertIsInstance(result_data["config"]["corpora_props"], dict)
self.assertIsInstance(result_data["config"]["parameters"], dict)
corpora_props = result_data["config"]["corpora_props"]
parameters = result_data["config"]["parameters"]
self.assertEqual(corpora_props["version"]["corpora_schema_version"], "1.0.0")
self.assertEqual(corpora_props["organism"], "human")
self.assertEqual(parameters["default_embedding"], "tsne")
@@ -0,0 +1,61 @@
from http import HTTPStatus
import unittest
import math
from local_server.test import start_test_server, stop_test_server
import local_server.test.unit.decode_fbs as decode_fbs
import requests
VERSION = "v0.2"
BAD_FILTER = {"filter": {"obs": {"annotation_value": [{"name": "xyz"}]}}}
class WithNaNs(unittest.TestCase):
"""Test Case for endpoints"""
@classmethod
def setUpClass(cls):
cls.ps, cls.server = start_test_server(["test/fixtures/nan.h5ad"])
@classmethod
def tearDownClass(cls):
stop_test_server(cls.ps)
def setUp(self):
self.session = requests.Session()
self.url_base = f"{self.server}/api/{VERSION}/"
def test_initialize(self):
endpoint = "schema"
url = f"{self.url_base}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
def test_data(self):
endpoint = "data/var"
url = f"{self.url_base}{endpoint}"
filter = {"filter": {"var": {"index": [[0, 20]]}}}
result = self.session.put(url, json=filter)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertTrue(math.isnan(df["columns"][3][3]))
def test_annotation_obs(self):
endpoint = "annotations/obs"
url = f"{self.url_base}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertTrue(math.isnan(df["columns"][2][0]))
def test_annotation_var(self):
endpoint = "annotations/var"
url = f"{self.url_base}{endpoint}"
result = self.session.get(url)
self.assertEqual(result.status_code, HTTPStatus.OK)
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
df = decode_fbs.decode_matrix_FBS(result.content)
self.assertTrue(math.isnan(df["columns"][2][0]))
@@ -0,0 +1,79 @@
import unittest
from urllib.parse import parse_qs
from werkzeug.datastructures import MultiDict
from local_server.common.rest import _query_parameter_to_filter
from local_server.common.errors import FilterError
def _qsparse(qs):
""" emulate what Flask/Werkzeug do to our QS """
return MultiDict(parse_qs(qs))
class FilterParseTests(unittest.TestCase):
""" Test cases for various filter parsing """
def test_queryparam_to_filter_parse(self):
# categories
self.assertEqual(
_query_parameter_to_filter(_qsparse("obs:foo=bar&var:baz=133&var:baz=A&obs:baz=foo")),
{
"obs": {"annotation_value": [{"name": "foo", "values": ["bar"]}, {"name": "baz", "values": ["foo"]}]},
"var": {"annotation_value": [{"name": "baz", "values": ["133", "A"]}]},
},
)
# ranges
self.assertEqual(
_query_parameter_to_filter(_qsparse("obs:A=1,99&obs:B=*,100&obs:C=0,*")),
{
"obs": {
"annotation_value": [
{"name": "A", "min": 1, "max": 99.0},
{"name": "B", "max": 100.0},
{"name": "C", "min": 0.0},
]
},
},
)
# combo
self.assertEqual(
_query_parameter_to_filter(_qsparse("var:B=YES&var:A=1,99&var:B=NO")),
{
"var": {
"annotation_value": [
{"name": "B", "values": ["YES", "NO"]},
{"name": "A", "min": 1.0, "max": 99.0},
]
},
},
)
def test_queryparam_to_filter_escaping(self):
self.assertEqual(
_query_parameter_to_filter(_qsparse("obs:var=%2521%252C%253AOK%253D&obs:A%2521=YO")),
{"obs": {"annotation_value": [{"name": "var", "values": ["!,:OK="]}, {"name": "A!", "values": ["YO"]}]}},
)
def test_queryparam_to_filter_errors(self):
# should raise FilterError
filter_errors = [
"foo=bar", # no axis
"X=&Y=3", # no value
"X&Y=3", # no value
"moo:foo=bar", # bad axis
"obs:x=1,A", # non-numeric range
"var:X=1,2&var:X=3,4", # duplicate ranges
"var:Y=,",
"var:Y=2,",
"var:Y=,5",
"var:Y=*,",
"var:Y=,*",
"var:Y=*,*",
]
for qs in filter_errors:
with self.assertRaises(FilterError):
_query_parameter_to_filter(_qsparse(qs))
@@ -0,0 +1,168 @@
import json
import shutil
import unittest
from os import path, listdir
import numpy as np
import pandas as pd
import local_server.test.unit.decode_fbs as decode_fbs
from local_server.common.rest import schema_get_helper, annotations_put_fbs_helper
from local_server.data_common.matrix_loader import MatrixDataType
from local_server.test import data_with_tmp_annotations, make_fbs
class WritableAnnotationTest(unittest.TestCase):
def setUp(self):
self.data, self.tmp_dir, self.annotations = data_with_tmp_annotations(MatrixDataType.H5AD)
self.data.dataset_config.user_annotations = self.annotations
def tearDown(self):
shutil.rmtree(self.tmp_dir)
def annotation_put_fbs(self, fbs):
annotations_put_fbs_helper(self.data, fbs)
res = json.dumps({"status": "OK"})
return res
def test_error_checks(self):
# verify that the expected errors are generated
n_rows = self.data.get_shape()[0]
fbs_bad = make_fbs({"louvain": pd.Series(["undefined"] * n_rows, dtype="category")})
# ensure we catch attempt to overwrite non-writable data
with self.assertRaises(KeyError):
self.annotation_put_fbs(fbs_bad)
def test_write_to_file(self):
# verify the file is written as expected
n_rows = self.data.get_shape()[0]
fbs = make_fbs(
{
"cat_A": pd.Series(["label_A"] * n_rows, dtype="category"),
"cat_B": pd.Series(["label_B"] * n_rows, dtype="category"),
}
)
res = self.annotation_put_fbs(fbs)
self.assertEqual(res, json.dumps({"status": "OK"}))
self.assertTrue(path.exists(self.annotations.output_file))
df = pd.read_csv(self.annotations.output_file, index_col=0, header=0, comment="#")
self.assertEqual(df.shape, (n_rows, 2))
self.assertEqual(set(df.columns), {"cat_A", "cat_B"})
self.assertTrue(self.data.original_obs_index.equals(df.index))
self.assertTrue(np.all(df["cat_A"] == ["label_A"] * n_rows))
self.assertTrue(np.all(df["cat_B"] == ["label_B"] * n_rows))
# verify complete overwrite on second attempt, AND rotation occurs
fbs = make_fbs(
{
"cat_A": pd.Series(["label_A1"] * n_rows, dtype="category"),
"cat_C": pd.Series(["label_C"] * n_rows, dtype="category"),
}
)
res = self.annotation_put_fbs(fbs)
self.assertEqual(res, json.dumps({"status": "OK"}))
self.assertTrue(path.exists(self.annotations.output_file))
df = pd.read_csv(self.annotations.output_file, index_col=0, header=0, comment="#")
self.assertEqual(set(df.columns), {"cat_A", "cat_C"})
self.assertTrue(np.all(df["cat_A"] == ["label_A1"] * n_rows))
self.assertTrue(np.all(df["cat_C"] == ["label_C"] * n_rows))
# rotation
name, ext = path.splitext(self.annotations.output_file)
backup_dir = f"{name}-backups"
self.assertTrue(path.isdir(backup_dir))
found_files = listdir(backup_dir)
self.assertEqual(len(found_files), 1)
def test_file_rotation_to_max_9(self):
# verify we stop rotation at 9
n_rows = self.data.get_shape()[0]
fbs = make_fbs(
{
"cat_A": pd.Series(["label_A"] * n_rows, dtype="category"),
"cat_B": pd.Series(["label_B"] * n_rows, dtype="category"),
}
)
for i in range(0, 11):
res = self.annotation_put_fbs(fbs)
self.assertEqual(res, json.dumps({"status": "OK"}))
name, ext = path.splitext(self.annotations.output_file)
backup_dir = f"{name}-backups"
self.assertTrue(path.isdir(backup_dir))
found_files = listdir(backup_dir)
self.assertTrue(len(found_files) <= 9)
def test_put_get_roundtrip(self):
# verify that OBS PUTs (annotation_put_fbs) are accessible via
# GET (annotation_to_fbs_matrix)
n_rows = self.data.get_shape()[0]
fbs = make_fbs(
{
"cat_A": pd.Series(["label_A"] * n_rows, dtype="category"),
"cat_B": pd.Series(["label_B"] * n_rows, dtype="category"),
}
)
# put
res = self.annotation_put_fbs(fbs)
self.assertEqual(res, json.dumps({"status": "OK"}))
# get
labels = self.annotations.read_labels(None)
fbsAll = self.data.annotation_to_fbs_matrix("obs", None, labels)
schema = schema_get_helper(self.data)
annotations = decode_fbs.decode_matrix_FBS(fbsAll)
obs_index_col_name = schema["annotations"]["obs"]["index"]
self.assertEqual(annotations["n_rows"], n_rows)
self.assertEqual(annotations["n_cols"], 7)
self.assertIsNone(annotations["row_idx"])
self.assertEqual(
annotations["col_idx"],
[obs_index_col_name, "n_genes", "percent_mito", "n_counts", "louvain", "cat_A", "cat_B"],
)
col_idx = annotations["col_idx"]
self.assertEqual(annotations["columns"][col_idx.index("cat_A")], ["label_A"] * n_rows)
self.assertEqual(annotations["columns"][col_idx.index("cat_B")], ["label_B"] * n_rows)
# verify the schema was updated
all_col_schema = {c["name"]: c for c in schema["annotations"]["obs"]["columns"]}
self.assertEqual(
all_col_schema["cat_A"],
{"name": "cat_A", "type": "categorical", "categories": ["label_A"], "writable": True},
)
self.assertEqual(
all_col_schema["cat_B"],
{"name": "cat_B", "type": "categorical", "categories": ["label_B"], "writable": True},
)
def test_put_float_data(self):
# verify that OBS PUTs (annotation_put_fbs) are accessible via
# GET (annotation_to_fbs_matrix)
n_rows = self.data.get_shape()[0]
# verifies that floating point with decimals fail.
fbs = make_fbs({"cat_F_FAIL": pd.Series([1.1] * n_rows, dtype=np.dtype("float"))})
with self.assertRaises(ValueError) as exception_context:
res = self.annotation_put_fbs(fbs)
self.assertEqual(str(exception_context.exception), "Columns may not have floating point types")
# verifies that floating point that can be converted to int passes
fbs = make_fbs({"cat_F_PASS": pd.Series([1.0] * n_rows, dtype="float")})
res = self.annotation_put_fbs(fbs)
self.assertEqual(res, json.dumps({"status": "OK"}))
# check read_labels
labels = self.annotations.read_labels(None)
fbsAll = self.data.annotation_to_fbs_matrix("obs", None, labels)
schema = schema_get_helper(self.data)
annotations = decode_fbs.decode_matrix_FBS(fbsAll)
self.assertEqual(annotations["n_rows"], n_rows)
all_col_schema = {c["name"]: c for c in schema["annotations"]["obs"]["columns"]}
self.assertEqual(
all_col_schema["cat_F_PASS"],
{"name": "cat_F_PASS", "type": "int32", "writable": True},
)
@@ -0,0 +1,217 @@
import unittest
from time import time
from unittest.mock import patch
import numpy as np
from pandas import Series, DataFrame
from local_server.common.utils.type_conversion_utils import (
can_cast_to_float32,
can_cast_to_int32,
get_dtype_of_array,
get_schema_type_hint_of_array,
get_dtypes_and_schemas_of_dataframe,
convert_pandas_series_to_numpy,
)
class TestTypeConversionUtils(unittest.TestCase):
def test__can_cast_to_float32__string_is_false(self):
array_to_convert = Series(data=["1", "2", "3"], dtype=str)
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_float32__float64_is_true_warning_outputted(self):
array_to_convert = Series(data=[1, 2, 3], dtype=np.dtype(np.float64))
with self.assertLogs(level="WARN") as logger:
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertIn("may lose precision", logger.output[0])
self.assertTrue(can_cast)
@patch("logging.warning")
def test__can_cast_to_float32__float32_is_false(self, mock_log_warning):
array_to_convert = Series(data=[1, 2, 3], dtype=np.dtype(np.float32))
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
assert not mock_log_warning.called
def test__can_cast_to_float32__categorical_float64_is_false(self):
array_to_convert = Series(data=[1.1, 2.2, 3.3], dtype="category")
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_float32__categorical_int64_with_nans_is_true(self):
array_to_convert = Series(data=[1, 2, np.NaN], dtype="category")
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_float_32__float_32_with_nans_is_true(self):
array_to_convert = Series(data=[1, 2, np.NaN], dtype=np.dtype(np.float32))
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_int32__string_is_false(self):
array_to_convert = Series(data=["1", "2", "3"], dtype=str)
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_int32__int64_is_true(self):
array_to_convert = Series(data=["1", "2", "3"], dtype=np.dtype(np.int64))
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_int32__int16_is_true(self):
array_to_convert = Series(data=["1", "2", "3"], dtype=np.dtype(np.int16))
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_int32__int64_with_large_value_is_false(self):
array_to_convert = Series(data=["3000000000", "2", "3"], dtype=np.dtype(np.int64))
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_int32__int64_with_nans_is_false(self):
array_to_convert = Series(data=[np.NaN, "2", "3"], dtype="category")
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__get_dtype_of_array__supported_dtypes_return_as_expected(self):
types = [np.float32, np.int32, np.bool_, str]
expected_dtypes = [np.float32, np.int32, np.uint8, str]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_dtype_of_array with type {types[test_type_index].__name__}", i=test_type_index
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_dtype_of_array(array), expected_dtypes[test_type_index])
def test__get_dtype_of_array__categories_return_as_expected(self):
array = Series(data=["a", "b", "c"], dtype="category")
expected_dtype = str
actual_dtype = get_dtype_of_array(array)
self.assertEqual(expected_dtype, actual_dtype)
def test__get_dtype_of_array__unordered_integer_categories_return_as_expected(self):
array = Series(data=[2, 3, 1, 3, 1, 2], dtype="category")
expected_dtype = np.int32
actual_dtype = get_dtype_of_array(array)
self.assertEqual(expected_dtype, actual_dtype)
def test__get_dtype_of_array__castable_dtypes_return_as_expected(self):
types = [np.float64, np.int64]
expected_dtypes = [np.float32, np.int32]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_dtype_of_array with castable type {types[test_type_index].__name__}", i=test_type_index
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_dtype_of_array(array), expected_dtypes[test_type_index])
def test__get_dtype_of_array__unsupported_type_raises_exception(self):
unsupported_array = Series(list([time() for _ in range(2)]), dtype="datetime64[ns]")
with self.assertRaises(TypeError) as exception_context:
get_dtype_of_array(unsupported_array)
self.assertIn("unsupported", str(exception_context.exception))
def test__get_schema_type_hint_of_array__supported_dtypes_return_as_expected(self):
types = [np.float32, np.int32, np.bool_, str]
expected_schema_hints = [{"type": "float32"}, {"type": "int32"}, {"type": "boolean"}, {"type": "string"}]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_schema_type_hint_of_array with type {types[test_type_index].__name__}", i=test_type_index
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_schema_type_hint_of_array(array), expected_schema_hints[test_type_index])
def test__get_schema_type_hint_of_array__categories_return_as_expected(self):
array = Series(data=["a", "b", "b"], dtype="category")
expected_schema_hint = {"type": "categorical", "categories": ["a", "b"]}
actual_schema_hint = get_schema_type_hint_of_array(array)
self.assertEqual(expected_schema_hint, actual_schema_hint)
def test__get_schema_type_hint_of_array__castable_dtypes_return_as_expected(self):
types = [np.float64, np.int64]
expected_schema_hints = [{"type": "float32"}, {"type": "int32"}]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_schema_type_hint_of_array with castable type {types[test_type_index].__name__}",
i=test_type_index,
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_schema_type_hint_of_array(array), expected_schema_hints[test_type_index])
def test__get_dtypes_and_schemas_of_dataframe__dtype_and_schema_returns_as_expected(self):
float_array = Series(data=[1, 2, 3], dtype=np.dtype(np.float64))
category_array = Series(data=["a", "b", "b"], dtype="category")
dataframe = DataFrame({"float_array": float_array, "category_array": category_array})
expected_data_types_dict = {"float_array": np.float32, "category_array": str}
expected_schema_type_hints_dict = {
"float_array": {"type": "float32"},
"category_array": {"type": "categorical", "categories": ["a", "b"]},
}
actual_dataframe_data_types, actual_dataframe_schema_type_hints = get_dtypes_and_schemas_of_dataframe(dataframe)
self.assertEqual(expected_data_types_dict, actual_dataframe_data_types)
self.assertEqual(expected_schema_type_hints_dict, actual_dataframe_schema_type_hints)
def test__convert_pandas_series_to_numpy__categorical_float64_to_float64_with_nans(self):
expected_float_array = np.array([1.1, 2.2, np.NaN], dtype=np.float64)
float_series = Series(data=[1.1, 2.2, np.NaN], dtype="category")
actual_float_array = convert_pandas_series_to_numpy(float_series, np.float64)
np.testing.assert_equal(expected_float_array, actual_float_array)
def test__convert_pandas_series_to_numpy__float64_to_float64(self):
expected_float_array = np.array([1.1, 2.2], dtype=np.float64)
float_series = Series(data=[1.1, 2.2], dtype=np.dtype(np.float64))
actual_float_array = convert_pandas_series_to_numpy(float_series, np.float64)
np.testing.assert_equal(expected_float_array, actual_float_array)
def test__convert_pandas_series_to_numpy__int64_to_int32_with_nans_throws_error(self):
int_series = Series(data=[1, 2, np.NaN], dtype="category")
with self.assertLogs(level="ERROR") as logger:
convert_pandas_series_to_numpy(int_series, np.int32)
self.assertIn(
"Cannot convert a pandas Series object to an integer dtype if it contains NaNs", logger.output[0]
)
@@ -0,0 +1,34 @@
import os
import shutil
import unittest
from local_server.common.utils.utils import import_plugins
from local_server.test import PROJECT_ROOT, random_string
class TestPlugins(unittest.TestCase):
""" Test plugin import functionality """
plugins_dir = f"{PROJECT_ROOT}/local_server/test/plugins"
test_plugin_path = f"{plugins_dir}/foo.py"
secret = random_string(8)
@classmethod
def setUpClass(cls) -> None:
if not os.path.isdir(cls.plugins_dir):
os.mkdir(cls.plugins_dir)
with open(cls.test_plugin_path, "w") as fh:
fh.write(f'SECRET = "{cls.secret}"\n')
@classmethod
def tearDownClass(cls) -> None:
if os.path.isdir(cls.plugins_dir):
shutil.rmtree(cls.plugins_dir)
def test_import_plugins(self):
self.assertTrue(os.path.isfile(self.test_plugin_path))
loaded_modules = import_plugins("local_server.test.plugins")
# test that import plugins found the file
self.assertEqual(["local_server.test.plugins.foo"], [ele.__name__ for ele in loaded_modules])
# test that the module was properly executed
self.assertEqual(self.secret, loaded_modules[0].SECRET)
@@ -0,0 +1,62 @@
import unittest
import numpy as np
from local_server.data_common.matrix_loader import MatrixDataLoader
from local_server.test import PROJECT_ROOT, app_config
class DiffExpTest(unittest.TestCase):
"""Tests the diffexp returns the expected results for one test case, using different
adaptor types and different algorithms."""
def load_dataset(self, path, extra_server_config={}, extra_dataset_config={}):
config = app_config(path, extra_server_config=extra_server_config, extra_dataset_config=extra_dataset_config)
loader = MatrixDataLoader(path)
adaptor = loader.open(config)
return adaptor
def get_mask(self, adaptor, start, stride):
"""Simple function to return a mask or rows"""
rows = adaptor.get_shape()[0]
sel = list(range(start, rows, stride))
mask = np.zeros(rows, dtype=bool)
mask[sel] = True
return mask
def compare_diffexp_results(self, results, expects):
self.assertEqual(len(results), len(expects))
for result, expect in zip(results, expects):
self.assertEqual(result[0], expect[0])
self.assertTrue(np.isclose(result[1], expect[1], 1e-6, 1e-4))
self.assertTrue(np.isclose(result[2], expect[2], 1e-6, 1e-4))
self.assertTrue(np.isclose(result[3], expect[3], 1e-6, 1e-4))
def check_1_10_2_10(self, results):
"""Checks the results for a specific set of rows selections"""
expects = [
[956, 0.016060986, 0.0008649321884808977, 1.0],
[1124, 0.96602094, 0.0011717216548271284, 1.0],
[1809, 1.1110606, 0.0019304405196777848, 1.0],
[1712, -0.5525154, 0.0051788902660723345, 1.0],
[1754, 0.5201581, 0.005691734062127954, 1.0],
[948, 1.6390722, 0.006622111055981219, 1.0],
[1810, 0.78618884, 0.007055917428377063, 1.0],
[779, 1.5241305, 0.007202934422407284, 1.0],
[1575, 1.0317602, 0.007830310753043345, 1.0],
[576, 0.97873515, 0.008272092578813124, 1.0],
]
self.compare_diffexp_results(results, expects)
def get_X_col(self, adaptor, cols):
varmask = np.zeros(adaptor.get_shape()[1], dtype=bool)
varmask[cols] = True
return adaptor.get_X_array(None, varmask)
def test_anndata_default(self):
"""Test an anndata adaptor with its default diffexp algorithm (diffexp_generic)"""
adaptor = self.load_dataset(f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad")
maskA = self.get_mask(adaptor, 1, 10)
maskB = self.get_mask(adaptor, 2, 10)
results = adaptor.compute_diffexp_ttest(maskA, maskB, 10)
self.check_1_10_2_10(results)
@@ -0,0 +1,61 @@
import os
import unittest
import pandas as pd
from local_server.test import FIXTURES_ROOT
from local_server.converters.schema import gene_symbol
class TestHGNCSymbolChecker(unittest.TestCase):
def setUp(self):
self.test_hgnc_path = os.path.join(FIXTURES_ROOT, "hgnc_example.txt.gz")
self.hgnc_checker = gene_symbol.HGNCSymbolChecker.from_hgnc_records(self.test_hgnc_path)
def test_symbol_upgrade(self):
self.assertEqual(self.hgnc_checker.upgrade_symbol("SEPT1"), "SEPTIN1")
self.assertEqual(self.hgnc_checker.upgrade_symbol("ADRB2R"), "ADRB2")
self.assertEqual(self.hgnc_checker.upgrade_symbol("BAR"), "ADRB2")
self.assertEqual(self.hgnc_checker.upgrade_symbol("sept1"), "SEPTIN1")
self.assertEqual(self.hgnc_checker.upgrade_symbol("AdRb2R"), "ADRB2")
self.assertEqual(self.hgnc_checker.upgrade_symbol("bar"), "ADRB2")
# Strip off seurat endings when appropriate
self.assertEqual(self.hgnc_checker.upgrade_symbol("SEPT1.1"), "SEPTIN1")
self.assertEqual(self.hgnc_checker.upgrade_symbol("ADRB2-1"), "ADRB2")
# DIFF6 is ambiguous so don't upgrade it
self.assertEqual(self.hgnc_checker.upgrade_symbol("DIFF6"), "DIFF6")
self.assertEqual(self.hgnc_checker.upgrade_symbol("diff6"), "diff6")
# ARG1 is approved
self.assertEqual(self.hgnc_checker.upgrade_symbol("ARG1"), "ARG1")
self.assertEqual(self.hgnc_checker.upgrade_symbol("arg1"), "ARG1")
# HAP1 is both approved and withdrawn
self.assertEqual(self.hgnc_checker.upgrade_symbol("HAP1"), "HAP1")
self.assertEqual(self.hgnc_checker.upgrade_symbol("hap1"), "HAP1")
# Leave unknown symbols alone
self.assertEqual(self.hgnc_checker.upgrade_symbol("NOTASYMBOL"), "NOTASYMBOL")
self.assertEqual(self.hgnc_checker.upgrade_symbol("notasymbol"), "notasymbol")
# Upgrade HGNC ids unless you can't find it
self.assertEqual(self.hgnc_checker.upgrade_symbol("HGNC:286"), "ADRB2")
self.assertEqual(self.hgnc_checker.upgrade_symbol("HGNC:4812"), "HAP1")
self.assertEqual(self.hgnc_checker.upgrade_symbol("HGNC:123456"), "HGNC:123456")
def test_check_symbol(self):
self.assertEqual(self.hgnc_checker.check_symbol("SEPT1"), gene_symbol.SymbolStatus.UPGRADABLE)
self.assertEqual(self.hgnc_checker.check_symbol("DIFF6"), gene_symbol.SymbolStatus.AMBIGUOUS)
self.assertEqual(self.hgnc_checker.check_symbol("NOTASYMBOL"), gene_symbol.SymbolStatus.UNKNOWN)
# HAP1 is one of the approved and withdrawn symbols
self.assertEqual(self.hgnc_checker.check_symbol("HAP1"), gene_symbol.SymbolStatus.APPROVED)
def test_upgrade_index(self):
index = pd.Index(["SEPT1", "DIFF6", "NOTASYMBOL", "bar", "SEPTIN1"])
var_df = pd.DataFrame([[0] * len(index)], index=index)
upgraded_index = gene_symbol.get_upgraded_var_index(var_df, hgnc_path=self.test_hgnc_path)
self.assertEqual(upgraded_index.tolist(), ["SEPTIN1", "DIFF6", "NOTASYMBOL", "ADRB2", "SEPTIN1"])
@@ -0,0 +1,129 @@
import json
import unittest
import unittest.mock
from local_server.converters.schema import ontology
class TestOntologyParsing(unittest.TestCase):
def setUp(self):
self.curies = ["UBERON:0002048", "HsapDv:0000174", "NCBITaxon:9606", "EFO:0008995"]
self.names = ["UBERON", "HsapDv", "NCBITaxon", "EFO"]
self.values = ["0002048", "0000174", "9606", "0008995"]
self.iris = [
"http://purl.obolibrary.org/obo/UBERON_0002048",
"http://purl.obolibrary.org/obo/HsapDv_0000174",
"http://purl.obolibrary.org/obo/NCBITaxon_9606",
"http://www.ebi.ac.uk/efo/EFO_0008995",
]
URL_ROOT = "http://www.ebi.ac.uk/ols/api/ontologies/"
self.urls = [
URL_ROOT + "UBERON/terms/http%253A%252F%252Fpurl.obolibrary.org%252Fobo%252FUBERON_0002048",
URL_ROOT + "HsapDv/terms/http%253A%252F%252Fpurl.obolibrary.org%252Fobo%252FHsapDv_0000174",
URL_ROOT + "NCBITaxon/terms/http%253A%252F%252Fpurl.obolibrary.org%252Fobo%252FNCBITaxon_9606",
URL_ROOT + "EFO/terms/http%253A%252F%252Fwww.ebi.ac.uk%252Fefo%252FEFO_0008995",
]
self.responses = {
"UBERON:0002048": {
"iri": "http://purl.obolibrary.org/obo/UBERON_0002048",
"description": ["Respiration organ that develops as an outpocketing of the esophagus."],
"label": "lung",
},
"HsapDv:0000174": {
"iri": "http://purl.obolibrary.org/obo/HsapDv_0000174",
"description": ["Infant stage that refers to an infant who is over 1 and under 2 months old."],
"label": "1-month-old human stage",
},
"NCBITaxon:9606": {
"iri": "http://purl.obolibrary.org/obo/NCBITaxon_9606",
"description": None,
"label": "Homo sapiens",
},
"EFO:0008995": {
"iri": "http://www.ebi.ac.uk/efo/EFO_0008995",
"description": [
(
'10X is a "synthetic long-read" technology and works by capturing a barcoded oligo-coated '
"gel-bead and 0.3x genome copies into a single emulsion droplet, processing the equivalent "
"of 1 million pipetting steps. Successive versions of the 10x chemistry use different "
"barcode locations to improve the sequencing yield and quality of 10x experiments."
)
],
"label": "10X sequencing",
},
}
def test_ontololgy_name(self):
for curie, expected_name in zip(self.curies, self.names):
self.assertEqual(ontology._ontology_name(curie), expected_name)
def test_ontololgy_value(self):
for curie, expected_value in zip(self.curies, self.values):
self.assertEqual(ontology._ontology_value(curie), expected_value)
def test_iri(self):
for curie, expected_iri in zip(self.curies, self.iris):
self.assertEqual(ontology._iri(curie), expected_iri)
def test_ontology_info_url(self):
for curie, expected_url in zip(self.curies, self.urls):
self.assertEqual(ontology._ontology_info_url(curie), expected_url)
def test_empty_ontology_info_url(self):
self.assertEqual(ontology._ontology_info_url(""), "")
class TestOntologyLookup(unittest.TestCase):
def setUp(self):
self.responses = {
"UBERON:0002048": {
"iri": "http://purl.obolibrary.org/obo/UBERON_0002048",
"description": ["Respiration organ that develops as an outpocketing of the esophagus."],
"label": "lung",
},
"HsapDv:0000174": {
"iri": "http://purl.obolibrary.org/obo/HsapDv_0000174",
"description": ["Infant stage that refers to an infant who is over 1 and under 2 months old."],
"label": "1-month-old human stage",
},
"NCBITaxon:9606": {
"iri": "http://purl.obolibrary.org/obo/NCBITaxon_9606",
"description": None,
"label": "Homo sapiens",
},
"EFO:0008995": {
"iri": "http://www.ebi.ac.uk/efo/EFO_0008995",
"description": [
('10X is a "synthetic long-read" technology and works by capturing a barcoded oligo-coated '
'gel-bead and 0.3x genome copies into a single emulsion droplet, processing the equivalent '
'of 1 million pipetting steps. Successive versions of the 10x chemistry use different barcode '
'locations to improve the sequencing yield and quality of 10x experiments.')
],
"label": "10X sequencing",
},
}
self.labels = {
"UBERON:0002048": "lung",
"HsapDv:0000174": "1-month-old human stage",
"NCBITaxon:9606": "Homo sapiens",
"EFO:0008995": "10X sequencing",
}
@unittest.mock.patch("requests.get")
def test_lookup_label(self, mock_get):
for curie, response in self.responses.items():
mock_get.return_value.content = json.dumps(response)
mock_get.return_value.json.return_value = response
mock_get.return_value.status_code = 200
label = ontology.get_ontology_label(curie)
self.assertEqual(label, self.labels[curie])
@@ -0,0 +1,257 @@
import json
import os
import unittest
import unittest.mock
import anndata
import numpy
import pandas as pd
import scanpy as sc
from local_server.converters.schema import remix
PROJECT_ROOT = os.popen("git rev-parse --show-toplevel").read().strip()
class TestApplySchema(unittest.TestCase):
def setUp(self):
self.source_h5ad_path = f"{PROJECT_ROOT}/local_server/test/fixtures/pbmc3k-CSC-gz.h5ad"
self.output_h5ad_path = f"{PROJECT_ROOT}/local_server/test/fixtures/test_remix.h5ad"
self.config_path = f"{PROJECT_ROOT}/local_server/test/fixtures/test_config.yaml"
self.bad_config_path = f"{PROJECT_ROOT}/local_server/test/fixtures/test_bad_config.yaml"
def tearDown(self):
try:
os.remove(self.output_h5ad_path)
except OSError:
pass
@unittest.mock.patch("local_server.converters.schema.ontology.get_ontology_label")
def test_apply_schema(self, mock_get_ontology_label):
mock_get_ontology_label.return_value = "test label"
remix.apply_schema(self.source_h5ad_path, self.config_path, self.output_h5ad_path)
new_adata = sc.read_h5ad(self.output_h5ad_path)
self.assertIn("cell_type", new_adata.obs.columns)
self.assertListEqual(["test label"], new_adata.obs["cell_type"].unique().tolist())
self.assertListEqual(
["CL:00001", "CL:00002", "CL:00003", "CL:00004", "CL:00005", "CL:00006", "CL:00007", "CL:00008"],
sorted(new_adata.obs["cell_type_ontology_term_id"].unique().tolist())
)
self.assertIn("version", new_adata.uns_keys())
@unittest.mock.patch("local_server.converters.schema.ontology.get_ontology_label")
def test_apply_bad_schema(self, mock_get_ontology_label):
mock_get_ontology_label.return_value = "test label"
remix.apply_schema(self.source_h5ad_path, self.bad_config_path, self.output_h5ad_path)
new_adata = sc.read_h5ad(self.output_h5ad_path)
# Should refuse to write the version
self.assertNotIn("version", new_adata.uns_keys())
class TestFieldParsing(unittest.TestCase):
def test_is_curie(self):
self.assertTrue(remix.is_curie("EFO:00001"))
self.assertTrue(remix.is_curie("UBERON:123456"))
self.assertTrue(remix.is_curie("HsapDv:0001"))
self.assertFalse(remix.is_curie("UBERON"))
self.assertFalse(remix.is_curie("UBERON:"))
self.assertFalse(remix.is_curie("123456"))
def test_is_ontology_field(self):
self.assertTrue(remix.is_ontology_field("tissue_ontology_term_id"))
self.assertTrue(remix.is_ontology_field("cell_type_ontology_term_id"))
self.assertFalse(remix.is_ontology_field("cell_ontology"))
self.assertFalse(remix.is_ontology_field("method"))
def test_get_label_field_name(self):
self.assertEqual("tissue", remix.get_label_field_name("tissue_ontology_term_id"))
self.assertEqual("cell_type", remix.get_label_field_name("cell_type_ontology_term_id"))
def test_split_suffix(self):
self.assertEqual(("UBERON:1234", " (organoid)"), remix.split_suffix("UBERON:1234 (organoid)"))
self.assertEqual(("UBERON:1234", " (cell culture)"), remix.split_suffix("UBERON:1234 (cell culture)"))
self.assertEqual(("UBERON:1234", ""), remix.split_suffix("UBERON:1234"))
self.assertEqual(("UBERON:1234 (something)", ""), remix.split_suffix("UBERON:1234 (something)"))
@unittest.mock.patch("local_server.converters.schema.ontology.get_ontology_label")
def test_get_curie_and_label(self, mock_get_ontology_label):
mock_get_ontology_label.return_value = "test label"
self.assertEqual(
remix.get_curie_and_label("UBERON:1234"),
("UBERON:1234", "test label")
)
self.assertEqual(
remix.get_curie_and_label("UBERON:1234 (cell culture)"),
("UBERON:1234 (cell culture)", "test label (cell culture)")
)
self.assertEqual(
remix.get_curie_and_label("whatever"),
("", "whatever")
)
class TestManipulateAnndata(unittest.TestCase):
def setUp(self):
self.cell_count = 20
self.gene_count = 200
X = numpy.random.randint(0, 1000, (self.cell_count, self.gene_count))
uns = {"organism": "monkey", "experiment": "monkey experiment"}
obs = pd.DataFrame(
index=[f"Cell{d}" for d in range(self.cell_count)],
columns=["tissue", "CellType"],
data=[["lung", "epithelial"]] * (self.cell_count // 2) + [["lung", "endothelial"]] * (self.cell_count // 2)
)
var = pd.DataFrame(index=[f"SEPT{d}" for d in range(self.gene_count)])
self.adata = anndata.AnnData(X=X, obs=obs, var=var, uns=uns)
def test_safe_add_field(self):
remix.safe_add_field(self.adata.obs, "tissue", ["monkey lung"] * self.cell_count)
self.assertEqual(self.adata.obs["tissue_original"].tolist(), ["lung"] * self.cell_count)
self.assertEqual(self.adata.obs["tissue"].tolist(), ["monkey lung"] * self.cell_count)
remix.safe_add_field(self.adata.uns, "contributors", [{"name": "contributor1"}, {"name": "contributor2"}])
self.assertEqual(
self.adata.uns["contributors"],
json.dumps([{"name": "contributor1"}, {"name": "contributor2"}])
)
@unittest.mock.patch("local_server.converters.schema.ontology.get_ontology_label")
def test_remix_uns(self, mock_get_ontology_label):
mock_get_ontology_label.return_value = "Pan troglodytes"
uns_config = {
"version": {
"corpora_schema_version": "1.0.0",
"corpora_encoding_version": "0.1.0"
},
"organism_ontology_term_id": "NCBITaxon:9598",
"contributors": [
{
"name": "scientist",
"email": "scientist@science.com"
}
]
}
remix.remix_uns(self.adata, uns_config)
self.assertEqual(
sorted(self.adata.uns_keys()),
sorted(["organism_original", "organism", "organism_ontology_term_id",
"contributors", "version", "experiment"])
)
self.assertEqual(self.adata.uns['organism'], "Pan troglodytes")
self.assertEqual(self.adata.uns['organism_original'], "monkey")
self.assertEqual(self.adata.uns['organism_ontology_term_id'], "NCBITaxon:9598")
self.assertEqual(self.adata.uns['contributors'],
json.dumps([{"name": "scientist", "email": "scientist@science.com"}]))
@unittest.mock.patch("local_server.converters.schema.ontology.get_ontology_label")
def test_remix_obs(self, mock_get_ontology_label):
mock_get_ontology_label.return_value = "lung (in a monkey)"
obs_config = {
"tissue_ontology_term_id": {
"tissue": {
"lung": "UBERON:00000"
}
},
"cell_color": {
"CellType": {
"epithelial": "fuschia",
"endothelial": "khaki"
}
},
"sex": "male"
}
remix.remix_obs(self.adata, obs_config)
self.assertEqual(
sorted(self.adata.obs_keys()),
sorted(["tissue", "tissue_ontology_term_id", "tissue_original", "CellType", "cell_color", "sex"])
)
self.assertTrue(all(v == "lung" for v in self.adata.obs.tissue_original))
self.assertTrue(all(v == "UBERON:00000" for v in self.adata.obs.tissue_ontology_term_id))
self.assertTrue(all(v == "lung (in a monkey)" for v in self.adata.obs.tissue))
self.assertTrue(all(v == "male" for v in self.adata.obs.sex))
self.assertTrue(all(v in (("epithelial", "fuschia"), ("endothelial", "khaki"))
for v in zip(self.adata.obs.CellType, self.adata.obs.cell_color)))
class TestFixupGeneSymbols(unittest.TestCase):
def setUp(self):
self.seurat_path = f"{PROJECT_ROOT}/local_server/test/fixtures/schema_test_data/seurat_tutorial.h5ad"
self.seurat_merged_path = f"{PROJECT_ROOT}/local_server/test/fixtures/schema_test_data/seurat_tutorial_merged.h5ad"
self.sctransform_path = f"{PROJECT_ROOT}/local_server/test/fixtures/schema_test_data/sctransform.h5ad"
self.sctransform_merged_path = f"{PROJECT_ROOT}/local_server/test/fixtures/schema_test_data/sctransform_merged.h5ad"
# There's lots of MALAT1, but it doesn't collide with any other names,
# so it shouldn't change during merging.
self.stable_gene = "MALAT1"
def test_fixup_gene_symbols_seurat(self):
if not os.path.isfile(self.seurat_path):
return unittest.skip(
"Skipping gene symbol conversion tests because test h5ads are not present. To create them, "
"run local_server/test/fixtures/schema_test_data/generate_test_data.sh"
)
original_adata = sc.read_h5ad(self.seurat_path)
merged_adata = sc.read_h5ad(self.seurat_merged_path)
fixup_config = {"X": "log1p", "counts": "raw", "scale.data": "log1p"}
fixed_adata = remix.fixup_gene_symbols(original_adata, fixup_config)
self.assertEqual(
merged_adata.layers["counts"][:, merged_adata.var.index == self.stable_gene].sum(),
fixed_adata.raw.X[:, fixed_adata.var.index == self.stable_gene].sum()
)
self.assertAlmostEqual(
merged_adata.X[:, merged_adata.var.index == self.stable_gene].sum(),
fixed_adata.X[:, fixed_adata.var.index == self.stable_gene].sum()
)
self.assertAlmostEqual(
merged_adata.layers["scale.data"][:, merged_adata.var.index == self.stable_gene].sum(),
fixed_adata.layers["scale.data"][:, fixed_adata.var.index == self.stable_gene].sum()
)
def test_fixup_gene_symbols_sctransform(self):
if not os.path.isfile(self.sctransform_path):
return unittest.skip(
"Skipping gene symbol conversion tests because test h5ads are not present. To create them, "
"run local_server/test/fixtures/schema_test_data/generate_test_data.sh"
)
original_adata = sc.read_h5ad(self.sctransform_path)
merged_adata = sc.read_h5ad(self.sctransform_merged_path)
fixup_config = {"X": "log1p", "counts": "raw"}
fixed_adata = remix.fixup_gene_symbols(original_adata, fixup_config)
# sctransform does a bunch of stuff, including slightly modifying the
# raw counts. So we can't assert for exact equality the way we do with
# the vanilla seurat tutorial. But, the results should still be very
# close.
merged_raw_stable = merged_adata.layers["counts"][:, merged_adata.var.index == self.stable_gene].sum()
fixed_raw_stable = fixed_adata.raw.X[:, fixed_adata.var.index == self.stable_gene].sum()
self.assertLess(abs(merged_raw_stable - fixed_raw_stable), .001 * merged_raw_stable)
self.assertAlmostEqual(
merged_adata.X[:, merged_adata.var.index == self.stable_gene].sum(),
fixed_adata.X[:, fixed_adata.var.index == self.stable_gene].sum(),
0
)
@@ -0,0 +1,435 @@
import json
import os
import unittest
import pandas as pd
import scanpy as sc
from local_server.converters.schema import validate
PROJECT_ROOT = os.popen("git rev-parse --show-toplevel").read().strip()
class TestFieldValidation(unittest.TestCase):
def test_validate_stringified_list_of_dicts(self):
good = json.dumps([{"a": 1}, {2: "x", "z": "y"}])
not_stringified = [{"a": 1}, {2: "x", "z": "y"}]
not_a_list = json.dumps({"bad": "dict"})
not_json = "oh hey!"
self.assertTrue(validate._validate_stringified_list_of_dicts(good))
self.assertFalse(validate._validate_stringified_list_of_dicts(not_stringified))
self.assertFalse(validate._validate_stringified_list_of_dicts(not_a_list))
self.assertFalse(validate._validate_stringified_list_of_dicts(not_json))
def test_validate_human_readable_string(self):
good = "oh hey!"
curie = "EFO:0001"
ensg = "ENSG000001234"
enst = "ENST000005678"
self.assertTrue(validate._validate_human_readable_string(good))
self.assertFalse(validate._validate_human_readable_string(curie))
self.assertFalse(validate._validate_human_readable_string(ensg))
self.assertFalse(validate._validate_human_readable_string(enst))
def test_validate_curie(self):
self.assertTrue(validate._validate_curie("UBERON:00001", ["UBERON", "EFO"]))
self.assertTrue(validate._validate_curie("HsapDv:00002", ["HsapDv"]))
self.assertFalse(validate._validate_curie("HsapDv:00002", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_curie("EFO:00002 (organoid)", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_curie("EFO:00002 extra", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_curie("UBERON:ABCD", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_curie("Uberon:00002", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_curie("UBERON:", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_curie("UBERON", ["UBERON", "EFO"]))
def test_validate_suffixed_curie(self):
self.assertTrue(validate._validate_suffixed_curie("EFO:00001", ["UBERON", "EFO"]))
self.assertTrue(validate._validate_suffixed_curie("UBERON:00001 (cell culture)", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("HsapDv:00002 (organoid)", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("HsapDv:00002(organoid)", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("HsapDv:00002", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("EFO:00002 extra", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("UBERON:ABCD", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("Uberon:00002", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("UBERON:", ["UBERON", "EFO"]))
self.assertFalse(validate._validate_suffixed_curie("UBERON", ["UBERON", "EFO"]))
class TestColumnValidation(unittest.TestCase):
def test_validate_unique(self):
unique = pd.DataFrame([["abc", "def"], ["ghi", "jkl"], ["mnop", "qrs"]],
index=["X", "Y", "Z"], columns=["col1", "col2"])
duped = pd.DataFrame([["abc", "def"], ["ghi", "qrs"], ["abc", "qrs"]],
index=["X", "Y", "X"], columns=["col1", "col2"])
schema_def = {"unique": True}
errors = validate._validate_column(unique.index, "index", "unique_df", schema_def)
self.assertFalse(errors)
errors = validate._validate_column(duped.index, "index", "duped_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("is not unique", errors[0])
errors = validate._validate_column(unique["col1"], "col1", "unique_df", schema_def)
self.assertFalse(errors)
errors = validate._validate_column(duped["col1"], "col1", "duped_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("is not unique", errors[0])
schema_def = {"unique": False}
errors = validate._validate_column(duped["col1"], "col1", "duped_df", schema_def)
self.assertFalse(errors)
def test_validate_nullable(self):
non_null = pd.DataFrame([["abc", "def"], ["ghi", "jkl"], ["mnop", "qrs"]],
index=["X", "Y", "Z"], columns=["col1", "col2"])
has_null = pd.DataFrame([["abc", "", None], ["ghi", "jkl", 1], ["mnop", "qrs", 2]],
index=["X", "Y", "Z"], columns=["col1", "col2", "col3"])
schema_def = {"nullable": False}
errors = validate._validate_column(non_null["col1"], "col1", "nonnull_df", schema_def)
self.assertFalse(errors)
errors = validate._validate_column(has_null["col1"], "col1", "hasnull_df", schema_def)
self.assertFalse(errors)
errors = validate._validate_column(has_null["col2"], "col2", "hasnull_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("contains empty values", errors[0])
errors = validate._validate_column(has_null["col3"], "col3", "hasnull_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("contains empty values", errors[0])
schema_def = {"nullable": True}
errors = validate._validate_column(has_null["col2"], "col2", "hasnull_df", schema_def)
self.assertFalse(errors)
def test_human_readable(self):
hr_df = pd.DataFrame(
[["for you, a human", "UBERON:12345", "UBERON:1234 (thundercat)"],
["hope you're well", "bit of lungs", "brain"]],
index=["ENSG00001", "ENSG00002"],
columns=["good", "curie", "suffixed_curie"])
schema_def = {"type": "human-readable string"}
errors = validate._validate_column(hr_df["good"], "good", "hr", schema_def)
self.assertFalse(errors)
errors = validate._validate_column(hr_df["curie"], "curie", "hr", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("non-human-readable", errors[0])
errors = validate._validate_column(hr_df["suffixed_curie"], "suffixed_curie", "hr", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("non-human-readable", errors[0])
errors = validate._validate_column(hr_df.index, "ensg", "hr", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("non-human-readable", errors[0])
def test_curie(self):
curie_df = pd.DataFrame(
[["EFO:00001", "HsapDv:00001 (cell culture)", "EFO:", "MONDO:0001 cell culture"],
["UBERON:00002", "HsapDv:00002 (organoid)", "EFO:12345", "MONDO:0002 (baba yaga)"],
["EFO:0000000005", "HsapDv:000004 (humanzee)", "EFO:000002", "MONDO:0004 (TMNT)"]],
index=["X", "Y", "Z"],
columns=["good", "good_suffix", "bad", "bad_suffix"])
# Good
schema_def = {"type": "curie", "prefixes": ["EFO", "UBERON"]}
errors = validate._validate_column(curie_df["good"], "good", "curie_df", schema_def)
self.assertFalse(errors)
# Good suffix
schema_def = {"type": "suffixed curie", "prefixes": ["HsapDv", "WHATEVER"]}
errors = validate._validate_column(curie_df["good_suffix"], "good_suffix", "curie_df", schema_def)
self.assertFalse(errors)
# Bad prefix
schema_def = {"type": "curie", "prefixes": ["EFO"]}
errors = validate._validate_column(curie_df["good"], "good", "curie_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("invalid ontology", errors[0])
self.assertIn("must be curies from one of these", errors[0])
# Bad curies
schema_def = {"type": "curie", "prefixes": ["EFO"]}
errors = validate._validate_column(curie_df["bad"], "bad", "curie_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("invalid ontology", errors[0])
# Bad suffixes
schema_def = {"type": "suffixed curie", "prefixes": ["EFO"]}
errors = validate._validate_column(curie_df["bad_suffix"], "bad_suffix", "curie_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("invalid ontology", errors[0])
def test_enum(self):
enum_df = pd.DataFrame(
[["abc", "ghi"],
["def", "jkl"]],
index=["X", "Y"],
columns=["col1", "col2"])
# All match
schema_def = {"type": "string", "enum": ["abc", "def", "xyz"]}
errors = validate._validate_column(enum_df["col1"], "col1", "enum_df", schema_def)
self.assertFalse(errors)
# Missing value
schema_def = {"type": "string", "enum": ["abc", "xyz"]}
errors = validate._validate_column(enum_df["col1"], "col1", "enum_df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("unpermitted values", errors[0])
class TestDictValidations(unittest.TestCase):
def test_key_presence(self):
schema_def = {"keys": {"abc": None, "def": None}}
dict_ = {"abc": "123", "def": "456"}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertFalse(errors)
# Missing keys are bad
dict_ = {"abc": "123"}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("missing key", errors[0])
# Extra keys are okay
dict_ = {"abc": "123", "def": "456", "xyz": "789"}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertFalse(errors)
# Better not be empty come on
dict_ = {}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertEqual(len(errors), 2)
def test_nullable(self):
schema_def = {"keys": {"abc": {"type": "string", "nullable": False},
"def": {"type": "string", "nullable": True}}}
dict_ = {"abc": "xyz", "def": ""}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertFalse(errors)
dict_ = {"abc": "", "def": ""}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("empty value", errors[0])
def test_recurse(self):
schema_def = {
"keys": {
"subdict": {
"type": "dict",
"keys": {
"subdict_key1": None,
"subdict_key2": None
}
},
"ontology": {
"type": "curie",
"prefixes": ["ONTOLOGY"]
},
"blob": {
"type": "stringified list of dicts"
}
}
}
dict_ = {
"subdict": {"subdict_key1": "any", "subdict_key2": "any"},
"ontology": "ONTOLOGY:123456",
"blob": json.dumps([{"abc": 123}, {"def": 456}])
}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertFalse(errors)
dict_ = {
"subdict": {"subdict_key1": "any"},
"ontology": "ONTOLOGY:123456",
"blob": json.dumps([{"abc": 123}, {"def": 456}])
}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("missing key", errors[0])
dict_ = {
"subdict": {"subdict_key1": "any", "subdict_key2": "any"},
"ontology": "oh no not an ontology term",
"blob": json.dumps([{"abc": 123}, {"def": 456}])
}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("invalid ontology", errors[0])
dict_ = {
"subdict": {"subdict_key1": "any", "subdict_key2": "any"},
"ontology": "ONTOLOGY:123456",
"blob": [{"abc": 123}, {"def": 456}]
}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("JSON-encoded list of dicts", errors[0])
# Multiple errors
dict_ = {
"subdict": {"subdict_key1": "any"},
"ontology": "oh no not an ontology term",
"blob": json.dumps([{"abc": 123}, {"def": 456}])
}
errors = validate._validate_dict(dict_, "d", schema_def)
self.assertEqual(len(errors), 2)
class TestDataframeValidation(unittest.TestCase):
def test_column_presence(self):
df = pd.DataFrame(
[["abc", "EFO:123"],
["def", "UBERON:456"]],
columns=["hr_string", "ontology"],
index=["X", "Y"]
)
schema_def = {
"columns": {
"hr_string": {"type": "human-readable string"},
"ontology": {"type": "curie", "prefixes": ["EFO", "UBERON"]}
}
}
errors = validate._validate_dataframe(df, "df", schema_def)
self.assertFalse(errors)
schema_def = {
"columns": {
"hr_string": {"type": "human-readable string"},
"another_hr_string": {"type": "human-readable string"},
"ontology": {"type": "curie", "prefixes": ["EFO", "UBERON"]}
}
}
errors = validate._validate_dataframe(df, "df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("missing column", errors[0])
# Extra is okay
df = pd.DataFrame(
[["abc", "EFO:123", "extra"],
["def", "UBERON:456", "extra"]],
columns=["hr_string", "ontology", "extra"],
index=["X", "Y"]
)
schema_def = {
"columns": {
"hr_string": {"type": "human-readable string"},
"ontology": {"type": "curie", "prefixes": ["EFO", "UBERON"]}
}
}
errors = validate._validate_dataframe(df, "df", schema_def)
self.assertFalse(errors)
def test_index(self):
df = pd.DataFrame(
[["abc", "123"],
["def", "456"]],
columns=["col1", "col2"],
index=["ENSG0001", "ENSG0002"]
)
schema_def = {"index": {"unique": True}}
errors = validate._validate_dataframe(df, "df", schema_def)
self.assertFalse(errors)
schema_def = {"index": {"type": "human-readable string"}}
errors = validate._validate_dataframe(df, "df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("non-human-readable", errors[0])
df = pd.DataFrame(
[["abc", "123"],
["def", "456"]],
columns=["col1", "col2"],
index=["ENSG0001", "ENSG0001"]
)
schema_def = {"index": {"unique": True}}
errors = validate._validate_dataframe(df, "df", schema_def)
self.assertEqual(len(errors), 1)
self.assertIn("is not unique", errors[0])
def test_recurse(self):
df = pd.DataFrame(
[["abc", "HsapDv:0001"],
["EFO:123", "UBERON:456"]],
columns=["hr_string", "ontology"],
index=["X", "Y"]
)
schema_def = {
"columns": {
"hr_string": {"type": "human-readable string"},
"ontology": {"type": "curie", "prefixes": ["EFO", "UBERON"]}
}
}
errors = validate._validate_dataframe(df, "df", schema_def)
self.assertEqual(len(errors), 2)
self.assertEqual(len([e for e in errors if "non-human-readable" in e]), 1)
self.assertEqual(len([e for e in errors if "invalid ontology" in e]), 1)
class TestGetSchema(unittest.TestCase):
def test_get_schema(self):
self.assertIsInstance(validate.get_schema_definition("1.0.0"), dict)
with self.assertRaises(ValueError):
validate.get_schema_definition("10.1.5")
class TestValidate(unittest.TestCase):
def setUp(self):
self.source_h5ad_path = f"{PROJECT_ROOT}/local_server/test/fixtures/pbmc3k-CSC-gz.h5ad"
def test_shallow(self):
adata = sc.read_h5ad(self.source_h5ad_path)
self.assertFalse(validate.validate_adata(adata, True))
adata.uns["version"] = {
"corpora_schema_version": "1.0.0",
"corpora_encoding_version": "0.1.0"
}
self.assertTrue(validate.validate_adata(adata, True))
def test_deep(self):
adata = sc.read_h5ad(self.source_h5ad_path)
self.assertFalse(validate.validate_adata(adata, False))
adata.uns["version"] = {
"corpora_schema_version": "1.0.0",
"corpora_encoding_version": "0.1.0"
}
self.assertFalse(validate.validate_adata(adata, False))
@@ -0,0 +1,232 @@
import json
import sys
import time
import unittest
import numpy as np
import pandas as pd
import pytest
from parameterized import parameterized_class
import local_server.test.unit.decode_fbs as decode_fbs
from local_server.common.data_locator import DataLocator
from local_server.common.errors import FilterError
from local_server.data_anndata.anndata_adaptor import AnndataAdaptor
from local_server.test import PROJECT_ROOT, app_config, FIXTURES_ROOT
from local_server.test.fixtures.fixtures import pbmc3k_colors
"""
Test the anndata adaptor using the pbmc3k data set.
"""
@parameterized_class(
("data_locator", "backed"),
[
(f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad", False),
(f"{FIXTURES_ROOT}/pbmc3k-CSC-gz.h5ad", False),
(f"{FIXTURES_ROOT}/pbmc3k-CSR-gz.h5ad", False),
(f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad", True),
(f"{FIXTURES_ROOT}/pbmc3k-CSC-gz.h5ad", True),
(f"{FIXTURES_ROOT}/pbmc3k-CSR-gz.h5ad", True),
],
)
class AdaptorTest(unittest.TestCase):
def setUp(self):
config = app_config(self.data_locator, self.backed)
self.data = AnndataAdaptor(DataLocator(self.data_locator), config)
def test_init(self):
self.assertEqual(self.data.cell_count, 2638)
self.assertEqual(self.data.gene_count, 1838)
epsilon = 0.000_005
self.assertTrue(self.data.data.X[0, 0] - -0.171_469_51 < epsilon)
def test_mandatory_annotations(self):
obs_index_col_name = self.data.get_schema()["annotations"]["obs"]["index"]
self.assertIn(obs_index_col_name, self.data.data.obs)
self.assertEqual(list(self.data.data.obs.index), list(range(2638)))
var_index_col_name = self.data.get_schema()["annotations"]["var"]["index"]
self.assertIn(var_index_col_name, self.data.data.var)
self.assertEqual(list(self.data.data.var.index), list(range(1838)))
@pytest.mark.filterwarnings("ignore:Anndata data matrix")
def test_data_type(self):
# don't run the test on the more exotic data types, as they don't
# support the astype() interface (used by this test, but not underlying app)
if isinstance(self.data.data.X, np.ndarray):
self.data.data.X = self.data.data.X.astype("float64")
with self.assertWarns(UserWarning):
self.data._validate_data_types()
def test_filter_idx(self):
filter_ = {"filter": {"var": {"index": [1, 99, [200, 300]]}}}
fbs = self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
data = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(data["n_rows"], 2638)
self.assertEqual(data["n_cols"], 102)
def test_filter_complex(self):
filter_ = {
"filter": {"var": {"annotation_value": [{"name": "n_cells", "min": 10}], "index": [1, 99, [200, 300]]}}
}
fbs = self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
data = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(data["n_rows"], 2638)
self.assertEqual(data["n_cols"], 91)
def test_obs_and_var_names(self):
self.assertEqual(np.sum(self.data.data.var[self.data.get_schema()["annotations"]["var"]["index"]].isna()), 0)
self.assertEqual(np.sum(self.data.data.obs[self.data.get_schema()["annotations"]["obs"]["index"]].isna()), 0)
def test_get_colors(self):
self.assertEqual(self.data.get_colors(), pbmc3k_colors)
def test_get_schema(self):
with open(f"{FIXTURES_ROOT}/schema.json") as fh:
schema = json.load(fh)
self.assertDictEqual(self.data.get_schema(), schema)
def test_schema_produces_error(self):
self.data.data.obs["time"] = pd.Series(
list([time.time() for i in range(self.data.cell_count)]), dtype="datetime64[ns]",
)
with pytest.raises(TypeError):
self.data._create_schema()
def test_layout(self):
fbs = self.data.layout_to_fbs_matrix(fields=None)
layout = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(layout["n_cols"], 6)
self.assertEqual(layout["n_rows"], 2638)
X = layout["columns"][0]
self.assertTrue((X >= 0).all() and (X <= 1).all())
Y = layout["columns"][1]
self.assertTrue((Y >= 0).all() and (Y <= 1).all())
def test_layout_fields(self):
""" X_pca, X_tsne, X_umap are available """
fbs = self.data.layout_to_fbs_matrix(["pca"])
layout = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(layout["n_cols"], 2)
self.assertEqual(layout["n_rows"], 2638)
self.assertCountEqual(layout["col_idx"], ["pca_0", "pca_1"])
fbs = self.data.layout_to_fbs_matrix(["tsne", "pca"])
layout = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(layout["n_cols"], 4)
self.assertEqual(layout["n_rows"], 2638)
self.assertCountEqual(layout["col_idx"], ["tsne_0", "tsne_1", "pca_0", "pca_1"])
def test_annotations(self):
fbs = self.data.annotation_to_fbs_matrix("obs")
annotations = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(annotations["n_rows"], 2638)
self.assertEqual(annotations["n_cols"], 5)
obs_index_col_name = self.data.get_schema()["annotations"]["obs"]["index"]
self.assertEqual(
annotations["col_idx"], [obs_index_col_name, "n_genes", "percent_mito", "n_counts", "louvain"],
)
fbs = self.data.annotation_to_fbs_matrix("var")
annotations = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(annotations["n_rows"], 1838)
self.assertEqual(annotations["n_cols"], 2)
var_index_col_name = self.data.get_schema()["annotations"]["var"]["index"]
self.assertEqual(annotations["col_idx"], [var_index_col_name, "n_cells"])
def test_annotation_fields(self):
fbs = self.data.annotation_to_fbs_matrix("obs", ["n_genes", "n_counts"])
annotations = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(annotations["n_rows"], 2638)
self.assertEqual(annotations["n_cols"], 2)
var_index_col_name = self.data.get_schema()["annotations"]["var"]["index"]
fbs = self.data.annotation_to_fbs_matrix("var", [var_index_col_name])
annotations = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(annotations["n_rows"], 1838)
self.assertEqual(annotations["n_cols"], 1)
def test_diffexp_topN(self):
f1 = {"filter": {"obs": {"index": [[0, 500]]}}}
f2 = {"filter": {"obs": {"index": [[500, 1000]]}}}
result = json.loads(self.data.diffexp_topN(f1["filter"], f2["filter"]))
self.assertEqual(len(result), 10)
result = json.loads(self.data.diffexp_topN(f1["filter"], f2["filter"], 20))
self.assertEqual(len(result), 20)
def test_data_frame(self):
f1 = {"var": {"index": [[0, 10]]}}
fbs = self.data.data_frame_to_fbs_matrix(f1, "var")
data = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(data["n_rows"], 2638)
self.assertEqual(data["n_cols"], 10)
with self.assertRaises(ValueError):
self.data.data_frame_to_fbs_matrix(None, "obs")
def test_filtered_data_frame(self):
filter_ = {"filter": {"var": {"annotation_value": [{"name": "n_cells", "min": 100}]}}}
fbs = self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
data = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(data["n_rows"], 2638)
self.assertEqual(data["n_cols"], 1040)
filter_ = {"filter": {"obs": {"annotation_value": [{"name": "n_counts", "min": 3000}]}}}
with self.assertRaises(FilterError):
self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
def test_data_named_gene(self):
var_index_col_name = self.data.get_schema()["annotations"]["var"]["index"]
filter_ = {"filter": {"var": {"annotation_value": [{"name": var_index_col_name, "values": ["RER1"]}]}}}
fbs = self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
data = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(data["n_rows"], 2638)
self.assertEqual(data["n_cols"], 1)
self.assertEqual(data["col_idx"], [4])
filter_ = {
"filter": {"var": {"annotation_value": [{"name": var_index_col_name, "values": ["SPEN", "TYMP", "PRMT2"]}]}}
}
fbs = self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
data = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(data["n_rows"], 2638)
self.assertEqual(data["n_cols"], 3)
self.assertTrue((data["col_idx"] == [15, 1818, 1837]).all())
def test_compute_embedding(self):
filter = {"obs": {"index": [[0, 100]]}}
# Verify that we correctly handle the case where we lack scanpy
import unittest.mock
with unittest.mock.patch.dict(sys.modules, {"scanpy": None}):
with self.assertRaises(NotImplementedError):
self.data.compute_embedding("umap", filter)
# if we happen to have scanpy, test the full API, else punt
import importlib
scanpy_spec = importlib.util.find_spec("scanpy")
if scanpy_spec is None:
print("Skipping compute_embedding test as ScanPy not installed")
return
# this feature is unsupported in backed mode, and we expect an error
if self.data.data.isbacked:
with self.assertRaises(NotImplementedError):
self.data.compute_embedding("umap", filter)
return
schema = self.data.compute_embedding("umap", filter)
self.assertIsInstance(schema["name"], str)
name = schema["name"]
self.assertEqual(schema["type"], "float32")
self.assertEqual(schema["dims"], [f"{name}_0", f"{name}_1"])
emb = self.data.data.obsm[f"X_{name}"]
self.assertEqual(emb.shape, (2638, 2))
self.assertTrue(np.isfinite(emb[0:100]).all())
self.assertTrue(np.isnan(emb[100:]).all())
@@ -0,0 +1,81 @@
import unittest
import json
from local_server.data_anndata.anndata_adaptor import AnndataAdaptor
from local_server.common.data_locator import DataLocator
from local_server.common.config.app_config import AppConfig
from local_server.test import PROJECT_ROOT
class DataLoadAdaptorTest(unittest.TestCase):
"""
Test file loading, including deferred loading/update.
"""
def setUp(self):
self.data_file = DataLocator(f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad")
config = AppConfig()
config.update_server_config(single_dataset__datapath=self.data_file.path)
config.update_server_config(app__flask_secret_key="secret")
config.complete_config()
self.data = AnndataAdaptor(self.data_file, config)
def test_delayed_load_data(self):
self.data._create_schema()
self.assertEqual(self.data.cell_count, 2638)
self.assertEqual(self.data.gene_count, 1838)
epsilon = 0.000_005
self.assertTrue(self.data.data.X[0, 0] - -0.171_469_51 < epsilon)
def test_diffexp_topN(self):
f1 = {"filter": {"obs": {"index": [[0, 500]]}}}
f2 = {"filter": {"obs": {"index": [[500, 1000]]}}}
result = json.loads(self.data.diffexp_topN(f1["filter"], f2["filter"]))
self.assertEqual(len(result), 10)
result = json.loads(self.data.diffexp_topN(f1["filter"], f2["filter"], 20))
self.assertEqual(len(result), 20)
class DataLocatorAdaptorTest(unittest.TestCase):
"""
Test various types of data locators we expect to consume
"""
def get_basic_config(self):
config = AppConfig()
config.update_server_config(
single_dataset__obs_names=None, single_dataset__var_names=None,
)
config.update_server_config(app__flask_secret_key="secret")
config.update_dataset_config(
embeddings__names=["umap"], presentation__max_categories=100, diffexp__lfc_cutoff=0.01,
)
return config
def stdAsserts(self, data):
""" run these each time we load the data """
self.assertIsNotNone(data)
self.assertEqual(data.cell_count, 2638)
self.assertEqual(data.gene_count, 1838)
def test_posix_file(self):
locator = DataLocator("../example-dataset/pbmc3k.h5ad")
config = self.get_basic_config()
config.update_server_config(single_dataset__datapath=locator.path)
config.complete_config()
data = AnndataAdaptor(locator, config)
self.stdAsserts(data)
def test_url_https(self):
url = "https://raw.githubusercontent.com/chanzuckerberg/cellxgene/main/example-dataset/pbmc3k.h5ad"
locator = DataLocator(url)
config = self.get_basic_config()
data = AnndataAdaptor(locator, config)
self.stdAsserts(data)
def test_url_http(self):
url = "http://raw.githubusercontent.com/chanzuckerberg/cellxgene/main/example-dataset/pbmc3k.h5ad"
locator = DataLocator(url)
config = self.get_basic_config()
data = AnndataAdaptor(locator, config)
self.stdAsserts(data)
@@ -0,0 +1,64 @@
import math
import unittest
import warnings
import pytest
import local_server.test.unit.decode_fbs as decode_fbs
from local_server.common.data_locator import DataLocator
from local_server.common.errors import FilterError
from local_server.data_anndata.anndata_adaptor import AnndataAdaptor
from local_server.test import app_config, FIXTURES_ROOT
class NaNTest(unittest.TestCase):
def setUp(self):
self.data_locator = DataLocator(f"{FIXTURES_ROOT}/nan.h5ad")
self.config = app_config(self.data_locator.path)
with warnings.catch_warnings():
warnings.simplefilter("ignore", category=UserWarning)
self.data = AnndataAdaptor(self.data_locator, self.config)
self.data._create_schema()
def test_load(self):
with self.assertLogs(level="WARN") as logger:
self.data = AnndataAdaptor(self.data_locator, self.config)
self.assertTrue(logger.output)
def test_init(self):
self.assertEqual(self.data.cell_count, 100)
self.assertEqual(self.data.gene_count, 100)
epsilon = 0.000_005
self.assertTrue(self.data.data.X[0, 0] - -0.171_469_51 < epsilon)
def test_dataframe(self):
data_frame_var = decode_fbs.decode_matrix_FBS(self.data.data_frame_to_fbs_matrix(None, "var"))
self.assertIsNotNone(data_frame_var)
self.assertEqual(data_frame_var["n_rows"], 100)
self.assertEqual(data_frame_var["n_cols"], 100)
self.assertTrue(math.isnan(data_frame_var["columns"][3][3]))
with pytest.raises(FilterError):
self.data.data_frame_to_fbs_matrix("an erroneous filter", "var")
with pytest.raises(FilterError):
filter_ = {"filter": {"obs": {"index": [1, 99, [200, 300]]}}}
self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
def test_dataframe_obs_not_implemented(self):
with self.assertRaises(ValueError) as cm:
decode_fbs.decode_matrix_FBS(self.data.data_frame_to_fbs_matrix(None, "obs"))
self.assertIsNotNone(cm.exception)
def test_annotation(self):
annotations = decode_fbs.decode_matrix_FBS(self.data.annotation_to_fbs_matrix("obs"))
obs_index_col_name = self.data.schema["annotations"]["obs"]["index"]
self.assertEqual(annotations["col_idx"], [obs_index_col_name, "n_genes", "percent_mito", "n_counts", "louvain"])
self.assertEqual(annotations["n_rows"], 100)
self.assertTrue(math.isnan(annotations["columns"][2][0]))
annotations = decode_fbs.decode_matrix_FBS(self.data.annotation_to_fbs_matrix("var"))
var_index_col_name = self.data.schema["annotations"]["var"]["index"]
self.assertEqual(annotations["col_idx"], [var_index_col_name, "n_cells", "var_with_nans"])
self.assertEqual(annotations["n_rows"], 100)
self.assertTrue(math.isnan(annotations["columns"][2][0]))
@@ -0,0 +1,82 @@
import unittest
import pandas as pd
import numpy as np
from scipy import sparse
import local_server.test.unit.decode_fbs as decode_fbs
from local_server.data_common.fbs.matrix import encode_matrix_fbs, decode_matrix_fbs
class FbsTests(unittest.TestCase):
"""Test Case for Matrix FBS data encode/decode """
def test_encode_boundary(self):
""" test various boundary checks """
# row indexing is unsupported
with self.assertRaises(ValueError):
encode_matrix_fbs(matrix=pd.DataFrame(), row_idx=[])
# matrix must be 2D
with self.assertRaises(ValueError):
encode_matrix_fbs(matrix=np.zeros((3, 2, 1)))
with self.assertRaises(ValueError):
encode_matrix_fbs(matrix=np.ones((10,)))
def fbs_checks(self, fbs, dims, expected_types, expected_column_idx):
d = decode_fbs.decode_matrix_FBS(fbs)
self.assertEqual(d["n_rows"], dims[0])
self.assertEqual(d["n_cols"], dims[1])
self.assertIsNone(d["row_idx"])
self.assertEqual(len(d["columns"]), dims[1])
for i in range(0, len(d["columns"])):
self.assertEqual(len(d["columns"][i]), dims[0])
self.assertIsInstance(d["columns"][i], expected_types[i][0])
if expected_types[i][1] is not None:
self.assertEqual(d["columns"][i].dtype, expected_types[i][1])
if expected_column_idx is not None:
self.assertSetEqual(set(expected_column_idx), set(d["col_idx"]))
def test_encode_DataFrame(self):
df = pd.DataFrame(
data={
"a": np.zeros((10,), dtype=np.float32),
"b": np.ones((10,), dtype=np.int64),
"c": np.array([i for i in range(0, 10)], dtype=np.uint16),
"d": pd.Series(["x", "y", "z", "x", "y", "z", "a", "x", "y", "z"], dtype="category"),
}
)
expected_types = ((np.ndarray, np.float32), (np.ndarray, np.int32), (np.ndarray, np.uint32), (list, None))
fbs = encode_matrix_fbs(matrix=df, row_idx=None, col_idx=df.columns)
self.fbs_checks(fbs, (10, 4), expected_types, ["a", "b", "c", "d"])
def test_encode_ndarray(self):
arr = np.zeros((3, 2), dtype=np.float32)
expected_types = ((np.ndarray, np.float32), (np.ndarray, np.float32), (np.ndarray, np.float32))
fbs = encode_matrix_fbs(matrix=arr, row_idx=None, col_idx=None)
self.fbs_checks(fbs, (3, 2), expected_types, None)
def test_encode_sparse(self):
csc = sparse.csc_matrix(np.array([[0, 1, 2], [3, 0, 4]]))
expected_types = ((np.ndarray, np.int32), (np.ndarray, np.int32), (np.ndarray, np.int32))
fbs = encode_matrix_fbs(matrix=csc, row_idx=None, col_idx=None)
self.fbs_checks(fbs, (2, 3), expected_types, None)
def test_roundtrip(self):
dfSrc = pd.DataFrame(
data={
"a": np.zeros((10,), dtype=np.float32),
"b": np.ones((10,), dtype=np.int64),
"c": np.array([i for i in range(0, 10)], dtype=np.uint16),
"d": pd.Series(["x", "y", "z", "x", "y", "z", "a", "x", "y", "z"], dtype="category"),
}
)
dfDst = decode_matrix_fbs(encode_matrix_fbs(matrix=dfSrc, col_idx=dfSrc.columns))
self.assertEqual(dfSrc.shape, dfDst.shape)
self.assertEqual(set(dfSrc.columns), set(dfDst.columns))
for c in dfSrc.columns:
self.assertTrue(c in dfDst.columns)
if isinstance(dfSrc[c], pd.Series):
self.assertTrue(np.all(dfSrc[c] == dfDst[c]))
else:
self.assertEqual(dfSrc[c], dfDst[c])
+31
View File
@@ -0,0 +1,31 @@
"""
Code to decode, for testing purposes, the flatbuffer encoded blobs.
This code will need to be updated if fbs/matrix.fbs changes. For more information, see fbs/matrix.fbs and
local_server/data_common/fbs/
"""
import local_server.data_common.fbs.NetEncoding.Matrix as Matrix
from local_server.data_common.fbs.matrix import deserialize_typed_array
def decode_matrix_FBS(buf):
"""
Given a FBS Matrix, return an decoded Python dict containing same info in native format.
NOTE / TODO: row_idx not currently implemented
"""
df = Matrix.Matrix.GetRootAsMatrix(buf, 0)
n_rows = df.NRows()
n_cols = df.NCols()
columns_length = df.ColumnsLength()
decoded_columns = []
for col_idx in range(0, columns_length):
col = df.Columns(col_idx)
tarr = (col.UType(), col.U())
decoded_columns.append(deserialize_typed_array(tarr))
cidx = deserialize_typed_array((df.ColIndexType(), df.ColIndex()))
return {"n_rows": n_rows, "n_cols": n_cols, "columns": decoded_columns, "col_idx": cidx, "row_idx": None}