from atlas_um.helpers.filtering import QueryFiltering from sqlalchemy.orm import Query from sqlalchemy.schema import Column class TestQueryFiltering: def test_filter_all_columns(self, app, faker): columns = [ Column(faker.pystr()) for _ in range(faker.pyint(min_value=1, max_value=10)) ] test_value = faker.pystr() filter_string = ";".join(f"{c.name}:{test_value}" for c in columns) expected_query = Query(columns).filter( *[c.in_([test_value]) for c in columns] ) with app.test_request_context(): actual_query = QueryFiltering( Query(columns), columns, filter_string ).query assert str(actual_query) == str(expected_query) def test_filter_multiple_values(self, app, faker): columns = [ Column(faker.pystr()) for _ in range(faker.pyint(min_value=1, max_value=10)) ] test_value = faker.pystr() test_value2 = faker.pystr() filter_string = ";".join( f"{c.name}:{test_value},{test_value2}" for c in columns ) expected_query = Query(columns).filter( *[c.in_([test_value, test_value2]) for c in columns] ) with app.test_request_context(): actual_query = QueryFiltering( Query(columns), columns, filter_string ).query assert str(actual_query) == str(expected_query) def test_filter_mapped_columns(self, app, faker): columns = [ Column(faker.pystr()) for _ in range(faker.pyint(min_value=1, max_value=10)) ] test_value = faker.pystr() filter_string = ";".join(f"m_{c.name}:{test_value}" for c in columns) fields_mapping = {f"m_{c.name}": c.name for c in columns} expected_query = Query(columns).filter( *[c.in_([test_value]) for c in columns] ) with app.test_request_context(): actual_query = QueryFiltering( Query(columns), columns, filter_string, fields_mapping ).query assert str(actual_query) == str(expected_query) def test_filter_allowed_columns(self, app, faker): columns = [ Column(faker.pystr()) for _ in range(faker.pyint(min_value=1, max_value=10)) ] test_value = faker.pystr() filter_string = ";".join(f"{c.name}:{test_value}" for c in columns) expected_query = Query(columns).filter( *[c.in_([test_value]) for c in columns[:1]] ) with app.test_request_context(): actual_query = QueryFiltering( Query(columns), columns[:1], filter_string ).query assert str(actual_query) == str(expected_query) def test_filter_wrong_columns(self, app, faker): columns = [ Column(faker.pystr()) for _ in range(faker.pyint(min_value=1, max_value=10)) ] test_value = faker.pystr() filter_string = ";".join(f"{c.name}typo:{test_value}" for c in columns) expected_query = Query(columns).filter() with app.test_request_context(): actual_query = QueryFiltering( Query(columns), columns, filter_string ).query assert str(actual_query) == str(expected_query)