Skip to content

Commit

Permalink
add test units to import data from csv files
Browse files Browse the repository at this point in the history
  • Loading branch information
@khaledk2
khaledk2 committed Dec 27, 2024
1 parent 524dc3c commit 9149f0a
Show file tree
Hide file tree
Showing 2 changed files with 28 additions and 301 deletions.
7 changes: 7 additions & 0 deletions .github/workflows/main.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,13 @@ jobs:
python manage.py restore_postgresql_database -s omero_train
python manage.py get_index_data_from_database -b False -s omero_train
rm app_data/omero_train.pgdump
# set csv data source
python manage.py set_data_source_files -n test_csv
# download images csv file
wget https://github.com/khaledk2/ice-archh-64/releases/download/test_csv/test_images.zip -P app_data
unzip app_data/test_images.zip -d app_data/
rm app_data/test_images.zip
python manage.py get_index_data_from_csv_files -s test_csv -f app_data/test_images.csv -r image
# run tests
python -m unittest discover -s unit_tests
upload:
Expand Down
322 changes: 21 additions & 301 deletions unit_tests/test_app.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,11 @@
from omero_search_engine.api.v1.resources.utils import (
elasticsearch_query_builder,
search_resource_annotation,
get_data_sources,
)

from omero_search_engine.api.v1.resources.query_handler import (
simple_search,
)

from omero_search_engine.cache_functions.elasticsearch.elasticsearch_templates import ( # noqa
Expand Down Expand Up @@ -79,316 +84,31 @@ def setUp(self):
def tearDown(self):
pass

def test_api_v1(self):
"""test url"""
tester = search_omero_app.test_client(self)
response = tester.get("/api/v1/resources/", content_type="html/text")
self.assertEqual(response.status_code, 200)

def test_searchannotation(self):
"""test url"""
tester = search_omero_app.test_client(self)
query = {"query_details": {}}

response = tester.post(
"/api/v1/resources/image/searchannotation/", data=query
) # noqa
self.assertEqual(response.status_code, 200)
Error = response.json["Error"]
self.assertIsInstance(Error, str)

def test_not_found(self):
"""
test not found url
"""
tester = search_omero_app.test_client(self)
response = tester.get("a", content_type="html/text")
self.assertEqual(response.status_code, 404)

def test_query_database(self):
def test_data_sources(self):
"""
test connection with postgresql database
Test available data sources
"""
for data_source in search_omero_app.config.database_connectors.keys():
res = search_omero_app.config.database_connectors[
data_source
].execute_query(sql)
self.assertIsNotNone(res)
found_db_name = False
for source in search_omero_app.config.get("DATA_SOURCES"):
if (
source.get("DATABASE").get("DATABASE_NAME")
== res[0]["current_database"]
):
found_db_name = True
break

self.assertTrue(found_db_name)
# self.assertEqual(res[0]["current_database"],
# search_omero_app.config.database_connectors[data_source]["DATABASE_NAME"])
# self.assertEqual(res[0]["current_database"],
# search_omero_app.config.database_connectors[data_source]["DATABASE_NAME"])

def validate_json_syntax(self, json_template):
try:
return json.loads(json_template)
except ValueError:
print("DEBUG: JSON data contains an error")
return False

def validate_json_syntax_for_es_templates(self):
self.assertTrue(self.validate_json_syntax(image_template))
self.assertTrue(self.validate_json_syntax(image_template))
data_sources =get_data_sources()
self.assertTrue ("test_csv" in data_sources)

def test_is_valid_json_for_query(self):
def test_csv_data_query(self):
"""
test output of query builderis valid json
Test available data sources
"""
query = elasticsearch_query_builder(valid_and_filters, valid_or_filters, False)
self.assertTrue(self.validate_json_syntax(query))

def test_is_not_valid_json_query(self):
"""
test output of query builderis valid json
"""
no_valid_message = elasticsearch_query_builder(
not_valid_and_filters, not_valid_or_filters, False
results = simple_search(
"organism",
"homo sapiens",
"equals",
None,
None,
"image",
None,
"test_csv"
)
self.assertTrue("Error" in no_valid_message.keys())

def test_add_submit_query_delete_es_index(self):
"""'
test submit query and get results
"""
# table = "image1"
table_2 = "image"
es_index = "image_keyvalue_pair_metadata_1"
es_index_2 = "key_values_resource_cach"
create_es_index_2 = True
all_all_indices = get_all_indexes_from_elasticsearch()
if es_index_2 in all_all_indices.keys():
create_es_index_2 = False

if es_index not in all_all_indices.keys():
self.assertTrue(create_index(es_index, image_template))
if create_es_index_2:
self.assertTrue(
create_index(es_index_2, key_values_resource_cache_template)
)
res = search_resource_annotation(table_2, query)
assert len(res.get("results")) >= 0
self.assertTrue(delete_es_index(es_index))
if create_es_index_2:
self.assertTrue(delete_es_index(es_index_2))

def test_single_query(self):
"""
test query the search engine and compare
its results with the results from the database
"""
for data_source in search_omero_app.config.database_connectors.keys():
for resource, cases in simple_queries.items():
for case in cases:
name = case[0]
value = case[1]
validator = Validator(data_source, deep_check)
validator.set_simple_query(resource, name, value)
validator.get_results_db("equals")
validator.get_results_searchengine("equals")
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
validator.get_results_db("not_equals")
validator.get_results_searchengine("not_equals")
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
self.assertTrue(validator.identical)
self.assertEqual( results.get("results").get("size") , 15756)

def test_and_query(self):
name = "query_image_and"
for data_source in search_omero_app.config.database_connectors.keys():
for cases in query_image_and:
validator = Validator(data_source, deep_check)
validator.set_complex_query(name, cases)
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
self.assertTrue(validator.identical)

def test_or_query(self):
for data_source in search_omero_app.config.database_connectors.keys():
name = "query_image_or"
for cases in query_image_or:
validator = Validator(data_source, deep_check)
validator.set_complex_query(name, cases)
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
self.assertTrue(validator.identical)

# def test_no_images_containers(self):
# for data_source in search_omero_app.config.database_connectors.keys():
# self.assertTrue(check_number_images_sql_containers_using_ids(data_source))

def test_multi_or_quries(self):
pass

def test_complex_query(self):
for data_source in search_omero_app.config.database_connectors.keys():
name = "query_image_and_or"
for cases in query_image_and_or:
validator = Validator(data_source, deep_check)
validator.set_complex_query(name, cases)
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
self.assertTrue(validator.identical)

def test_in_query(self):
for data_source in search_omero_app.config.database_connectors.keys():
for resource, cases in query_in.items():
for case in cases:
validator = Validator(data_source, deep_check)
validator.set_in_query(case, resource)
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
self.assertTrue(validator.identical)

def test_not_in_query(self):
for data_source in search_omero_app.config.database_connectors.keys():
for resource, cases in query_in.items():
for case in cases:
validator = Validator(data_source, deep_check)
validator.set_in_query(case, resource, type="not_in_clause")
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
self.assertTrue(validator.identical)

def test_seach_for_any_value(self):
for data_source in search_omero_app.config.database_connectors.keys():
for part in images_value_parts:
validator = Validator(data_source, deep_check)
validator.set_simple_query("image", None, part, type="buckets")
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("total_number_of_buckets"),
)

def test_available_values_for_key(self):
for data_source in search_omero_app.config.database_connectors.keys():
for image_key in images_keys:
validator = Validator(data_source, deep_check)
validator.set_simple_query("image", image_key, None, type="buckets")
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("total_number_of_buckets"),
)

def test_contains_not_contains_queries(self):
for data_source in search_omero_app.config.database_connectors.keys():
for resource, cases in contains_not_contains_queries.items():
for case in cases:
name = case[0]
value = case[1]
validator = Validator(data_source, deep_check)
validator.set_contains_not_contains_query(resource, name, value)
validator.get_results_db("contains")
validator.get_results_searchengine("contains")
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
validator.get_results_db("not_contains")
validator.get_results_searchengine("not_contains")
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)
self.assertTrue(validator.identical)

def test_owner(self):
for data_source in search_omero_app.config.database_connectors.keys():
for resource, cases in image_owner.items():
for case in cases:
name = case[0]
value = case[1]
owner_id = case[2]
validator = Validator(data_source, deep_check)
validator.set_simple_query(resource, name, value)
validator.set_owner_group(owner_id=owner_id)
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)

def test_group(self):
for data_source in search_omero_app.config.database_connectors.keys():
for resource, cases in image_group.items():
for case in cases:
name = case[0]
value = case[1]
group_id = case[2]
validator = Validator(data_source, deep_check)
validator.set_simple_query(resource, name, value)
validator.set_owner_group(group_id=group_id)
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)

def test_owner_group(self):
for data_source in search_omero_app.config.database_connectors.keys():
for resource, cases in image_owner_group.items():
for case in cases:
name = case[0]
value = case[1]
owner_id = case[2]
group_id = case[3]
validator = Validator(data_source, deep_check)
validator.set_simple_query(resource, name, value)
validator.set_owner_group(owner_id=owner_id, group_id=group_id)
validator.compare_results()
self.assertEqual(
len(validator.postgres_results),
validator.searchengine_results.get("size"),
)

# def test_add_delete_es_index(self):
# '''
# test create index in elastic search
# :return:
# '''
# from datetime import datetime
# es_index_name="test_image_%s"%str(datetime.now().second)

# self.assertTrue (create_index(es_index_name, image_template))
# self.assertTrue (delete_es_index(es_index_name))

def test_log_in_log_out(self):
"""
test login and log out functions
:return:
"""
pass


if __name__ == "__main__":
Expand Down

0 comments on commit 9149f0a

Please sign in to comment.