fix: scorer case preservation, duplicate score guard, OR ~ prefix, stopword warning

- Preserve caller-provided scorer casing (score('MyScorer') no longer
  uppercased to MYSCORER)
- Raise ValueError when multiple score() expressions in same query
- Preserve ~ optional-term prefix in OR operands
- Fix misleading stopword warning when all tokens are stopwords
- Remove redundant inner pytest import in test_translator.py
- Add 3 new tests (373 total)

Author: nkanu17

sql_redis/parser.py

@@ -462,7 +462,11 @@ def _process_select_expression_inner(
                 if expression.expressions:
                     scorer_val = self._extract_literal_value(expression.expressions[0])
                     if scorer_val is not None:
-                        scorer = str(scorer_val).upper()
+                        scorer = str(scorer_val)
+                if result.scoring is not None:
+                    raise ValueError(
+                        "Only one score() expression is allowed per query."
+                    )
                 result.scoring = ScoringSpec(
                     alias=alias or "score",
                     scorer=scorer,

sql_redis/query_builder.py

@@ -154,10 +154,21 @@ def build_text_condition(
                         "must contain at least one search term."
                     )
                 if len(words) > 1:
-                    escaped = " ".join(self._escape_fulltext_term(w) for w in words)
-                    or_parts.append(f"({escaped})")
+                    escaped_tokens = []
+                    for w in words:
+                        if w.startswith("~"):
+                            escaped_tokens.append(
+                                "~" + self._escape_fulltext_term(w[1:])
+                            )
+                        else:
+                            escaped_tokens.append(self._escape_fulltext_term(w))
+                    or_parts.append(f"({' '.join(escaped_tokens)})")
                 else:
-                    or_parts.append(self._escape_fulltext_term(words[0]))
+                    token = words[0]
+                    if token.startswith("~"):
+                        or_parts.append("~" + self._escape_fulltext_term(token[1:]))
+                    else:
+                        or_parts.append(self._escape_fulltext_term(token))
             search_value = f"({'|'.join(or_parts)})"
         elif " " in value:
             # FULLTEXT/MATCH with multi-word: tokenized search with stopword filtering.
@@ -173,8 +184,12 @@ def build_text_condition(
             ]
 
             if removed_stopwords:
+                if filtered_words:
+                    sw_action = f"Stopwords {removed_stopwords} were removed from"
+                else:
+                    sw_action = f"All tokens in '{value}' are stopwords and may not be indexed by"
                 warnings.warn(
-                    f"Stopwords {removed_stopwords} were removed from text search '{value}'. "
+                    f"{sw_action} text search '{value}'. "
                     "By default, Redis does not index stopwords. "
                     "To include stopwords in your index, create it with STOPWORDS 0. "
                     "Use = operator for exact phrase matching that preserves stopwords.",

tests/test_query_builder.py

@@ -690,3 +690,9 @@ def test_or_leading_empty_operand_raises(self):
         builder = QueryBuilder()
         with pytest.raises(ValueError, match="Empty operand in OR expression"):
             builder.build_text_condition("title", "FULLTEXT", " OR tablet")
+
+    def test_or_preserves_optional_prefix(self):
+        """OR operand with ~ prefix preserves optional-term semantics."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("title", "FULLTEXT", "laptop OR ~gaming")
+        assert result == "@title:(laptop|~gaming)"

tests/test_translator.py

@@ -785,6 +785,25 @@ def test_score_custom_scorer(self, translator: Translator, basic_index: str):
         scorer_idx = result.args.index("SCORER")
         assert result.args[scorer_idx + 1] == "TFIDF"
 
+    def test_score_custom_scorer_preserves_case(
+        self, translator: Translator, basic_index: str
+    ):
+        """score('MyScorer') preserves caller-provided casing."""
+        result = translator.translate(
+            f"SELECT title, score('MyScorer') AS relevance FROM {basic_index} "
+            "WHERE fulltext(title, 'laptop')"
+        )
+        scorer_idx = result.args.index("SCORER")
+        assert result.args[scorer_idx + 1] == "MyScorer"
+
+    def test_duplicate_score_raises(self, translator: Translator, basic_index: str):
+        """Multiple score() expressions in the same query raise ValueError."""
+        with pytest.raises(ValueError, match="Only one score"):
+            translator.translate(
+                f"SELECT score() AS s1, score('TFIDF') AS s2 FROM {basic_index} "
+                "WHERE fulltext(title, 'laptop')"
+            )
+
     def test_no_score_no_withscores(self, translator: Translator, basic_index: str):
         """Without score() → no WITHSCORES flag."""
         result = translator.translate(
@@ -808,8 +827,6 @@ def test_score_with_aggregate_raises(
         self, translator: Translator, basic_index: str
     ):
         """score() combined with GROUP BY (forces FT.AGGREGATE) raises ValueError."""
-        import pytest
-
         with pytest.raises(ValueError, match="score.*not supported.*FT.AGGREGATE"):
             translator.translate(
                 f"SELECT COUNT(*), score() AS relevance FROM {basic_index} "