mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-09-15 20:57:56 +08:00
f3015cb9df
* Fix Makefile whitespace and .PHONY use
* Fix Makefile filename
* Modularize Makefile into client and server Makefiles
Part of the reason that the Makefile in the root directory is a bit
complicated is that it tries to handle tasks that can be handled
separately in the client and server modules.
This commit pushes some of the make logic specific to each module into
their own makefiles and calls out to those makefiles from that in the
project root.
* Add auto-formatting to client and server modules
One thing that can make linting faster is auto-formatting. This commit
adds the yapf auto-formatting tool to the server module and uses
eslint's "fix" functionality to speed up the linting/formatting process.
* Add yapf for automatic code formatting
* Add a root test target that calls sub-tests
* Apply yapf to python files
* Do not duplicate npm commands, simply pass through
* Update documentation
* Do not shadow reserved word len
* Add general test target
* Fix make call in dev-env
* Use black instead of yapf
* Run flake8 from the root directory
* Revert "Apply yapf to python files"
This reverts commit cdca128a01.
* Apply black to python code
* Resolve lint errors resulting from black format
* Add explanation of server unit tests in dev guidelines
195 lines
8.1 KiB
Python
195 lines
8.1 KiB
Python
import json
|
|
from os import path
|
|
import pytest
|
|
import time
|
|
import unittest
|
|
import decode_fbs
|
|
from parameterized import parameterized_class
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from server.app.scanpy_engine.scanpy_engine import ScanpyEngine
|
|
from server.app.util.errors import FilterError, DisabledFeatureError
|
|
from server.app.util.data_locator import DataLocator
|
|
|
|
"""
|
|
Test the scanpy engine using the pbmc3k data set.
|
|
"""
|
|
|
|
|
|
@parameterized_class(
|
|
("data_locator", "backed"),
|
|
[
|
|
("../example-dataset/pbmc3k.h5ad", False),
|
|
("test/test_datasets/pbmc3k-CSC-gz.h5ad", False),
|
|
("test/test_datasets/pbmc3k-CSR-gz.h5ad", False),
|
|
("../example-dataset/pbmc3k.h5ad", True),
|
|
("test/test_datasets/pbmc3k-CSC-gz.h5ad", True),
|
|
("test/test_datasets/pbmc3k-CSR-gz.h5ad", True),
|
|
],
|
|
)
|
|
class EngineTest(unittest.TestCase):
|
|
def setUp(self):
|
|
args = {
|
|
"layout": ["umap"],
|
|
"max_category_items": 100,
|
|
"obs_names": None,
|
|
"var_names": None,
|
|
"diffexp_lfc_cutoff": 0.01,
|
|
"layout_file": None,
|
|
"backed": self.backed,
|
|
}
|
|
self.data = ScanpyEngine(DataLocator(self.data_locator), args)
|
|
|
|
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:Scanpy 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_schema(self):
|
|
with open(path.join(path.dirname(__file__), "schema.json")) as fh:
|
|
schema = json.load(fh)
|
|
self.assertEqual(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_config(self):
|
|
self.assertEqual(
|
|
self.data.features["layout"]["obs"], {"available": True, "interactiveLimit": 50000},
|
|
)
|
|
|
|
def test_layout(self):
|
|
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):
|
|
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_annotation_put(self):
|
|
with self.assertRaises(DisabledFeatureError):
|
|
self.data.annotation_put_fbs(None, "obs")
|
|
|
|
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())
|