mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-09-30 00:28:11 +08:00
Flatbuffer cleanup (#598)
* dead code and route removal * more dead code cleanup * fix scanpy_engine tests * lint * add missing catch in filter parsing * update scanpy NaN tests * more fbs tests and dead test removal * remove forced default for content type negotiation * bit of cleanup * more fbs test cleanup * lint * remove swagger * swagger cleanup * lint * correctly handle lack of templates * more dead code removal * remove unused files * fix dev build * lint
This commit is contained in:
Bruce Martin
+71
-292
@@ -57,16 +57,6 @@ class EndPoints(unittest.TestCase):
|
||||
self.assertEqual(result_data["config"]["displayNames"]["dataset"], "pbmc3k")
|
||||
self.assertEqual(len(result_data["config"]["features"]), 4)
|
||||
|
||||
def test_get_layout(self):
|
||||
endpoint = "layout/obs"
|
||||
url = f"{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["layout"]["ndims"], 2)
|
||||
self.assertEqual(len(result_data["layout"]["coordinates"]), 2638)
|
||||
|
||||
def test_get_layout_fbs(self):
|
||||
endpoint = "layout/obs"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
@@ -82,53 +72,11 @@ class EndPoints(unittest.TestCase):
|
||||
self.assertIsNone(df['row_idx'])
|
||||
self.assertEqual(len(df['columns']), df['n_cols'])
|
||||
|
||||
# def test_put_layout(self):
|
||||
# endpoint = "layout/obs"
|
||||
# url = f"{URL_BASE}{endpoint}"
|
||||
# obs_filter = {
|
||||
# "filter": {
|
||||
# "obs": {
|
||||
# "annotation_value": [
|
||||
# {"name": "louvain", "values": ["NK cells", "CD8 T cells"]},
|
||||
# {"name": "n_counts", "min": 3000},
|
||||
# ],
|
||||
# "index": [1, 99, [1000, 2000]]
|
||||
# }
|
||||
# }
|
||||
# }
|
||||
# result = self.session.put(url, json=obs_filter)
|
||||
# self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
# result_data = result.json()
|
||||
# self.assertEqual(len(result_data["layout"]["coordinates"]), 15)
|
||||
|
||||
def test_bad_filter(self):
|
||||
endpoints = ["annotations/obs", "annotations/var", "data/obs", "data/var"]
|
||||
for endpoint in endpoints:
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
result = self.session.put(url, json=BAD_FILTER)
|
||||
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
|
||||
|
||||
def test_get_annotations_obs(self):
|
||||
endpoint = "annotations/obs"
|
||||
endpoint = "data/var"
|
||||
url = f"{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["names"], ["name", "n_genes", "percent_mito", "n_counts", "louvain"])
|
||||
self.assertEqual(len(result_data["data"]), 2638)
|
||||
self.assertEqual(len(result_data["data"][0]), 6)
|
||||
|
||||
def test_get_annotations_obs_keys(self):
|
||||
endpoint = "annotations/obs"
|
||||
query = "annotation-name=n_genes&annotation-name=percent_mito"
|
||||
url = f"{URL_BASE}{endpoint}?{query}"
|
||||
result = self.session.get(url)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
result_data = result.json()
|
||||
self.assertEqual(result_data["names"], ["n_genes", "percent_mito"])
|
||||
self.assertEqual(len(result_data["data"][0]), 3)
|
||||
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"
|
||||
@@ -146,6 +94,23 @@ class EndPoints(unittest.TestCase):
|
||||
self.assertEqual(len(df['columns']), df['n_cols'])
|
||||
self.assertListEqual(df['col_idx'], ['name', 'n_genes', 'percent_mito', 'n_counts', 'louvain'])
|
||||
|
||||
def test_get_annotations_obs_keys_fbs(self):
|
||||
endpoint = "annotations/obs"
|
||||
query = "annotation-name=n_genes&annotation-name=percent_mito"
|
||||
url = f"{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.assertIsNotNone(df['col_idx'])
|
||||
self.assertIsNone(df['row_idx'])
|
||||
self.assertEqual(len(df['columns']), df['n_cols'])
|
||||
self.assertListEqual(df['col_idx'], ['n_genes', 'percent_mito'])
|
||||
|
||||
def test_get_annotations_obs_error(self):
|
||||
endpoint = "annotations/obs"
|
||||
query = "annotation-name=notakey"
|
||||
@@ -153,50 +118,6 @@ class EndPoints(unittest.TestCase):
|
||||
result = self.session.get(url)
|
||||
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
|
||||
|
||||
def test_put_annotations_obs(self):
|
||||
endpoint = "annotations/obs"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
obs_filter = {
|
||||
"filter": {
|
||||
"obs": {
|
||||
"annotation_value": [
|
||||
{"name": "louvain", "values": ["NK cells", "CD8 T cells"]},
|
||||
{"name": "n_counts", "min": 3000},
|
||||
],
|
||||
"index": [1, 99, [1000, 2000]],
|
||||
}
|
||||
}
|
||||
}
|
||||
result = self.session.put(url, json=obs_filter)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
result_data = result.json()
|
||||
self.assertEqual(result_data["names"], ["name", "n_genes", "percent_mito", "n_counts", "louvain"])
|
||||
self.assertEqual(len(result_data["data"]), 15)
|
||||
|
||||
def test_filter_put_annotations_obs(self):
|
||||
endpoint = "annotations/obs"
|
||||
query = "annotation-name=n_genes&annotation-name=percent_mito"
|
||||
url = f"{URL_BASE}{endpoint}?{query}"
|
||||
obs_filter = {
|
||||
"filter": {
|
||||
"obs": {
|
||||
"annotation_value": [
|
||||
{"name": "louvain", "values": ["NK cells", "CD8 T cells"]},
|
||||
{"name": "n_counts", "min": 3000},
|
||||
],
|
||||
"index": [1, 99, [1000, 2000]],
|
||||
}
|
||||
}
|
||||
}
|
||||
result = self.session.put(url, json=obs_filter)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
result_data = result.json()
|
||||
self.assertEqual(result_data["names"], ["n_genes", "percent_mito"])
|
||||
self.assertEqual(len(result_data["data"][0]), 3)
|
||||
self.assertEqual(len(result_data["data"]), 15)
|
||||
|
||||
def test_diff_exp(self):
|
||||
endpoint = "diffexp/obs"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
@@ -227,28 +148,6 @@ class EndPoints(unittest.TestCase):
|
||||
result_data = result.json()
|
||||
self.assertEqual(len(result_data), 10)
|
||||
|
||||
def test_get_annotations_var(self):
|
||||
endpoint = "annotations/var"
|
||||
url = f"{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["names"], ["name", "n_cells"])
|
||||
self.assertEqual(len(result_data["data"]), 1838)
|
||||
self.assertEqual(len(result_data["data"][0]), 3)
|
||||
|
||||
def test_get_annotations_var_keys(self):
|
||||
endpoint = "annotations/var"
|
||||
query = "annotation-name=n_cells"
|
||||
url = f"{URL_BASE}{endpoint}?{query}"
|
||||
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["names"], ["n_cells"])
|
||||
self.assertEqual(len(result_data["data"][0]), 2)
|
||||
|
||||
def test_get_annotations_var_fbs(self):
|
||||
endpoint = "annotations/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
@@ -265,6 +164,23 @@ class EndPoints(unittest.TestCase):
|
||||
self.assertEqual(len(df['columns']), df['n_cols'])
|
||||
self.assertListEqual(df['col_idx'], ['name', 'n_cells'])
|
||||
|
||||
def test_get_annotations_var_keys_fbs(self):
|
||||
endpoint = "annotations/var"
|
||||
query = "annotation-name=n_cells"
|
||||
url = f"{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.assertIsNotNone(df['col_idx'])
|
||||
self.assertIsNone(df['row_idx'])
|
||||
self.assertEqual(len(df['columns']), df['n_cols'])
|
||||
self.assertListEqual(df['col_idx'], ['n_cells'])
|
||||
|
||||
def test_get_annotations_var_error(self):
|
||||
endpoint = "annotations/var"
|
||||
query = "annotation-name=notakey"
|
||||
@@ -272,95 +188,36 @@ class EndPoints(unittest.TestCase):
|
||||
result = self.session.get(url)
|
||||
self.assertEqual(result.status_code, HTTPStatus.BAD_REQUEST)
|
||||
|
||||
def test_put_annotations_var(self):
|
||||
endpoint = "annotations/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
var_filter = {"filter": {"var": {"annotation_value": [{"name": "name", "values": ["ATAD3C", "RER1"]}]}}}
|
||||
result = self.session.put(url, json=var_filter)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
result_data = result.json()
|
||||
self.assertEqual(result_data["names"], ["name", "n_cells"])
|
||||
self.assertEqual(len(result_data["data"]), 2)
|
||||
|
||||
def test_filter_put_annotations_var(self):
|
||||
endpoint = "annotations/var"
|
||||
query = "annotation-name=n_cells"
|
||||
url = f"{URL_BASE}{endpoint}?{query}"
|
||||
var_filter = {"filter": {"var": {"annotation_value": [{"name": "name", "values": ["ATAD3C", "RER1"]}]}}}
|
||||
result = self.session.put(url, json=var_filter)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
result_data = result.json()
|
||||
self.assertEqual(result_data["names"], ["n_cells"])
|
||||
self.assertEqual(len(result_data["data"][0]), 2)
|
||||
self.assertEqual(len(result_data["data"]), 2)
|
||||
|
||||
def test_get_data(self):
|
||||
for axis in ["obs", "var"]:
|
||||
endpoint = f"data/{axis}"
|
||||
query = "accept-type=application/json"
|
||||
url = f"{URL_BASE}{endpoint}?{query}"
|
||||
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(len(result_data["obs"]), 2638)
|
||||
|
||||
def test_data_mimetype_error(self):
|
||||
for axis in ["obs", "var"]:
|
||||
endpoint = f"data/{axis}"
|
||||
query = "accept-type=xxx"
|
||||
url = f"{URL_BASE}{endpoint}?{query}"
|
||||
result = self.session.get(url)
|
||||
self.assertEqual(result.status_code, HTTPStatus.NOT_ACCEPTABLE)
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
header = {"Accept": "sdkljfa;dsjalkj"}
|
||||
result = self.session.get(url, headers=header)
|
||||
self.assertEqual(result.status_code, HTTPStatus.NOT_ACCEPTABLE)
|
||||
endpoint = f"data/var"
|
||||
header = {"Accept": "xxx"}
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
result = self.session.put(url, headers=header)
|
||||
self.assertEqual(result.status_code, HTTPStatus.NOT_ACCEPTABLE)
|
||||
|
||||
def test_json_default(self):
|
||||
for axis in ["obs", "var"]:
|
||||
endpoint = f"data/{axis}"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
result = self.session.get(url)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
|
||||
def test_data_filter(self):
|
||||
for axis in ["obs", "var"]:
|
||||
endpoint = f"data/{axis}"
|
||||
query = "accept-type=application/json&obs:louvain=NK cells&obs:louvain=CD8 T cells&obs:n_counts=3000,*"
|
||||
url = f"{URL_BASE}{endpoint}?{query}"
|
||||
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(len(result_data["obs"]), 38)
|
||||
|
||||
def test_data_json_put(self):
|
||||
for axis in ["obs", "var"]:
|
||||
endpoint = f"data/{axis}"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
header = {"Accept": "application/json"}
|
||||
obs_filter = {
|
||||
"filter": {
|
||||
"obs": {
|
||||
"annotation_value": [
|
||||
{"name": "louvain", "values": ["NK cells", "CD8 T cells"]},
|
||||
{"name": "n_counts", "min": 3000},
|
||||
],
|
||||
"index": [1, 99, [1000, 2000]],
|
||||
}
|
||||
}
|
||||
}
|
||||
result = self.session.put(url, headers=header, json=obs_filter)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
result_data = result.json()
|
||||
self.assertEqual(len(result_data["obs"]), 15)
|
||||
def test_fbs_default(self):
|
||||
endpoint = f"data/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
result = self.session.put(url)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/octet-stream")
|
||||
|
||||
def test_data_put_fbs(self):
|
||||
endpoint = f"data/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
header = {"Accept": "application/octet-stream"}
|
||||
result = self.session.put(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'], 1838)
|
||||
self.assertIsNotNone(df['columns'])
|
||||
self.assertListEqual(df['col_idx'].tolist(), [])
|
||||
self.assertIsNone(df['row_idx'])
|
||||
self.assertEqual(len(df['columns']), df['n_cols'])
|
||||
|
||||
def test_data_put_filter_fbs(self):
|
||||
endpoint = f"data/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
header = {"Accept": "application/octet-stream"}
|
||||
@@ -384,94 +241,16 @@ class EndPoints(unittest.TestCase):
|
||||
self.assertListEqual(df['col_idx'].tolist(), [0, 1, 4])
|
||||
|
||||
def test_data_put_single_var(self):
|
||||
for axis in ["obs", "var"]:
|
||||
endpoint = f"data/{axis}"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
header = {"Accept": "application/json"}
|
||||
var_filter = {"filter": {"var": {"annotation_value": [{"name": "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/json")
|
||||
result_data = result.json()
|
||||
if axis == "obs":
|
||||
self.assertEqual(len(result_data["obs"][0]), 2)
|
||||
self.assertEqual(len(result_data["var"]), 1)
|
||||
elif axis == "var":
|
||||
self.assertEqual(len(result_data["obs"]), 2638)
|
||||
self.assertEqual(len(result_data["var"][0]), 2639)
|
||||
|
||||
def test_cache(self):
|
||||
endpoint = "annotations/var"
|
||||
endpoint = f"data/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
f1 = {
|
||||
"filter": {
|
||||
"var": {
|
||||
"annotation_value": [
|
||||
{
|
||||
"name": "name",
|
||||
"values": [
|
||||
"HLA-DRB1",
|
||||
"HLA-DQA1",
|
||||
"HLA-DQB1",
|
||||
"HLA-DPA1",
|
||||
"HLA-DPB1",
|
||||
"MS4A1",
|
||||
"IL32",
|
||||
"CCL5",
|
||||
"CD79B",
|
||||
"CD79A",
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
result = self.session.put(url, json=f1)
|
||||
header = {"Accept": "application/octet-stream"}
|
||||
var_filter = {"filter": {"var": {"annotation_value": [{"name": "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/json")
|
||||
result_data1 = result.json()
|
||||
f2 = {
|
||||
"filter": {
|
||||
"var": {
|
||||
"annotation_value": [
|
||||
{
|
||||
"name": "name",
|
||||
"values": ["FGFBP2", "GZMA", "LTB", "PRF1", "CTSW", "GZMH", "CCL5", "CCL4", "CST7", "NKG7"],
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
result = self.session.put(url, json=f2)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
result_data2 = result.json()
|
||||
self.assertNotEqual(result_data1, result_data2)
|
||||
|
||||
def test_cache_nofilter(self):
|
||||
endpoint = "annotations/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
f1 = {"filter": {}}
|
||||
result = self.session.put(url, json=f1)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
result_data1 = result.json()
|
||||
f2 = {
|
||||
"filter": {
|
||||
"var": {
|
||||
"annotation_value": [
|
||||
{
|
||||
"name": "name",
|
||||
"values": ["FGFBP2", "GZMA", "LTB", "PRF1", "CTSW", "GZMH", "CCL5", "CCL4", "CST7", "NKG7"],
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
result = self.session.put(url, json=f2)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
self.assertEqual(result.headers["Content-Type"], "application/json")
|
||||
result_data2 = result.json()
|
||||
self.assertNotEqual(result_data1, result_data2)
|
||||
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_static(self):
|
||||
endpoint = "static"
|
||||
|
||||
@@ -1,82 +0,0 @@
|
||||
import json
|
||||
from os import path
|
||||
import unittest
|
||||
|
||||
from numpy import float32, int32
|
||||
from werkzeug.datastructures import ImmutableMultiDict
|
||||
|
||||
from server.app.util.filter import _convert_variable, parse_filter, QueryStringError
|
||||
|
||||
|
||||
class UtilTest(unittest.TestCase):
|
||||
"""Test Case for endpoints"""
|
||||
|
||||
def setUp(self):
|
||||
with open(path.join(path.dirname(__file__), "schema.json")) as fh:
|
||||
schema = json.load(fh)
|
||||
self.schema = schema["annotations"]
|
||||
|
||||
def test_convert(self):
|
||||
five = _convert_variable("int32", "5")
|
||||
self.assertEqual(five, int32(5))
|
||||
|
||||
def test_convert_zero(self):
|
||||
zero = _convert_variable("int32", "0")
|
||||
self.assertEqual(zero, 0)
|
||||
|
||||
def test_convert_float(self):
|
||||
str_to_convert = "4.38719237129"
|
||||
val = _convert_variable("float32", str_to_convert)
|
||||
self.assertAlmostEqual(val, float32(str_to_convert))
|
||||
|
||||
def test_convert_bool(self):
|
||||
str_to_convert = "false"
|
||||
val = _convert_variable("boolean", str_to_convert)
|
||||
self.assertFalse(val)
|
||||
str_to_convert = "true"
|
||||
val = _convert_variable("boolean", str_to_convert)
|
||||
self.assertTrue(val)
|
||||
str_to_convert = "0"
|
||||
with self.assertRaises(AssertionError):
|
||||
val = _convert_variable("boolean", str_to_convert)
|
||||
|
||||
def test_empty_convert(self):
|
||||
empty = _convert_variable("int32", None)
|
||||
self.assertIsNone(empty)
|
||||
|
||||
def test_bad_convert(self):
|
||||
with self.assertRaises(ValueError):
|
||||
_convert_variable("int32", "5.5")
|
||||
|
||||
def test_bad_datatype(self):
|
||||
with self.assertRaises(AssertionError):
|
||||
_convert_variable("jkasdslkja", 1)
|
||||
|
||||
def test_complex_filter(self):
|
||||
filter_dict = ImmutableMultiDict(
|
||||
[("obs:louvain", "NK cells"), ("obs:louvain", "CD8 T cells"), ("obs:n_counts", "3000,*")]
|
||||
)
|
||||
filter_ = parse_filter(filter_dict, self.schema)
|
||||
self.assertIn("obs", filter_)
|
||||
self.assertEqual(
|
||||
filter_["obs"]["annotation_value"],
|
||||
[
|
||||
{"name": "louvain", "values": ["NK cells", "CD8 T cells"]},
|
||||
{"name": "n_counts", "max": None, "min": 3000.0},
|
||||
],
|
||||
)
|
||||
|
||||
def test_bad_filter(self):
|
||||
bad_annotation_type = ImmutableMultiDict([("obs:tissue", "lung")])
|
||||
with self.assertRaises(QueryStringError):
|
||||
parse_filter(bad_annotation_type, self.schema)
|
||||
bad_axis = ImmutableMultiDict([("xyz:n_genes", "100,1000")])
|
||||
with self.assertRaises(QueryStringError):
|
||||
parse_filter(bad_axis, self.schema)
|
||||
|
||||
def test_boolean_filter(self):
|
||||
schema = {"obs": [{"name": "bool_filter", "type": "boolean"}]}
|
||||
filter_dict = ImmutableMultiDict([("obs:bool_filter", "false")])
|
||||
filter_ = parse_filter(filter_dict, schema)
|
||||
self.assertIn("obs", filter_)
|
||||
self.assertEqual(filter_["obs"]["annotation_value"], [{"name": "bool_filter", "values": [False]}])
|
||||
@@ -2,6 +2,9 @@ from http import HTTPStatus
|
||||
from subprocess import Popen
|
||||
import unittest
|
||||
import time
|
||||
import math
|
||||
|
||||
import decode_fbs
|
||||
|
||||
import requests
|
||||
|
||||
@@ -43,9 +46,29 @@ class WithNaNs(unittest.TestCase):
|
||||
result = self.session.get(url)
|
||||
self.assertEqual(result.status_code, HTTPStatus.OK)
|
||||
|
||||
def test_errors(self):
|
||||
endpoints = ["annotations/obs", "annotations/var", "data/obs", "data/var"]
|
||||
for endpoint in endpoints:
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
result = self.session.get(url)
|
||||
self.assertEqual(result.status_code, HTTPStatus.INTERNAL_SERVER_ERROR)
|
||||
def test_data(self):
|
||||
endpoint = "data/var"
|
||||
url = f"{URL_BASE}{endpoint}"
|
||||
result = self.session.put(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"][3][3]))
|
||||
|
||||
def test_annotation_obs(self):
|
||||
endpoint = "annotations/obs"
|
||||
url = f"{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"{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]))
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import json
|
||||
import pytest
|
||||
import unittest
|
||||
import warnings
|
||||
@@ -7,7 +6,7 @@ import math
|
||||
import decode_fbs
|
||||
|
||||
from server.app.scanpy_engine.scanpy_engine import ScanpyEngine
|
||||
from server.app.util.errors import JSONEncodingValueError
|
||||
from server.app.util.errors import FilterError
|
||||
|
||||
|
||||
class NaNTest(unittest.TestCase):
|
||||
@@ -42,10 +41,15 @@ class NaNTest(unittest.TestCase):
|
||||
self.assertEqual(data_frame_var["n_cols"], 100)
|
||||
self.assertTrue(math.isnan(data_frame_var["columns"][3][3]))
|
||||
|
||||
with pytest.raises(JSONEncodingValueError):
|
||||
json.loads(self.data.data_frame(None, "obs"))
|
||||
with pytest.raises(JSONEncodingValueError):
|
||||
json.loads(self.data.data_frame(None, "var"))
|
||||
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:
|
||||
@@ -65,8 +69,3 @@ class NaNTest(unittest.TestCase):
|
||||
self.assertEqual(annotations["col_idx"], ["name", "n_cells", "var_with_nans"])
|
||||
self.assertEqual(annotations["n_rows"], 100)
|
||||
self.assertTrue(math.isnan(annotations["columns"][2][0]))
|
||||
|
||||
with pytest.raises(JSONEncodingValueError):
|
||||
json.loads(self.data.annotation(None, "obs"))
|
||||
with pytest.raises(JSONEncodingValueError):
|
||||
json.loads(self.data.annotation(None, "var"))
|
||||
|
||||
@@ -3,11 +3,13 @@ from os import path
|
||||
import pytest
|
||||
import time
|
||||
import unittest
|
||||
import decode_fbs
|
||||
|
||||
import numpy as np
|
||||
from pandas import Series
|
||||
|
||||
from server.app.scanpy_engine.scanpy_engine import ScanpyEngine
|
||||
from server.app.util.errors import FilterError
|
||||
|
||||
|
||||
class UtilTest(unittest.TestCase):
|
||||
@@ -45,55 +47,29 @@ class UtilTest(unittest.TestCase):
|
||||
def test_filter_idx(self):
|
||||
filter_ = {
|
||||
"filter": {
|
||||
"var": {"index": [1, 99, [200, 300]]},
|
||||
"obs": {"index": [1, 99, [1000, 2000]]},
|
||||
"var": {"index": [1, 99, [200, 300]]}
|
||||
}
|
||||
}
|
||||
data = self.data.filter_dataframe(filter_["filter"])
|
||||
self.assertEqual(data.shape, (1002, 102))
|
||||
|
||||
def test_filter_annotation(self):
|
||||
filter_ = {
|
||||
"filter": {
|
||||
"obs": {
|
||||
"annotation_value": [
|
||||
{"name": "louvain", "values": ["NK cells", "CD8 T cells"]}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
data = self.data.filter_dataframe(filter_["filter"])
|
||||
self.assertEqual(data.shape, (470, 1838))
|
||||
filter_ = {
|
||||
"filter": {"obs": {"annotation_value": [{"name": "n_counts", "min": 3000}]}}
|
||||
}
|
||||
data = self.data.filter_dataframe(filter_["filter"])
|
||||
self.assertEqual(data.shape, (497, 1838))
|
||||
|
||||
def test_filter_annotation_no_uns(self):
|
||||
filter_ = {
|
||||
"filter": {
|
||||
"var": {"annotation_value": [{"name": "name", "values": ["RER1"]}]}
|
||||
}
|
||||
}
|
||||
data = self.data.filter_dataframe(filter_["filter"])
|
||||
self.assertEqual(data.shape[1], 1)
|
||||
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": {"index": [1, 99, [200, 300]]},
|
||||
"obs": {
|
||||
"var": {
|
||||
"annotation_value": [
|
||||
{"name": "louvain", "values": ["NK cells", "CD8 T cells"]},
|
||||
{"name": "n_counts", "min": 3000},
|
||||
{"name": "n_cells", "min": 10}
|
||||
],
|
||||
"index": [1, 99, [1000, 2000]],
|
||||
},
|
||||
"index": [1, 99, [200, 300]]
|
||||
}
|
||||
}
|
||||
}
|
||||
data = self.data.filter_dataframe(filter_["filter"])
|
||||
self.assertEqual(data.shape, (15, 102))
|
||||
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["name"].isna()), 0)
|
||||
@@ -119,60 +95,42 @@ class UtilTest(unittest.TestCase):
|
||||
)
|
||||
|
||||
def test_layout(self):
|
||||
layout = json.loads(self.data.layout(None))
|
||||
self.assertEqual(layout["layout"]["ndims"], 2)
|
||||
self.assertEqual(len(layout["layout"]["coordinates"]), 2638)
|
||||
self.assertEqual(layout["layout"]["coordinates"][0][0], 0)
|
||||
for idx, val in enumerate(layout["layout"]["coordinates"]):
|
||||
self.assertLessEqual(val[1], 1)
|
||||
self.assertLessEqual(val[2], 1)
|
||||
fbs = self.data.layout_to_fbs_matrix()
|
||||
layout = decode_fbs.decode_matrix_FBS(fbs)
|
||||
self.assertEqual(layout["n_cols"], 2)
|
||||
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_annotations(self):
|
||||
annotations = json.loads(self.data.annotation(None, "obs"))
|
||||
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)
|
||||
self.assertEqual(
|
||||
annotations["names"],
|
||||
annotations["col_idx"],
|
||||
["name", "n_genes", "percent_mito", "n_counts", "louvain"],
|
||||
)
|
||||
self.assertEqual(len(annotations["data"]), 2638)
|
||||
annotations = json.loads(self.data.annotation(None, "var"))
|
||||
self.assertEqual(annotations["names"], ["name", "n_cells"])
|
||||
self.assertEqual(len(annotations["data"]), 1838)
|
||||
|
||||
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)
|
||||
self.assertEqual(annotations["col_idx"], ["name", "n_cells"])
|
||||
|
||||
def test_annotation_fields(self):
|
||||
annotations = json.loads(
|
||||
self.data.annotation(None, "obs", ["n_genes", "n_counts"])
|
||||
)
|
||||
self.assertEqual(annotations["names"], ["n_genes", "n_counts"])
|
||||
self.assertEqual(len(annotations["data"]), 2638)
|
||||
annotations = json.loads(self.data.annotation(None, "var", ["name"]))
|
||||
self.assertEqual(annotations["names"], ["name"])
|
||||
self.assertEqual(len(annotations["data"]), 1838)
|
||||
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)
|
||||
|
||||
def test_filtered_annotation(self):
|
||||
filter_ = {
|
||||
"filter": {
|
||||
"obs": {"annotation_value": [{"name": "n_counts", "min": 3000}]},
|
||||
"var": {
|
||||
"annotation_value": [{"name": "name", "values": ["ATAD3C", "RER1"]}]
|
||||
},
|
||||
}
|
||||
}
|
||||
annotations = json.loads(self.data.annotation(filter_["filter"], "obs"))
|
||||
self.assertEqual(
|
||||
annotations["names"],
|
||||
["name", "n_genes", "percent_mito", "n_counts", "louvain"],
|
||||
)
|
||||
self.assertEqual(len(annotations["data"]), 497)
|
||||
annotations = json.loads(self.data.annotation(filter_["filter"], "var"))
|
||||
self.assertEqual(annotations["names"], ["name", "n_cells"])
|
||||
self.assertEqual(len(annotations["data"]), 2)
|
||||
|
||||
def test_filtered_layout(self):
|
||||
filter_ = {
|
||||
"filter": {"obs": {"annotation_value": [{"name": "n_counts", "min": 3000}]}}
|
||||
}
|
||||
layout = json.loads(self.data.layout(filter_["filter"]))
|
||||
self.assertEqual(len(layout["layout"]["coordinates"]), 497)
|
||||
fbs = self.data.annotation_to_fbs_matrix("var", ["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]]}}}
|
||||
@@ -183,42 +141,51 @@ class UtilTest(unittest.TestCase):
|
||||
self.assertEqual(len(result), 20)
|
||||
|
||||
def test_data_frame(self):
|
||||
data_frame_obs = json.loads(self.data.data_frame(None, "obs"))
|
||||
self.assertEqual(len(data_frame_obs["var"]), 1838)
|
||||
self.assertEqual(len(data_frame_obs["obs"]), 2638)
|
||||
data_frame_var = json.loads(self.data.data_frame(None, "var"))
|
||||
self.assertEqual(len(data_frame_var["var"]), 1838)
|
||||
self.assertEqual(len(data_frame_var["obs"]), 2638)
|
||||
fbs = self.data.data_frame_to_fbs_matrix(None, "var")
|
||||
data = decode_fbs.decode_matrix_FBS(fbs)
|
||||
self.assertEqual(data["n_rows"], 2638)
|
||||
self.assertEqual(data["n_cols"], 1838)
|
||||
|
||||
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}]}}
|
||||
}
|
||||
data_frame_obs = json.loads(self.data.data_frame(filter_["filter"], "obs"))
|
||||
self.assertEqual(len(data_frame_obs["var"]), 1838)
|
||||
self.assertEqual(len(data_frame_obs["obs"]), 497)
|
||||
self.assertIsInstance(data_frame_obs["obs"][0], (list, tuple))
|
||||
self.assertEqual(type(data_frame_obs["var"][0]), int)
|
||||
data_frame_var = json.loads(self.data.data_frame(filter_["filter"], "var"))
|
||||
self.assertEqual(len(data_frame_var["var"]), 1838)
|
||||
self.assertEqual(len(data_frame_var["obs"]), 497)
|
||||
self.assertIsInstance(data_frame_var["var"][0], (list, tuple))
|
||||
self.assertEqual(type(data_frame_var["obs"][0]), int)
|
||||
with self.assertRaises(FilterError):
|
||||
self.data.data_frame_to_fbs_matrix(filter_["filter"], "var")
|
||||
|
||||
def test_data_single_gene(self):
|
||||
for axis in ["obs", "var"]:
|
||||
filter_ = {
|
||||
"filter": {
|
||||
"var": {"annotation_value": [{"name": "name", "values": ["RER1"]}]}
|
||||
}
|
||||
def test_data_named_gene(self):
|
||||
filter_ = {
|
||||
"filter": {
|
||||
"var": {"annotation_value": [{"name": "name", "values": ["RER1"]}]}
|
||||
}
|
||||
data_frame_var = json.loads(self.data.data_frame(filter_["filter"], axis))
|
||||
if axis == "obs":
|
||||
self.assertEqual(type(data_frame_var["var"][0]), int)
|
||||
self.assertIsInstance(data_frame_var["obs"][0], (list, tuple))
|
||||
elif axis == "var":
|
||||
self.assertEqual(type(data_frame_var["obs"][0]), int)
|
||||
self.assertIsInstance(data_frame_var["var"][0], (list, tuple))
|
||||
}
|
||||
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": "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())
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user