mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-09-15 20:57:56 +08:00
876 lines
36 KiB
Python
876 lines
36 KiB
Python
import shutil
|
|
import time
|
|
import unittest
|
|
import zlib
|
|
from http import HTTPStatus
|
|
import tempfile
|
|
from os import path
|
|
import hashlib
|
|
from os.path import basename, splitext
|
|
|
|
import pandas as pd
|
|
import requests
|
|
import numpy as np
|
|
|
|
from parameterized import parameterized_class
|
|
|
|
import test.decode_fbs as decode_fbs
|
|
from server.data_common.matrix_loader import MatrixDataType
|
|
from test.unit import (
|
|
data_with_tmp_annotations,
|
|
make_fbs,
|
|
start_test_server,
|
|
stop_test_server,
|
|
)
|
|
from test.fixtures.fixtures import pbmc3k_colors
|
|
from test import PROJECT_ROOT, FIXTURES_ROOT
|
|
|
|
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
|
|
GENESETS_READONLY = False
|
|
|
|
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
|
|
)
|
|
|
|
# Check that all schema types are legal
|
|
legal_types = ["boolean", "string", "categorical", "float32", "int32"]
|
|
self.assertEqual(result_data["schema"]["dataframe"]["type"], "float32")
|
|
for column in result_data["schema"]["annotations"]["obs"]["columns"]:
|
|
self.assertIn(column["type"], legal_types)
|
|
for column in result_data["schema"]["annotations"]["var"]["columns"]:
|
|
self.assertIn(column["type"], legal_types)
|
|
|
|
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"])
|
|
|
|
if hasattr(self, "data_locator"):
|
|
title = splitext(basename(self.data_locator))[0]
|
|
else:
|
|
title = "pbmc3k"
|
|
self.assertEqual(result_data["config"]["displayNames"]["dataset"], title)
|
|
self.assertIsNotNone(result_data["config"]["parameters"])
|
|
|
|
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"])
|
|
for column in df["columns"]:
|
|
self.assertEqual(column.dtype, np.float32)
|
|
|
|
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 []),
|
|
)
|
|
for column in df["columns"]:
|
|
if type(column) is np.ndarray:
|
|
self.assertIn(column.dtype, [np.float32, np.int32])
|
|
|
|
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"])
|
|
for column in df["columns"]:
|
|
if type(column) is np.ndarray:
|
|
self.assertIn(column.dtype, [np.float32, np.int32])
|
|
|
|
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}"
|
|
headers = {"Content-Type": "application/json"}
|
|
result = self.session.put(url, headers=headers)
|
|
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
|
|
|
|
filter = {"filter": {"var": {"index": [0, 1, 4]}}}
|
|
result = self.session.put(url, json=filter, headers=headers)
|
|
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", "Content-Type": "application/json"}
|
|
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])
|
|
for column in df["columns"]:
|
|
if type(column) is np.ndarray:
|
|
self.assertIn(column.dtype, [np.float32, np.int32])
|
|
|
|
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)
|
|
for column in df["columns"]:
|
|
if type(column) is np.ndarray:
|
|
self.assertIn(column.dtype, [np.float32, np.int32])
|
|
|
|
@unittest.skip("This test is currently broken after upgrading Werkzeug.")
|
|
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)
|
|
for column in df["columns"]:
|
|
if type(column) is np.ndarray:
|
|
self.assertIn(column.dtype, [np.float32, np.int32])
|
|
|
|
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)
|
|
|
|
@unittest.skip("needs fix: https://github.com/chanzuckerberg/cellxgene/issues/2542")
|
|
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 test_genesets_config(self):
|
|
result = self.session.get(f"{self.URL_BASE}config")
|
|
config_data = result.json()
|
|
params = config_data["config"]["parameters"]
|
|
annotations_genesets = params["annotations_genesets"]
|
|
annotations_genesets_readonly = params["annotations_genesets_readonly"]
|
|
annotations_genesets_summary_methods = params["annotations_genesets_summary_methods"]
|
|
self.assertTrue(annotations_genesets)
|
|
self.assertEqual(annotations_genesets_readonly, self.GENESETS_READONLY)
|
|
self.assertEqual(annotations_genesets_summary_methods, ["mean"])
|
|
|
|
def test_get_genesets(self):
|
|
endpoint = "genesets"
|
|
url = f"{self.URL_BASE}{endpoint}"
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.headers["Content-Type"], "application/json")
|
|
result_data = result.json()
|
|
self.assertIsNotNone(result_data["genesets"])
|
|
|
|
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"])
|
|
|
|
|
|
@parameterized_class(
|
|
[
|
|
{"data_locator": f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad"},
|
|
{"data_locator": f"{FIXTURES_ROOT}/pbmc3k_64.h5ad"},
|
|
{"data_locator": f"{FIXTURES_ROOT}/pbmc3k-CSC-gz.h5ad"},
|
|
{"data_locator": f"{FIXTURES_ROOT}/pbmc3k-CSR-gz.h5ad"},
|
|
]
|
|
)
|
|
class EndPointsAnndata(unittest.TestCase, EndPoints):
|
|
"""Test Case for endpoints"""
|
|
|
|
ANNOTATIONS_ENABLED = False
|
|
GENESETS_READONLY = True
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if cls == EndPointsAnndata:
|
|
raise unittest.SkipTest("`parameterized_class` bug")
|
|
|
|
cls._setupClass(
|
|
cls,
|
|
[
|
|
cls.data_locator,
|
|
"--disable-annotations",
|
|
"--disable-gene-sets-save",
|
|
],
|
|
)
|
|
|
|
@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["positive"]), 7)
|
|
self.assertEqual(len(result_data["negative"]), 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["positive"]), 10)
|
|
self.assertEqual(len(result_data["negative"]), 10)
|
|
|
|
def test_get_summaryvar(self):
|
|
index_col_name = self.schema["schema"]["annotations"]["var"]["index"]
|
|
endpoint = "summarize/var"
|
|
|
|
# single column
|
|
filter = f"var:{index_col_name}=F5"
|
|
query = f"method=mean&{filter}"
|
|
query_hash = hashlib.sha1(query.encode()).hexdigest()
|
|
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.assertEqual(df["col_idx"], [query_hash])
|
|
self.assertAlmostEqual(df["columns"][0][0], -0.110451095)
|
|
|
|
# multi-column
|
|
col_names = ["F5", "BEB3", "SIK1"]
|
|
filter = "&".join([f"var:{index_col_name}={name}" for name in col_names])
|
|
query = f"method=mean&{filter}"
|
|
query_hash = hashlib.sha1(query.encode()).hexdigest()
|
|
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.assertEqual(df["col_idx"], [query_hash])
|
|
self.assertAlmostEqual(df["columns"][0][0], -0.16628358)
|
|
|
|
def test_post_summaryvar(self):
|
|
index_col_name = self.schema["schema"]["annotations"]["var"]["index"]
|
|
endpoint = "summarize/var"
|
|
headers = {"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/octet-stream"}
|
|
|
|
# single column
|
|
filter = f"var:{index_col_name}=F5"
|
|
query = f"method=mean&{filter}"
|
|
query_hash = hashlib.sha1(query.encode()).hexdigest()
|
|
url = f"{self.URL_BASE}{endpoint}?key={query_hash}"
|
|
result = self.session.post(url, headers=headers, data=query)
|
|
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.assertEqual(df["col_idx"], [query_hash])
|
|
self.assertAlmostEqual(df["columns"][0][0], -0.110451095)
|
|
|
|
# multi-column
|
|
col_names = ["F5", "BEB3", "SIK1"]
|
|
filter = "&".join([f"var:{index_col_name}={name}" for name in col_names])
|
|
query = f"method=mean&{filter}"
|
|
query_hash = hashlib.sha1(query.encode()).hexdigest()
|
|
url = f"{self.URL_BASE}{endpoint}?key={query_hash}"
|
|
result = self.session.post(url, headers=headers, data=query)
|
|
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.assertEqual(df["col_idx"], [query_hash])
|
|
self.assertAlmostEqual(df["columns"][0][0], -0.16628358)
|
|
|
|
|
|
class EndPointsAnndataAnnotations(unittest.TestCase, EndPointsAnnotations):
|
|
"""Test Case for endpoints"""
|
|
|
|
ANNOTATIONS_ENABLED = True
|
|
GENESETS_READONLY = False
|
|
|
|
@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.label_output_file, cls.data.get_location()])
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
shutil.rmtree(cls.tmp_dir)
|
|
stop_test_server(cls.ps)
|
|
|
|
|
|
class EndPointsAnnDataGenesets(unittest.TestCase, EndPoints):
|
|
ANNOTATIONS_ENABLED = False
|
|
GENESETS_READONLY = False
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmp_dir = tempfile.mkdtemp()
|
|
genesets_file = path.join(cls.tmp_dir, "test_genesets.csv")
|
|
shutil.copyfile(f"{FIXTURES_ROOT}/pbmc3k-genesets.csv", genesets_file)
|
|
cls._setupClass(
|
|
cls,
|
|
[
|
|
f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad",
|
|
"--disable-annotations",
|
|
"--gene-sets-file",
|
|
genesets_file,
|
|
],
|
|
)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
shutil.rmtree(cls.tmp_dir)
|
|
stop_test_server(cls.ps)
|
|
|
|
def test_get_genesets_json(self):
|
|
endpoint = "genesets"
|
|
url = f"{self.URL_BASE}{endpoint}"
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.headers["Content-Type"], "application/json")
|
|
result_data = result.json()
|
|
self.assertIsNotNone(result_data["genesets"])
|
|
self.assertIsNotNone(result_data["tid"])
|
|
|
|
self.assertEqual(
|
|
result_data,
|
|
{
|
|
"genesets": [
|
|
{
|
|
"genes": [
|
|
{"gene_description": " a gene_description", "gene_symbol": "F5"},
|
|
{"gene_description": "", "gene_symbol": "SUMO3"},
|
|
{"gene_description": "", "gene_symbol": "SRM"},
|
|
],
|
|
"geneset_description": "a description",
|
|
"geneset_name": "first gene set name",
|
|
},
|
|
{
|
|
"genes": [
|
|
{"gene_description": "", "gene_symbol": "RER1"},
|
|
{"gene_description": "", "gene_symbol": "SIK1"},
|
|
],
|
|
"geneset_description": "",
|
|
"geneset_name": "second_gene_set",
|
|
},
|
|
{"genes": [], "geneset_description": "", "geneset_name": "third gene set"},
|
|
{"genes": [], "geneset_description": "fourth description", "geneset_name": "fourth_gene_set"},
|
|
{"genes": [], "geneset_description": "", "geneset_name": "fifth_dataset"},
|
|
{
|
|
"genes": [
|
|
{"gene_description": "", "gene_symbol": "ACD"},
|
|
{"gene_description": "", "gene_symbol": "AATF"},
|
|
{"gene_description": "", "gene_symbol": "F5"},
|
|
{"gene_description": "", "gene_symbol": "PIGU"},
|
|
],
|
|
"geneset_description": "",
|
|
"geneset_name": "summary test",
|
|
},
|
|
{"genes": [], "geneset_description": "", "geneset_name": "geneset_to_delete"},
|
|
{"genes": [], "geneset_description": "", "geneset_name": "geneset_to_edit"},
|
|
{
|
|
"genes": [],
|
|
"geneset_description": "",
|
|
"geneset_name": "fill_this_geneset",
|
|
},
|
|
{
|
|
"genes": [{"gene_description": "", "gene_symbol": "SIK1"}],
|
|
"geneset_description": "",
|
|
"geneset_name": "empty_this_geneset",
|
|
},
|
|
{
|
|
"genes": [{"gene_description": "", "gene_symbol": "SIK1"}],
|
|
"geneset_description": "",
|
|
"geneset_name": "brush_this_gene",
|
|
},
|
|
],
|
|
"tid": 0,
|
|
},
|
|
)
|
|
|
|
def test_get_genesets_csv(self):
|
|
endpoint = "genesets"
|
|
url = f"{self.URL_BASE}{endpoint}"
|
|
result = self.session.get(url, headers={"Accept": "text/csv"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.headers["Content-Type"], "text/csv")
|
|
self.assertEqual(
|
|
result.text,
|
|
"""gene_set_name,gene_set_description,gene_symbol,gene_description\r
|
|
first gene set name,a description,F5, a gene_description\r
|
|
first gene set name,a description,SUMO3,\r
|
|
first gene set name,a description,SRM,\r
|
|
second_gene_set,,RER1,\r
|
|
second_gene_set,,SIK1,\r
|
|
third gene set,,,\r
|
|
fourth_gene_set,fourth description,,\r
|
|
fifth_dataset,,,\r
|
|
summary test,,ACD,\r
|
|
summary test,,AATF,\r
|
|
summary test,,F5,\r
|
|
summary test,,PIGU,\r
|
|
geneset_to_delete,,,\r
|
|
geneset_to_edit,,,\r
|
|
fill_this_geneset,,,\r
|
|
empty_this_geneset,,SIK1,\r
|
|
brush_this_gene,,SIK1,\r
|
|
""",
|
|
)
|
|
|
|
def test_put_genesets(self):
|
|
endpoint = "genesets"
|
|
url = f"{self.URL_BASE}{endpoint}"
|
|
|
|
# assume we start with TID 0
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.json()["tid"], 0)
|
|
|
|
test1 = {"tid": 3, "genesets": []}
|
|
result = self.session.put(url, json=test1)
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.json(), test1)
|
|
|
|
# stale TID
|
|
result = self.session.put(url, json=test1)
|
|
self.assertEqual(result.status_code, HTTPStatus.NOT_FOUND)
|
|
|
|
test2 = {
|
|
"tid": 4,
|
|
"genesets": [
|
|
{"geneset_name": "foobar", "genes": []},
|
|
{"geneset_name": "contains a space", "genes": []},
|
|
{"geneset_name": "contains_weird_characters: #$%^&*()_+=-!@<>,./?';:\"[]{}|\\", "genes": []},
|
|
],
|
|
}
|
|
test2_response = {
|
|
"tid": 4,
|
|
"genesets": [
|
|
{"geneset_name": "foobar", "geneset_description": "", "genes": []},
|
|
{"geneset_name": "contains a space", "geneset_description": "", "genes": []},
|
|
{
|
|
"geneset_name": "contains_weird_characters: #$%^&*()_+=-!@<>,./?';:\"[]{}|\\",
|
|
"geneset_description": "",
|
|
"genes": [],
|
|
},
|
|
],
|
|
}
|
|
result = self.session.put(url, json=test2)
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.json(), test2_response)
|
|
|
|
test3 = {
|
|
"tid": 5,
|
|
"genesets": [
|
|
{
|
|
"geneset_name": "foobar",
|
|
"geneset_description": "",
|
|
"genes": [
|
|
{
|
|
"gene_symbol": "F5",
|
|
"gene_description": "",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
result = self.session.put(url, json=test3)
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.json(), test3)
|
|
|
|
def test_put_genesets_malformed(self):
|
|
"""test malformed submissions that we expect the backend to catch/tolerate"""
|
|
endpoint = "genesets"
|
|
url = f"{self.URL_BASE}{endpoint}"
|
|
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
original_data = result.json()
|
|
tid = original_data["tid"]
|
|
|
|
def test_case(test, expected_code, original_data):
|
|
"""check for expected error AND that no change was made to the original state"""
|
|
result = self.session.put(url, json=test)
|
|
self.assertEqual(result.status_code, expected_code)
|
|
result = self.session.get(url, headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
self.assertEqual(result.json(), original_data)
|
|
|
|
# missing or malformed genesets
|
|
test_case(
|
|
{"tid": tid + 1},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": 99},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
|
|
# illegal geneset_name
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"geneset_name": " foo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"geneset_name": "foo ", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"geneset_name": "f oo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"geneset_name": "f\too", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"geneset_name": "f\roo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"geneset_name": "f\noo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"geneset_name": "f\voo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
|
|
# duplicate geneset_name
|
|
test_case(
|
|
{
|
|
"tid": tid + 1,
|
|
"genesets": [
|
|
{"geneset_name": "foo", "genes": []},
|
|
{"geneset_name": "foo", "genes": []},
|
|
],
|
|
},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
|
|
# missing geneset_name
|
|
test_case(
|
|
{"tid": tid + 1, "genesets": [{"genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
|
|
# non-numeric TID
|
|
test_case(
|
|
{"tid": [], "genesets": [{"geneset_name": "foo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": None, "genesets": [{"geneset_name": "foo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
test_case(
|
|
{"tid": "not a number", "genesets": [{"geneset_name": "foo", "genes": []}]},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
|
|
# duplicate gene_symbol
|
|
test_case(
|
|
{
|
|
"tid": "not a number",
|
|
"genesets": [{"geneset_name": "foo", "genes": [{"gene_symbol": "SIK1"}, {"gene_symbol": "SIK1"}]}],
|
|
},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
|
|
# gene_symbol is not a string
|
|
test_case(
|
|
{
|
|
"tid": "not a number",
|
|
"genesets": [{"geneset_name": "foo", "genes": [{"gene_symbol": 99}]}],
|
|
},
|
|
HTTPStatus.BAD_REQUEST,
|
|
original_data,
|
|
)
|
|
|
|
def test_get_geneset_summary_edge_cases(self):
|
|
# attempt to summarize _all_ genesets, including edge cases with zero or one gene
|
|
result = self.session.get(f"{self.URL_BASE}genesets", headers={"Accept": "application/json"})
|
|
self.assertEqual(result.status_code, HTTPStatus.OK)
|
|
genesets = result.json()["genesets"]
|
|
|
|
endpoint = "summarize/var"
|
|
index_col_name = self.schema["schema"]["annotations"]["var"]["index"]
|
|
for gs in genesets:
|
|
genes = [g["gene_symbol"] for g in gs["genes"]]
|
|
filter = "&".join([f"var:{index_col_name}={gene}" for gene in genes])
|
|
query = f"method=mean&{filter}"
|
|
query_hash = hashlib.sha1(query.encode()).hexdigest()
|
|
url = f"{self.URL_BASE}{endpoint}?{query}"
|
|
|
|
result = self.session.get(url, headers={"Accept": "application/octet-stream"})
|
|
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.assertEqual(df["col_idx"], [query_hash])
|