# (c) cavaliba.com - tests - search advanced import app_home.cache as cache from app_data.models import DataInstance from app_data.search import ( get_instance_from_advanced_query, parse_atom, parse_query, tokenize_query, ) from django.test import TestCase from tests.helper import add_instance, add_schema SCHEMA = 'srch_test' FIELD_DEF = { 'city': {'displayname': 'City', 'dataformat': 'string'}, 'pop': {'displayname': 'Population', 'dataformat': 'int'}, 'active': {'displayname': 'Active', 'dataformat': 'boolean'}, 'last_sync': {'displayname': 'Last Sync', 'dataformat': 'string'}, } # paris — city=Paris, pop=2000, active=True, last_sync set # lyon — city=Lyon, pop=500, active=True, last_sync absent # toulouse — city=Toulouse, pop=500, active=False, last_sync absent # bordeaux — city=Bordeaux, pop=800, active=True, last_sync set class SearchTokenizeTest(TestCase): def test_empty_string(self): self.assertEqual(tokenize_query(""), []) def test_none(self): self.assertEqual(tokenize_query(None), []) def test_simple_words(self): self.assertEqual(tokenize_query("AND OR NOT"), ["AND", "OR", "NOT"]) def test_field_colon_value(self): self.assertEqual(tokenize_query("city:Paris"), ["city:Paris"]) def test_double_quoted_value(self): result = tokenize_query('city:"Le Mans"') self.assertEqual(result, ["city:Le Mans"]) def test_single_quoted_value(self): result = tokenize_query("city:'Le Mans'") self.assertEqual(result, ["city:Le Mans"]) def test_complex_query(self): result = tokenize_query("NOT is_enabled:True AND pop:>=500") self.assertEqual(result, ["NOT", "is_enabled:True", "AND", "pop:>=500"]) class SearchParseAtomTest(TestCase): def test_fulltext(self): t = parse_atom("Paris") self.assertEqual(t["type"], "fulltext") self.assertEqual(t["value"], "Paris") def test_field_icontains(self): t = parse_atom("city:Paris") self.assertEqual(t["type"], "field") self.assertEqual(t["field"], "city") self.assertEqual(t["compare_op"], "icontains") self.assertEqual(t["value"], "Paris") def test_field_startswith(self): t = parse_atom("city:Par*") self.assertEqual(t["compare_op"], "istartswith") self.assertEqual(t["value"], "Par") def test_field_endswith(self): t = parse_atom("city:*ris") self.assertEqual(t["compare_op"], "iendswith") self.assertEqual(t["value"], "ris") def test_field_multiglob(self): t = parse_atom("city:*ari*") self.assertEqual(t["compare_op"], "icontains") self.assertEqual(t["value"], "ari") def test_field_gte(self): t = parse_atom("pop:>=500") self.assertEqual(t["type"], "field") self.assertEqual(t["compare_op"], "gte") self.assertEqual(t["value"], "500") def test_field_gt(self): t = parse_atom("pop:>500") self.assertEqual(t["compare_op"], "gt") def test_field_lte(self): t = parse_atom("pop:<=500") self.assertEqual(t["compare_op"], "lte") def test_field_lt(self): t = parse_atom("pop:<500") self.assertEqual(t["compare_op"], "lt") def test_exists(self): t = parse_atom("EXISTS:last_sync") self.assertEqual(t["type"], "exists") self.assertEqual(t["field"], "last_sync") def test_exists_lowercase(self): t = parse_atom("exists:last_sync") self.assertEqual(t["type"], "exists") def test_empty(self): t = parse_atom("EMPTY:last_sync") self.assertEqual(t["type"], "empty") self.assertEqual(t["field"], "last_sync") def test_related(self): t = parse_atom("RELATED:SERVER:srv01") self.assertEqual(t["type"], "related") self.assertEqual(t["ref_classname"], "SERVER") # preserved as typed self.assertEqual(t["ref_keyname"], "srv01") def test_related_lowercase_compat(self): t = parse_atom("related:SERVER:srv01") self.assertEqual(t["type"], "related") def test_meta_is_enabled(self): t = parse_atom("is_enabled:True") self.assertEqual(t["type"], "field") self.assertTrue(t["is_meta"]) self.assertEqual(t["field"], "is_enabled") def test_meta_keyname(self): t = parse_atom("keyname:paris") self.assertTrue(t["is_meta"]) def test_meta_displayname(self): t = parse_atom("displayname:Par*") self.assertTrue(t["is_meta"]) self.assertEqual(t["compare_op"], "istartswith") class SearchParseQueryTest(TestCase): def test_single_term(self): result = parse_query(["Paris"]) self.assertEqual(len(result), 1) connector, negate, term = result[0] self.assertEqual(connector, "AND") self.assertFalse(negate) self.assertEqual(term["type"], "fulltext") def test_or_between_terms(self): result = parse_query(["Lyon", "OR", "Toulouse"]) self.assertEqual(len(result), 2) self.assertEqual(result[0][0], "AND") self.assertEqual(result[1][0], "OR") def test_and_between_terms(self): result = parse_query(["city:Paris", "AND", "pop:>=500"]) self.assertEqual(len(result), 2) self.assertEqual(result[1][0], "AND") def test_implicit_and(self): result = parse_query(["city:Paris", "pop:>=500"]) self.assertEqual(len(result), 2) self.assertEqual(result[1][0], "AND") def test_not_modifier(self): result = parse_query(["NOT", "city:Paris"]) self.assertEqual(len(result), 1) connector, negate, term = result[0] self.assertTrue(negate) def test_not_resets_after_term(self): result = parse_query(["NOT", "city:Paris", "pop:>=500"]) self.assertEqual(len(result), 2) self.assertTrue(result[0][1]) # first term negated self.assertFalse(result[1][1]) # second term not negated def test_empty_tokens(self): result = parse_query([]) self.assertEqual(result, []) def test_only_operator(self): result = parse_query(["AND"]) self.assertEqual(result, []) class SearchQueryTest(TestCase): @classmethod def setUpClass(cls): super().setUpClass() add_schema(SCHEMA, field_definition=FIELD_DEF) add_instance(SCHEMA, 'paris', fields={'city': 'Paris', 'pop': 2000, 'active': True, 'last_sync': '2026-01-01'}) add_instance(SCHEMA, 'lyon', fields={'city': 'Lyon', 'pop': 500, 'active': True}) add_instance(SCHEMA, 'toulouse', fields={'city': 'Toulouse', 'pop': 500, 'active': False}) add_instance(SCHEMA, 'bordeaux', fields={'city': 'Bordeaux', 'pop': 800, 'active': True, 'last_sync': '2026-03-15'}) def setUp(self): cache.clear() def _keynames(self, qs): return set(qs.values_list('keyname', flat=True)) def _q(self, query): return get_instance_from_advanced_query(query=query, classname=SCHEMA) def test_empty_query_returns_all(self): qs = self._q("") self.assertEqual(qs.count(), 4) def test_none_query_returns_all(self): qs = get_instance_from_advanced_query(query=None, classname=SCHEMA) self.assertEqual(qs.count(), 4) def test_fulltext_single_word(self): qs = self._q("Paris") self.assertIn('paris', self._keynames(qs)) self.assertNotIn('lyon', self._keynames(qs)) def test_fulltext_or(self): qs = self._q("Lyon OR Toulouse") self.assertEqual(self._keynames(qs), {'lyon', 'toulouse'}) def test_fulltext_case_insensitive(self): qs = self._q("paris") self.assertIn('paris', self._keynames(qs)) def test_field_icontains(self): qs = self._q("city:Paris") self.assertEqual(self._keynames(qs), {'paris'}) def test_field_icontains_case_insensitive(self): qs = self._q("city:paris") self.assertEqual(self._keynames(qs), {'paris'}) def test_field_startswith(self): qs = self._q("city:Toul*") self.assertEqual(self._keynames(qs), {'toulouse'}) def test_field_endswith(self): qs = self._q("city:*ouse") self.assertEqual(self._keynames(qs), {'toulouse'}) def test_field_multiglob(self): qs = self._q("city:*ari*") self.assertEqual(self._keynames(qs), {'paris'}) def test_field_and(self): qs = self._q("city:Paris AND pop:2000") self.assertEqual(self._keynames(qs), {'paris'}) def test_field_and_no_match(self): qs = self._q("city:Paris AND pop:500") self.assertEqual(qs.count(), 0) def test_field_or(self): qs = self._q("city:Paris OR city:Lyon") self.assertEqual(self._keynames(qs), {'paris', 'lyon'}) def test_not_field(self): qs = self._q("NOT city:Paris") self.assertNotIn('paris', self._keynames(qs)) self.assertIn('lyon', self._keynames(qs)) self.assertEqual(qs.count(), 3) def test_comparison_gte_all(self): # all instances have pop >= 500 qs = self._q("pop:>=500") self.assertEqual(qs.count(), 4) def test_comparison_gt(self): # only bordeaux (800) and paris (2000) have pop > 500 qs = self._q("pop:>500") self.assertEqual(self._keynames(qs), {'bordeaux', 'paris'}) def test_comparison_lt(self): # lyon and toulouse have pop < 800 qs = self._q("pop:<800") self.assertEqual(self._keynames(qs), {'lyon', 'toulouse'}) def test_comparison_lte(self): qs = self._q("pop:<=500") self.assertEqual(self._keynames(qs), {'lyon', 'toulouse'}) def test_comparison_and(self): qs = self._q("pop:>=500 AND active:True") self.assertEqual(self._keynames(qs), {'paris', 'lyon', 'bordeaux'}) def test_exists_field_present(self): # paris and bordeaux have last_sync set qs = self._q("EXISTS:last_sync") self.assertEqual(self._keynames(qs), {'paris', 'bordeaux'}) def test_not_exists_field(self): qs = self._q("NOT EXISTS:last_sync") self.assertEqual(self._keynames(qs), {'lyon', 'toulouse'}) def test_empty_field(self): # lyon and toulouse have no last_sync EAV row qs = self._q("EMPTY:last_sync") self.assertEqual(self._keynames(qs), {'lyon', 'toulouse'}) def test_not_empty_field(self): # paris and bordeaux have a non-empty last_sync value qs = self._q("NOT EMPTY:last_sync") self.assertEqual(self._keynames(qs), {'paris', 'bordeaux'}) def test_meta_is_enabled_true(self): qs = self._q("is_enabled:True") self.assertEqual(qs.count(), 4) def test_meta_is_enabled_false(self): DataInstance.objects.filter(classname=SCHEMA, keyname='toulouse').update(is_enabled=False) qs = self._q("is_enabled:False") self.assertEqual(self._keynames(qs), {'toulouse'}) def test_meta_keyname(self): qs = self._q("keyname:paris") self.assertEqual(self._keynames(qs), {'paris'}) def test_meta_keyname_startswith(self): qs = self._q("keyname:bord*") self.assertEqual(self._keynames(qs), {'bordeaux'}) def test_not_and_combination(self): # NOT active:False (=> active != False => active=True) AND pop:>=500 qs = self._q("NOT active:False AND pop:>=500") self.assertEqual(self._keynames(qs), {'paris', 'lyon', 'bordeaux'}) def test_no_classname(self): qs = get_instance_from_advanced_query(query="Paris") self.assertIn('paris', self._keynames(qs)) def test_pagination_limit(self): qs = get_instance_from_advanced_query(query="", classname=SCHEMA, offset=0, limit=2) self.assertEqual(len(list(qs)), 2) def test_pagination_page(self): qs = get_instance_from_advanced_query(query="", classname=SCHEMA, page=2, size=2) self.assertEqual(len(list(qs)), 2)