From 1a1c54fe5a7b7ce9e940427f7d4524ba4f09cff8 Mon Sep 17 00:00:00 2001 From: juanpabloguerra16 Date: Mon, 1 Jun 2026 12:19:59 -0700 Subject: [PATCH 1/7] Improve CLI search and fetch signaling --- src/ctxd/cli.py | 67 +++++++++++++++++++++++++++-- tests/test_cli.py | 105 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+), 3 deletions(-) diff --git a/src/ctxd/cli.py b/src/ctxd/cli.py index 43b52a2..37be103 100644 --- a/src/ctxd/cli.py +++ b/src/ctxd/cli.py @@ -5,6 +5,7 @@ import json import os import re +import shlex import sys import webbrowser from typing import Sequence @@ -31,7 +32,9 @@ def main(argv: Sequence[str] | None = None) -> int: client = Client() if args.command == "search": - result = client.search(_normalize_search_query(args.query)) + query = _normalize_search_query(args.query) + _validate_search_query(query) + result = client.search(query) return _emit_result(result.model_dump(), as_json=True) if args.command == "fetch": result = client.fetch_document(args.document_uid) @@ -117,7 +120,18 @@ def _build_parser() -> argparse.ArgumentParser: epilog=( "Examples:\n" " ctxd search text:deployment application:slack\n" - ' ctxd search "text:deployment application:slack"' + ' ctxd search "text:deployment application:slack"\n' + ' ctxd search \'text:"incident response" application:google_drive\'\n' + " ctxd search 'text:(incident response) application:google_drive'\n" + " ctxd search 'text:incident AND text:response application:google_drive'\n" + " ctxd search 'application:google_drive OR application:slack text:onboarding'\n" + "\n" + "DSL notes:\n" + ' text:"a b" runs a semantic multi-word text search.\n' + " text:(a b) matches any listed term, equivalent to text:a OR text:b.\n" + " text:a AND text:b requires both terms.\n" + " Use application:x OR application:y for multi-app unions; grouped or repeated\n" + " application filters are rejected because they can otherwise look successful." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -180,6 +194,45 @@ def _quote_shell_stripped_text_token(token: str) -> str: return f'text:"{escaped_value}"' +def _validate_search_query(query: str) -> None: + tokens = _split_search_query(query) + _validate_application_filters(tokens) + + +def _split_search_query(query: str) -> list[str]: + try: + return shlex.split(query) + except ValueError: + return query.split() + + +def _validate_application_filters(tokens: Sequence[str]) -> None: + application_positions: list[int] = [] + for index, token in enumerate(tokens): + if not token.lower().startswith("application:"): + continue + + value = token[len("application:") :] + if not value: + raise ValueError( + "Invalid search query: application: requires an app name, for example application:slack." + ) + if value.startswith("("): + raise ValueError( + "Invalid search query: grouped application filters are not supported. " + "Use application:google_drive OR application:slack." + ) + application_positions.append(index) + + for left, right in zip(application_positions, application_positions[1:]): + between = [token.upper() for token in tokens[left + 1 : right]] + if "OR" not in between: + raise ValueError( + "Invalid search query: repeated application filters must be joined with OR, " + "for example application:google_drive OR application:slack." + ) + + def _handle_login(args: argparse.Namespace) -> int: del args api_key, should_save = _resolve_login_api_key() @@ -311,6 +364,14 @@ def _emit_result(payload: dict, *, as_json: bool) -> int: def _payload_has_error(payload: dict) -> bool: + if "results" in payload: + return bool(payload.get("error") or payload.get("dsl_parse_error")) if payload.get("error"): return True - return bool(payload.get("dsl_parse_error")) + if payload.get("dsl_parse_error"): + return True + return "error" in payload and _document_payload_is_unresolved(payload) + + +def _document_payload_is_unresolved(payload: dict) -> bool: + return not any(payload.get(key) for key in ("id", "title", "text", "url")) diff --git a/tests/test_cli.py b/tests/test_cli.py index 4476784..1af87f6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -50,6 +50,9 @@ def test_cli_search_help_describes_query_and_json_output() -> None: assert "Search output is always JSON." in output assert "QUERY" in output assert "text:deployment application:slack" in output + assert 'text:"a b" runs a semantic multi-word text search.' in output + assert "text:(a b) matches any listed term" in output + assert "application:x OR application:y" in output def test_cli_profile_json_calls_sdk() -> None: @@ -339,6 +342,32 @@ def test_cli_fetch_returns_document() -> None: assert "Deploy completed successfully." in output +def test_cli_fetch_returns_nonzero_exit_code_for_unresolved_document() -> None: + stdout = StringIO() + document = DocumentResult(error="Document UID could not be resolved.") + + with patch( + "ctxd.cli.Client.fetch_document", return_value=document + ), redirect_stdout(stdout): + exit_code = main(["fetch", "missing-doc"]) + + assert exit_code == 1 + assert stdout.getvalue() == "Error: Document UID could not be resolved.\n" + + +def test_cli_fetch_json_returns_nonzero_exit_code_for_unresolved_document() -> None: + stdout = StringIO() + document = DocumentResult(error="Document UID could not be resolved.") + + with patch( + "ctxd.cli.Client.fetch_document", return_value=document + ), redirect_stdout(stdout): + exit_code = main(["fetch", "missing-doc", "--json"]) + + assert exit_code == 1 + assert '"error": "Document UID could not be resolved."' in stdout.getvalue() + + def test_cli_search_returns_nonzero_exit_code_for_payload_errors() -> None: stdout = StringIO() @@ -356,6 +385,82 @@ def test_cli_search_returns_nonzero_exit_code_for_payload_errors() -> None: assert '"error": "bad query"' in stdout.getvalue() +def test_cli_search_rejects_empty_application_filter() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main(["search", "application:"]) + + assert exit_code == 1 + search.assert_not_called() + assert "application: requires an app name" in stderr.getvalue() + + +def test_cli_search_rejects_grouped_application_filter() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main( + ["search", "application:(google_drive OR slack)", "text:onboarding"] + ) + + assert exit_code == 1 + search.assert_not_called() + assert "grouped application filters are not supported" in stderr.getvalue() + + +def test_cli_search_rejects_repeated_application_filters_without_or() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main( + [ + "search", + "application:google_drive", + "application:slack", + "text:onboarding", + ] + ) + + assert exit_code == 1 + search.assert_not_called() + assert "repeated application filters must be joined with OR" in stderr.getvalue() + + +def test_cli_search_allows_application_filters_joined_by_or() -> None: + stdout = StringIO() + + with patch( + "ctxd.cli.Client.search", + return_value=type( + "SearchResultLike", + (), + { + "model_dump": lambda self: { + "results": [], + "error": None, + "dsl_parse_error": None, + } + }, + )(), + ) as search, redirect_stdout(stdout): + exit_code = main( + [ + "search", + "application:google_drive", + "OR", + "application:slack", + "text:onboarding", + ] + ) + + assert exit_code == 0 + search.assert_called_once_with( + "application:google_drive OR application:slack text:onboarding" + ) + assert '"results": []' in stdout.getvalue() + + def test_cli_search_prints_clean_message_for_network_errors( monkeypatch: pytest.MonkeyPatch, ) -> None: From f641625934767074f08919a5774246bf23b32cbb Mon Sep 17 00:00:00 2001 From: juanpabloguerra16 Date: Tue, 2 Jun 2026 09:55:34 -0700 Subject: [PATCH 2/7] Simplify CLI search query guidance --- README.md | 6 ++--- src/ctxd/cli.py | 54 ++++++++++++++++++++++++++---------------- tests/test_cli.py | 60 +++++++++++++++++++++++++++++++++-------------- 3 files changed, 80 insertions(+), 40 deletions(-) diff --git a/README.md b/README.md index 78f0227..95a1b2d 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ ctxd login # Prompt for an API key and store it ctxd status # Check whether an API key is configured ctxd install-app # Open the app installation page ctxd search "text:deployment" -ctxd search text:test application:slack +ctxd search application:slack text:test ctxd fetch doc-123 ctxd profile ctxd logout # Remove the stored API key @@ -54,7 +54,7 @@ from ctxd import Client client = Client(api_key="") -results = client.search("text:deployment application:slack") +results = client.search("application:slack text:deployment") profile = client.get_profile() document = client.fetch_document("doc-123") ``` @@ -65,7 +65,7 @@ API key example: from ctxd import Client client = Client(api_key="") -results = client.search("text:deployment application:slack") +results = client.search("application:slack text:deployment") ``` Async example: diff --git a/src/ctxd/cli.py b/src/ctxd/cli.py index 37be103..4c714ee 100644 --- a/src/ctxd/cli.py +++ b/src/ctxd/cli.py @@ -61,7 +61,7 @@ def _build_parser() -> argparse.ArgumentParser: "Examples:\n" " ctxd login\n" " ctxd install-app\n" - " ctxd search text:deployment application:slack\n" + " ctxd search application:slack text:deployment\n" " ctxd fetch doc-123 --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, @@ -119,19 +119,17 @@ def _build_parser() -> argparse.ArgumentParser: ), epilog=( "Examples:\n" - " ctxd search text:deployment application:slack\n" - ' ctxd search "text:deployment application:slack"\n' - ' ctxd search \'text:"incident response" application:google_drive\'\n' - " ctxd search 'text:(incident response) application:google_drive'\n" - " ctxd search 'text:incident AND text:response application:google_drive'\n" - " ctxd search 'application:google_drive OR application:slack text:onboarding'\n" + " ctxd search text:deployment\n" + " ctxd search application:slack text:deployment\n" + " ctxd search application:google_drive text:incident response\n" + ' ctxd search application:google_drive \'text:"incident response"\'\n' "\n" - "DSL notes:\n" - ' text:"a b" runs a semantic multi-word text search.\n' - " text:(a b) matches any listed term, equivalent to text:a OR text:b.\n" - " text:a AND text:b requires both terms.\n" - " Use application:x OR application:y for multi-app unions; grouped or repeated\n" - " application filters are rejected because they can otherwise look successful." + "Query format:\n" + " application: text:\n" + " application: is optional; omit it to search all connected apps.\n" + " If application: is present, it must be the first token.\n" + " Only one application filter is supported.\n" + ' Use text:"a b" when your shell preserves the quotes and you need exact text.' ), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -139,7 +137,7 @@ def _build_parser() -> argparse.ArgumentParser: "query", nargs="+", metavar="QUERY", - help="Search query or DSL tokens, for example: text:deployment application:slack.", + help="Search query tokens, for example: application:slack text:deployment.", ) fetch_parser = subparsers.add_parser( @@ -197,6 +195,7 @@ def _quote_shell_stripped_text_token(token: str) -> str: def _validate_search_query(query: str) -> None: tokens = _split_search_query(query) _validate_application_filters(tokens) + _validate_boolean_operators(tokens) def _split_search_query(query: str) -> list[str]: @@ -220,16 +219,31 @@ def _validate_application_filters(tokens: Sequence[str]) -> None: if value.startswith("("): raise ValueError( "Invalid search query: grouped application filters are not supported. " - "Use application:google_drive OR application:slack." + "Use one leading application filter, for example application:google_drive text:onboarding." ) application_positions.append(index) - for left, right in zip(application_positions, application_positions[1:]): - between = [token.upper() for token in tokens[left + 1 : right]] - if "OR" not in between: + if not application_positions: + return + if application_positions[0] != 0: + raise ValueError( + "Invalid search query: application: must be the first token, " + "for example application:slack text:deployment." + ) + if len(application_positions) > 1: + raise ValueError( + "Invalid search query: only one application filter is supported. " + "Omit application: to search all connected apps." + ) + + +def _validate_boolean_operators(tokens: Sequence[str]) -> None: + operators = {"AND", "OR"} + for token in tokens: + if token.upper() in operators: raise ValueError( - "Invalid search query: repeated application filters must be joined with OR, " - "for example application:google_drive OR application:slack." + "Invalid search query: AND/OR clauses are not supported. " + "Use application: text:, or omit application: to search all apps." ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 1af87f6..f490cf4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -35,7 +35,7 @@ def test_cli_help_describes_commands() -> None: assert "Store an API key for future CLI and SDK calls." in output assert "install-app" in output assert "Open the app installation page" in output - assert "ctxd search text:deployment application:slack" in output + assert "ctxd search application:slack text:deployment" in output def test_cli_search_help_describes_query_and_json_output() -> None: @@ -49,10 +49,12 @@ def test_cli_search_help_describes_query_and_json_output() -> None: assert "Search indexed app content using ctxd DSL." in output assert "Search output is always JSON." in output assert "QUERY" in output - assert "text:deployment application:slack" in output - assert 'text:"a b" runs a semantic multi-word text search.' in output - assert "text:(a b) matches any listed term" in output - assert "application:x OR application:y" in output + assert "ctxd search text:deployment" in output + assert "ctxd search application:slack text:deployment" in output + assert "application: is optional" in output + assert "must be the first token" in output + assert "Only one application filter is supported" in output + assert 'Use text:"a b"' in output def test_cli_profile_json_calls_sdk() -> None: @@ -409,7 +411,24 @@ def test_cli_search_rejects_grouped_application_filter() -> None: assert "grouped application filters are not supported" in stderr.getvalue() -def test_cli_search_rejects_repeated_application_filters_without_or() -> None: +def test_cli_search_rejects_application_filter_after_text() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main( + [ + "search", + "text:onboarding", + "application:google_drive", + ] + ) + + assert exit_code == 1 + search.assert_not_called() + assert "application: must be the first token" in stderr.getvalue() + + +def test_cli_search_rejects_repeated_application_filters() -> None: stderr = StringIO() with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): @@ -424,10 +443,21 @@ def test_cli_search_rejects_repeated_application_filters_without_or() -> None: assert exit_code == 1 search.assert_not_called() - assert "repeated application filters must be joined with OR" in stderr.getvalue() + assert "only one application filter is supported" in stderr.getvalue() + + +def test_cli_search_rejects_boolean_operators() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main(["search", "text:incident", "AND", "text:response"]) + + assert exit_code == 1 + search.assert_not_called() + assert "AND/OR clauses are not supported" in stderr.getvalue() -def test_cli_search_allows_application_filters_joined_by_or() -> None: +def test_cli_search_accepts_leading_application_filter() -> None: stdout = StringIO() with patch( @@ -448,16 +478,12 @@ def test_cli_search_allows_application_filters_joined_by_or() -> None: [ "search", "application:google_drive", - "OR", - "application:slack", "text:onboarding", ] ) assert exit_code == 0 - search.assert_called_once_with( - "application:google_drive OR application:slack text:onboarding" - ) + search.assert_called_once_with("application:google_drive text:onboarding") assert '"results": []' in stdout.getvalue() @@ -529,10 +555,10 @@ def test_cli_search_accepts_unquoted_query_tokens() -> None: }, )(), ) as search, redirect_stdout(stdout): - exit_code = main(["search", "text:test", "application:slack"]) + exit_code = main(["search", "application:slack", "text:test"]) assert exit_code == 0 - search.assert_called_once_with("text:test application:slack") + search.assert_called_once_with("application:slack text:test") assert '"results": []' in stdout.getvalue() @@ -553,10 +579,10 @@ def test_cli_search_restores_shell_stripped_text_quotes() -> None: }, )(), ) as search, redirect_stdout(stdout): - exit_code = main(["search", "text:deployment process", "application:slack"]) + exit_code = main(["search", "application:slack", "text:deployment process"]) assert exit_code == 0 - search.assert_called_once_with('text:"deployment process" application:slack') + search.assert_called_once_with('application:slack text:"deployment process"') assert '"results": []' in stdout.getvalue() From 090dfedf2b41eed3a655d7d1bd94040c048ccff7 Mon Sep 17 00:00:00 2001 From: juanpabloguerra16 Date: Wed, 3 Jun 2026 14:05:30 -0700 Subject: [PATCH 3/7] Reject complex text search forms --- src/ctxd/cli.py | 52 +++++++++++++++++++++++------------------------ tests/test_cli.py | 28 ++++++++++++++++++++++--- 2 files changed, 51 insertions(+), 29 deletions(-) diff --git a/src/ctxd/cli.py b/src/ctxd/cli.py index 4c714ee..0bd43bb 100644 --- a/src/ctxd/cli.py +++ b/src/ctxd/cli.py @@ -115,21 +115,20 @@ def _build_parser() -> argparse.ArgumentParser: help="Search indexed app content and print JSON results.", description=( "Search indexed app content using ctxd DSL. Search output is always JSON. " - "The query can be quoted or passed as separate tokens." + "The query can be passed as separate tokens." ), epilog=( "Examples:\n" " ctxd search text:deployment\n" " ctxd search application:slack text:deployment\n" " ctxd search application:google_drive text:incident response\n" - ' ctxd search application:google_drive \'text:"incident response"\'\n' "\n" "Query format:\n" " application: text:\n" " application: is optional; omit it to search all connected apps.\n" " If application: is present, it must be the first token.\n" " Only one application filter is supported.\n" - ' Use text:"a b" when your shell preserves the quotes and you need exact text.' + " Use simple text terms; quoted and parenthesized text values are not supported." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -167,34 +166,14 @@ def _build_parser() -> argparse.ArgumentParser: def _normalize_search_query(query_tokens: Sequence[str]) -> str: - normalized_tokens = [ - _quote_shell_stripped_text_token(token) for token in query_tokens - ] - return " ".join(normalized_tokens) - - -def _quote_shell_stripped_text_token(token: str) -> str: - if not token.lower().startswith("text:"): - return token - - value = token[5:] - if not value or not re.search(r"\s", value): - return token - - stripped_value = value.strip() - if ( - stripped_value.startswith(("\"", "'", "(")) - or stripped_value.endswith(("\"", "'", ")")) - ): - return token - - escaped_value = stripped_value.replace("\\", "\\\\").replace('"', '\\"') - return f'text:"{escaped_value}"' + return " ".join(query_tokens) def _validate_search_query(query: str) -> None: + _validate_raw_text_filters(query) tokens = _split_search_query(query) _validate_application_filters(tokens) + _validate_text_filters(tokens) _validate_boolean_operators(tokens) @@ -205,6 +184,14 @@ def _split_search_query(query: str) -> list[str]: return query.split() +def _validate_raw_text_filters(query: str) -> None: + if re.search(r"(?i)(?:^|\s)text:[\"'()]", query): + raise ValueError( + "Invalid search query: quoted and parenthesized text values are not supported. " + "Use simple text terms, for example text:incident response." + ) + + def _validate_application_filters(tokens: Sequence[str]) -> None: application_positions: list[int] = [] for index, token in enumerate(tokens): @@ -237,6 +224,19 @@ def _validate_application_filters(tokens: Sequence[str]) -> None: ) +def _validate_text_filters(tokens: Sequence[str]) -> None: + for token in tokens: + if not token.lower().startswith("text:"): + continue + + value = token[len("text:") :] + if value.startswith(("\"", "'", "(")) or value.endswith(("\"", "'", ")")): + raise ValueError( + "Invalid search query: quoted and parenthesized text values are not supported. " + "Use simple text terms, for example text:incident response." + ) + + def _validate_boolean_operators(tokens: Sequence[str]) -> None: operators = {"AND", "OR"} for token in tokens: diff --git a/tests/test_cli.py b/tests/test_cli.py index f490cf4..e80e37b 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -54,7 +54,7 @@ def test_cli_search_help_describes_query_and_json_output() -> None: assert "application: is optional" in output assert "must be the first token" in output assert "Only one application filter is supported" in output - assert 'Use text:"a b"' in output + assert "quoted and parenthesized text values are not supported" in output def test_cli_profile_json_calls_sdk() -> None: @@ -457,6 +457,28 @@ def test_cli_search_rejects_boolean_operators() -> None: assert "AND/OR clauses are not supported" in stderr.getvalue() +def test_cli_search_rejects_quoted_text_filter() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main(["search", "application:slack", 'text:"incident response"']) + + assert exit_code == 1 + search.assert_not_called() + assert "quoted and parenthesized text values are not supported" in stderr.getvalue() + + +def test_cli_search_rejects_parenthesized_text_filter() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main(["search", "application:slack", "text:(incident response)"]) + + assert exit_code == 1 + search.assert_not_called() + assert "quoted and parenthesized text values are not supported" in stderr.getvalue() + + def test_cli_search_accepts_leading_application_filter() -> None: stdout = StringIO() @@ -562,7 +584,7 @@ def test_cli_search_accepts_unquoted_query_tokens() -> None: assert '"results": []' in stdout.getvalue() -def test_cli_search_restores_shell_stripped_text_quotes() -> None: +def test_cli_search_keeps_multi_word_text_simple() -> None: stdout = StringIO() with patch( @@ -582,7 +604,7 @@ def test_cli_search_restores_shell_stripped_text_quotes() -> None: exit_code = main(["search", "application:slack", "text:deployment process"]) assert exit_code == 0 - search.assert_called_once_with('application:slack text:"deployment process"') + search.assert_called_once_with("application:slack text:deployment process") assert '"results": []' in stdout.getvalue() From e1b50ece0b8971f612f9b4c3218032c8bcfb1184 Mon Sep 17 00:00:00 2001 From: juanpabloguerra16 Date: Wed, 3 Jun 2026 14:48:08 -0700 Subject: [PATCH 4/7] Allow lowercase boolean words in search text --- src/ctxd/cli.py | 2 +- tests/test_cli.py | 26 ++++++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/src/ctxd/cli.py b/src/ctxd/cli.py index 0bd43bb..411f8a5 100644 --- a/src/ctxd/cli.py +++ b/src/ctxd/cli.py @@ -240,7 +240,7 @@ def _validate_text_filters(tokens: Sequence[str]) -> None: def _validate_boolean_operators(tokens: Sequence[str]) -> None: operators = {"AND", "OR"} for token in tokens: - if token.upper() in operators: + if token in operators: raise ValueError( "Invalid search query: AND/OR clauses are not supported. " "Use application: text:, or omit application: to search all apps." diff --git a/tests/test_cli.py b/tests/test_cli.py index e80e37b..d0b185d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -457,6 +457,32 @@ def test_cli_search_rejects_boolean_operators() -> None: assert "AND/OR clauses are not supported" in stderr.getvalue() +def test_cli_search_allows_lowercase_and_or_as_text_terms() -> None: + stdout = StringIO() + + with patch( + "ctxd.cli.Client.search", + return_value=type( + "SearchResultLike", + (), + { + "model_dump": lambda self: { + "results": [], + "error": None, + "dsl_parse_error": None, + } + }, + )(), + ) as search, redirect_stdout(stdout): + exit_code = main( + ["search", "text:research", "and", "development", "or", "testing"] + ) + + assert exit_code == 0 + search.assert_called_once_with("text:research and development or testing") + assert '"results": []' in stdout.getvalue() + + def test_cli_search_rejects_quoted_text_filter() -> None: stderr = StringIO() From d03033d04f47af960618e54f469713a1b35485f2 Mon Sep 17 00:00:00 2001 From: juanpabloguerra16 Date: Wed, 3 Jun 2026 14:58:42 -0700 Subject: [PATCH 5/7] Reject unsupported multi-word text filters --- src/ctxd/cli.py | 27 +++++++++++++++++++-------- tests/test_cli.py | 47 ++++++++++++++++++++++------------------------- 2 files changed, 41 insertions(+), 33 deletions(-) diff --git a/src/ctxd/cli.py b/src/ctxd/cli.py index 411f8a5..8b428e2 100644 --- a/src/ctxd/cli.py +++ b/src/ctxd/cli.py @@ -121,14 +121,13 @@ def _build_parser() -> argparse.ArgumentParser: "Examples:\n" " ctxd search text:deployment\n" " ctxd search application:slack text:deployment\n" - " ctxd search application:google_drive text:incident response\n" "\n" "Query format:\n" - " application: text:\n" + " application: text:\n" " application: is optional; omit it to search all connected apps.\n" " If application: is present, it must be the first token.\n" " Only one application filter is supported.\n" - " Use simple text terms; quoted and parenthesized text values are not supported." + " Use one text term; multi-word, quoted, and parenthesized text values are not supported." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -173,8 +172,8 @@ def _validate_search_query(query: str) -> None: _validate_raw_text_filters(query) tokens = _split_search_query(query) _validate_application_filters(tokens) - _validate_text_filters(tokens) _validate_boolean_operators(tokens) + _validate_text_filters(tokens) def _split_search_query(query: str) -> list[str]: @@ -188,7 +187,7 @@ def _validate_raw_text_filters(query: str) -> None: if re.search(r"(?i)(?:^|\s)text:[\"'()]", query): raise ValueError( "Invalid search query: quoted and parenthesized text values are not supported. " - "Use simple text terms, for example text:incident response." + "Use a single text term, for example text:incident." ) @@ -225,7 +224,7 @@ def _validate_application_filters(tokens: Sequence[str]) -> None: def _validate_text_filters(tokens: Sequence[str]) -> None: - for token in tokens: + for index, token in enumerate(tokens): if not token.lower().startswith("text:"): continue @@ -233,7 +232,19 @@ def _validate_text_filters(tokens: Sequence[str]) -> None: if value.startswith(("\"", "'", "(")) or value.endswith(("\"", "'", ")")): raise ValueError( "Invalid search query: quoted and parenthesized text values are not supported. " - "Use simple text terms, for example text:incident response." + "Use a single text term, for example text:incident." + ) + if re.search(r"\s", value): + raise ValueError( + "Invalid search query: multi-word text values are not supported by the current DSL. " + "Use a single text term, for example text:incident." + ) + if index + 1 < len(tokens) and not tokens[index + 1].lower().startswith( + ("application:", "text:") + ): + raise ValueError( + "Invalid search query: multi-word text values are not supported by the current DSL. " + "Use a single text term, for example text:incident." ) @@ -243,7 +254,7 @@ def _validate_boolean_operators(tokens: Sequence[str]) -> None: if token in operators: raise ValueError( "Invalid search query: AND/OR clauses are not supported. " - "Use application: text:, or omit application: to search all apps." + "Use application: text:, or omit application: to search all apps." ) diff --git a/tests/test_cli.py b/tests/test_cli.py index d0b185d..1090eb3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -54,7 +54,7 @@ def test_cli_search_help_describes_query_and_json_output() -> None: assert "application: is optional" in output assert "must be the first token" in output assert "Only one application filter is supported" in output - assert "quoted and parenthesized text values are not supported" in output + assert "multi-word, quoted, and parenthesized text values are not supported" in output def test_cli_profile_json_calls_sdk() -> None: @@ -457,7 +457,8 @@ def test_cli_search_rejects_boolean_operators() -> None: assert "AND/OR clauses are not supported" in stderr.getvalue() -def test_cli_search_allows_lowercase_and_or_as_text_terms() -> None: +@pytest.mark.parametrize("term", ["and", "or"]) +def test_cli_search_allows_lowercase_and_or_as_text_terms(term: str) -> None: stdout = StringIO() with patch( @@ -474,12 +475,10 @@ def test_cli_search_allows_lowercase_and_or_as_text_terms() -> None: }, )(), ) as search, redirect_stdout(stdout): - exit_code = main( - ["search", "text:research", "and", "development", "or", "testing"] - ) + exit_code = main(["search", f"text:{term}"]) assert exit_code == 0 - search.assert_called_once_with("text:research and development or testing") + search.assert_called_once_with(f"text:{term}") assert '"results": []' in stdout.getvalue() @@ -610,28 +609,26 @@ def test_cli_search_accepts_unquoted_query_tokens() -> None: assert '"results": []' in stdout.getvalue() -def test_cli_search_keeps_multi_word_text_simple() -> None: - stdout = StringIO() +def test_cli_search_rejects_shell_stripped_multi_word_text_filter() -> None: + stderr = StringIO() - with patch( - "ctxd.cli.Client.search", - return_value=type( - "SearchResultLike", - (), - { - "model_dump": lambda self: { - "results": [], - "error": None, - "dsl_parse_error": None, - } - }, - )(), - ) as search, redirect_stdout(stdout): + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): exit_code = main(["search", "application:slack", "text:deployment process"]) - assert exit_code == 0 - search.assert_called_once_with("application:slack text:deployment process") - assert '"results": []' in stdout.getvalue() + assert exit_code == 1 + search.assert_not_called() + assert "multi-word text values are not supported" in stderr.getvalue() + + +def test_cli_search_rejects_text_continuation_terms() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main(["search", "application:slack", "text:deployment", "process"]) + + assert exit_code == 1 + search.assert_not_called() + assert "multi-word text values are not supported" in stderr.getvalue() def test_cli_search_outputs_json_for_empty_success() -> None: From 0424c0d715d612d5cece99c54eaf5500096b61da Mon Sep 17 00:00:00 2001 From: juanpabloguerra16 Date: Thu, 4 Jun 2026 17:54:13 -0700 Subject: [PATCH 6/7] Reject repeated text search filters --- src/ctxd/cli.py | 8 ++++++++ tests/test_cli.py | 13 +++++++++++++ 2 files changed, 21 insertions(+) diff --git a/src/ctxd/cli.py b/src/ctxd/cli.py index 8b428e2..87c1b58 100644 --- a/src/ctxd/cli.py +++ b/src/ctxd/cli.py @@ -224,10 +224,12 @@ def _validate_application_filters(tokens: Sequence[str]) -> None: def _validate_text_filters(tokens: Sequence[str]) -> None: + text_positions: list[int] = [] for index, token in enumerate(tokens): if not token.lower().startswith("text:"): continue + text_positions.append(index) value = token[len("text:") :] if value.startswith(("\"", "'", "(")) or value.endswith(("\"", "'", ")")): raise ValueError( @@ -247,6 +249,12 @@ def _validate_text_filters(tokens: Sequence[str]) -> None: "Use a single text term, for example text:incident." ) + if len(text_positions) > 1: + raise ValueError( + "Invalid search query: only one text filter is supported. " + "Use a single text term, for example text:incident." + ) + def _validate_boolean_operators(tokens: Sequence[str]) -> None: operators = {"AND", "OR"} diff --git a/tests/test_cli.py b/tests/test_cli.py index 1090eb3..78c2cee 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -631,6 +631,19 @@ def test_cli_search_rejects_text_continuation_terms() -> None: assert "multi-word text values are not supported" in stderr.getvalue() +def test_cli_search_rejects_repeated_text_filters() -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main( + ["search", "application:slack", "text:incident", "text:response"] + ) + + assert exit_code == 1 + search.assert_not_called() + assert "only one text filter is supported" in stderr.getvalue() + + def test_cli_search_outputs_json_for_empty_success() -> None: stdout = StringIO() From d73caaba74d80877cfe3662b915d19f47e4e9acf Mon Sep 17 00:00:00 2001 From: juanpabloguerra16 Date: Thu, 4 Jun 2026 17:59:07 -0700 Subject: [PATCH 7/7] Require text filter for app search --- src/ctxd/cli.py | 11 +++++++++++ tests/test_cli.py | 20 ++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/src/ctxd/cli.py b/src/ctxd/cli.py index 87c1b58..91bd289 100644 --- a/src/ctxd/cli.py +++ b/src/ctxd/cli.py @@ -174,6 +174,7 @@ def _validate_search_query(query: str) -> None: _validate_application_filters(tokens) _validate_boolean_operators(tokens) _validate_text_filters(tokens) + _validate_application_scoped_search_shape(tokens) def _split_search_query(query: str) -> list[str]: @@ -266,6 +267,16 @@ def _validate_boolean_operators(tokens: Sequence[str]) -> None: ) +def _validate_application_scoped_search_shape(tokens: Sequence[str]) -> None: + if not tokens or not tokens[0].lower().startswith("application:"): + return + if len(tokens) != 2 or not tokens[1].lower().startswith("text:"): + raise ValueError( + "Invalid search query: application: must be followed by one text: filter, " + "for example application:slack text:deployment." + ) + + def _handle_login(args: argparse.Namespace) -> int: del args api_key, should_save = _resolve_login_api_key() diff --git a/tests/test_cli.py b/tests/test_cli.py index 78c2cee..3a6ca3c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -644,6 +644,26 @@ def test_cli_search_rejects_repeated_text_filters() -> None: assert "only one text filter is supported" in stderr.getvalue() +@pytest.mark.parametrize( + "query", + [ + ["search", "application:slack"], + ["search", "application:slack", "deployment"], + ], +) +def test_cli_search_rejects_application_filter_without_text_filter( + query: list[str], +) -> None: + stderr = StringIO() + + with patch("ctxd.cli.Client.search") as search, patch("sys.stderr", stderr): + exit_code = main(query) + + assert exit_code == 1 + search.assert_not_called() + assert "must be followed by one text:" in stderr.getvalue() + + def test_cli_search_outputs_json_for_empty_success() -> None: stdout = StringIO()