✅ Added sort test cases

pyproject.toml

@@ -98,7 +98,7 @@ warn_unused_configs = true
 pythonpath = ["src"]
 
 [tool.pytest.ini_options]
-addopts = "--cov=. --cov-report=xml --cov-report term-missing --ff -x --no-cov-on-fail --emoji -n4 -rP"
+addopts = "--cov=. --cov-report=xml --cov-report term-missing --ff -x --no-cov-on-fail --emoji -n4"
 
 [tool.coverage.run]
 #branch = true #https://github.com/nedbat/coveragepy/issues/605

src/click_opt/click_custom.py

@@ -7,6 +7,8 @@
 from click import Context, Parameter, ParamType
 from click_params.miscellaneous import JsonParamType
 
+from src.strings import invalid_query_format, invalid_sort_format
+
 if TYPE_CHECKING:
  from typing_extensions import Self
 
@@ -35,14 +37,10 @@ def convert(self: Self, value: Any, param: Parameter | None, ctx: Context | None
  except FormatError as e:
  self.fail(str(e), param, ctx)
  except ValueError:
- self.fail(f'Invalid input format: "{value}". Use the format "field:sort_order".', param, ctx)
+ self.fail(invalid_sort_format.format(value=value), param, ctx)
  else:
  return {field: sort_order}
 
- def __repr__(self: Self) -> str:
- """Return a string representation."""
- return str(self.name)
-
 
 sort = Sort()
 
@@ -55,8 +53,6 @@ class Json(JsonParamType): # type: ignore[misc]
  def convert(self: Self, value: Any, param: Parameter, ctx: Context) -> dict[str, Any]: # type: ignore[return]
  """Convert input to json."""
  try:
- if isinstance(value, dict):
- return value
  return json.loads( # type: ignore[no-any-return]
  value,
  cls=self._cls,
@@ -68,11 +64,7 @@ def convert(self: Self, value: Any, param: Parameter, ctx: Context) -> dict[str,
  **self._kwargs,
  )
  except json.JSONDecodeError as exc:
- self.fail(f"{value} is not a valid json string, caused {exc}", param, ctx)
-
- def __repr__(self: Self) -> str:
- """String representation of the object."""
- return self.name.upper()
+ self.fail(invalid_query_format.format(value=value, exc=exc), param, ctx)
 
 
 JSON = Json()

src/strings.py

@@ -5,4 +5,5 @@
 output_fields = "Output fields : {fields}."
 sorting_by = "Sorting by: {sort}."
 meta_field_not_found = "Meta Field {field} not found"
-meta_field_not_found = "Meta Field {field} not found"
+invalid_sort_format = 'Invalid input format: "{value}". Use the format "field:sort_order".'
+invalid_query_format = "{value} is not a valid json string, caused {exc}"

test/click/cli_test.py

@@ -1,21 +1,26 @@
 """Click CLI test cases."""
 import inspect
+import json
 from pathlib import Path
 from test.esxport._export_test import TestExport
+from test.esxport._prepare_search_query_test import TestSearchQuery
 from unittest.mock import patch
 
 from click.testing import CliRunner
 from typing_extensions import Self
 
 from src.esxport import EsXport
 from src.esxport_cli import cli
+from src.strings import invalid_query_format, invalid_sort_format
 
 args = {
  "q": '{"query":{"match_all":{}}}',
  "o": "output.csv",
  "i": "index1",
 }
 usage_error_code = 2
+random_pass = "password\n" # noqa: S105
+export_module = "src.esxport.EsXport"
 
 
 # noinspection PyTypeChecker
@@ -46,7 +51,7 @@ def test_index_is_mandatory(self: Self, cli_runner: CliRunner) -> None:
  def test_mandatory(self: Self, cli_runner: CliRunner, esxport_obj_with_data: EsXport) -> None:
  """Test Index param is mandatory."""
  esxport_obj_with_data.opts.output_file = f"{inspect.stack()[0].function}.csv"
- with patch("src.esxport.EsXport", return_value=esxport_obj_with_data):
+ with patch(export_module, return_value=esxport_obj_with_data):
  result = cli_runner.invoke(
  cli,
  ["-q", args["q"], "-o", args["o"], "-i", args["i"]],
@@ -58,3 +63,68 @@ def test_mandatory(self: Self, cli_runner: CliRunner, esxport_obj_with_data: EsX
  lines = len(fp.readlines())
  assert lines == esxport_obj_with_data.es_client.search()["hits"]["total"]["value"] + 1 # 1 for header
  TestExport.rm_csv_export_file(esxport_obj_with_data.opts.output_file)
+
+ def test_sort_type(self: Self, cli_runner: CliRunner, esxport_obj_with_data: EsXport) -> None:
+ """Test sort type is asc or desc."""
+ esxport_obj_with_data.opts.output_file = f"{inspect.stack()[0].function}.csv"
+ random_string = TestSearchQuery.random_string(10)
+ with patch(export_module, return_value=esxport_obj_with_data):
+ result = cli_runner.invoke(
+ cli,
+ ["-q", args["q"], "-o", args["o"], "-i", args["i"], "-S", f"field:{random_string}"],
+ input=random_pass,
+ catch_exceptions=False,
+ )
+ error_msg = f"Error: Invalid value for '-S' / '--sort': Invalid sort type {random_string}."
+ assert error_msg in result.output
+ assert result.exit_code == usage_error_code
+
+ result = cli_runner.invoke(
+ cli,
+ ["-q", args["q"], "-o", args["o"], "-i", args["i"], "-S", "field:desc"],
+ input=random_pass,
+ catch_exceptions=False,
+ )
+ error_msg = f"Error: Invalid value for '-S' / '--sort': Invalid sort type {random_string}."
+ assert error_msg not in result.output
+ assert result.exit_code == 0
+ TestExport.rm_csv_export_file(esxport_obj_with_data.opts.output_file)
+
+ def test_sort_format(self: Self, cli_runner: CliRunner, esxport_obj_with_data: EsXport) -> None:
+ """Test sort input is in the form field:sort_order."""
+ esxport_obj_with_data.opts.output_file = f"{inspect.stack()[0].function}.csv"
+ random_string = TestSearchQuery.random_string(10)
+ with patch(export_module, return_value=esxport_obj_with_data):
+ result = cli_runner.invoke(
+ cli,
+ ["-q", args["q"], "-o", args["o"], "-i", args["i"], "-S", f"field@{random_string}"],
+ input=random_pass,
+ catch_exceptions=False,
+ )
+ error_msg = invalid_sort_format.format(value=f"field@{random_string}")
+ assert error_msg in result.output
+ assert result.exit_code == usage_error_code
+
+ def test_query_accepts_dict(self: Self, cli_runner: CliRunner, esxport_obj_with_data: EsXport) -> None:
+ """Test sort input is in the form field:sort_order."""
+ esxport_obj_with_data.opts.output_file = f"{inspect.stack()[0].function}.csv"
+ with patch(export_module, return_value=esxport_obj_with_data):
+ result = cli_runner.invoke(
+ cli,
+ ["-q", json.dumps(args["q"]), "-o", args["o"], "-i", args["i"]],
+ input=random_pass,
+ catch_exceptions=False,
+ )
+ assert result.exit_code == 0
+
+ def test_error_is_rasied_on_invalid_json(self: Self, cli_runner: CliRunner) -> None:
+ """Test sort input is in the form field:sort_order."""
+ result = cli_runner.invoke(
+ cli,
+ ["-q", "@", "-o", args["o"], "-i", args["i"]],
+ input=random_pass,
+ catch_exceptions=False,
+ )
+ json_error_message = invalid_query_format.format(value="@", exc="")
+ assert json_error_message in result.output
+ assert result.exit_code == usage_error_code

test/esxport/_export_test.py

@@ -63,6 +63,7 @@ def test_headers_extraction(
  with Path(f"{esxport_obj.opts.output_file}.tmp").open(mode="w", encoding="utf-8") as tmp_file:
  tmp_file.write(json.dumps(test_json))
  tmp_file.write("\n")
+ assert Path(f"{esxport_obj.opts.output_file}.tmp").exists() is True
  keys = list(test_json.keys())
  assert esxport_obj._extract_headers() == keys
  TestExport.rm_export_file(f"{inspect.stack()[0].function}.csv")