diff --git a/skills/fessctl/SKILL.md b/skills/fessctl/SKILL.md index 14d5398..a77f63e 100644 --- a/skills/fessctl/SKILL.md +++ b/skills/fessctl/SKILL.md @@ -1,6 +1,6 @@ --- name: fessctl -description: Operate Fess via fessctl CLI (admin API client). Use when managing webconfigs, file/data configs, schedulers, users/roles/groups, dictionaries, access tokens, or any Fess admin resource from Claude. Covers 30 resource types and cross-feature workflows like initial crawl setup. +description: Operate Fess via fessctl CLI (admin API client). Use when managing webconfigs, file/data configs, schedulers, users/roles/groups, dictionaries, access tokens, or any Fess admin resource from Claude. Covers 31 resource types and cross-feature workflows like initial crawl setup. license: Apache-2.0 version: 0.1.0 --- @@ -61,6 +61,7 @@ Each file documents one Fess admin feature: what it is, when to use it, fessctl | Label types | references/features/labeltype.md | | User tags | references/features/tagtype.md | | Themes | references/features/theme.md | +| Dictionaries | references/features/dict.md | | Key match | references/features/keymatch.md | | Boost document | references/features/boostdoc.md | | Elevate word | references/features/elevateword.md | diff --git a/skills/fessctl/references/features/dict.md b/skills/fessctl/references/features/dict.md new file mode 100644 index 0000000..cf9d341 --- /dev/null +++ b/skills/fessctl/references/features/dict.md @@ -0,0 +1,66 @@ +# Dictionaries (fessctl `dict`) + +## What it is +Fess keeps its analyzer dictionaries as text files in the search engine's dictionary directory: synonyms, character mappings, Japanese (kuromoji) user words, stop words, protected words and stemmer overrides. The Admin UI edits them under **System → Dictionary**; the admin API lists the files and manages both their entries and the whole file. + +`fessctl dict list` shows the files (`type`, `id`, `path`, `timestamp`). Every other command is `fessctl dict ...`, where `` is the `id` from `dict list` (a URL-safe base64 of the path, e.g. `c3lub255bS50eHQ=` for `synonym.txt`). Which files exist depends on the search engine image; most languages have `stopwords`/`protwords`/`stemmeroverride` files, `ja` has `kuromoji` and `mapping`, and there is one `synonym.txt`. + +Available in all supported versions (14.19.2, 15.6.1, 15.7.0, 15.8.0, 15.9.0) with the same fields, except for `upload` (see Gotchas). + +## Subcommand surface +| Subcommand | Purpose | Required arguments | +| --- | --- | --- | +| `dict list` | List the dictionary files; `--type` narrows to one type. | none | +| `dict list` | Page through the entries of one file (`--page`, `--size`). | `` | +| `dict get` | Show one entry. | ` ` | +| `dict create` | Add an entry. | `` and every field option of the type | +| `dict update` | Change an entry; unspecified fields are kept. | ` ` | +| `dict delete` | Delete one entry. | ` ` | +| `dict upload` | Replace the whole file with a local file. Asks for confirmation unless `--yes`. | ` ` | +| `dict download` | Write the file to standard output, or to `--file` (refuses to overwrite unless `--force`). | `` | + +Always reconfirm with `fessctl dict --help`. + +## Types and their fields +| `` | Options of `create` / `update` | API fields | +| --- | --- | --- | +| `synonym` | `--source` (repeatable), `--target` (repeatable) | `inputs`, `outputs` | +| `mapping` | `--source` (repeatable), `--target` | `inputs`, `output` | +| `kuromoji` | `--token`, `--segmentation`, `--reading`, `--pos` | same names | +| `stopwords` | `--word` | `input` | +| `protwords` | `--word` | `input` | +| `stemmeroverride` | `--source`, `--target` | `input`, `output` | + +A repeatable option is sent as one value per line (`inputs` is `"a\nb"`). + +## Resource JSON shape +`list` returns the entries under `response.settings`, `get` returns one under `response.setting`: +```json +{"id": 2, "dict_id": "c3lub255bS50eHQ=", "inputs": "tv\ntelevision", "outputs": "tv"} +``` +`dict list` returns `{"id": "c3lub255bS50eHQ=", "type": "synonym", "path": "synonym.txt", "timestamp": "2026-10-16T12:24:48"}` rows. + +## Gotchas +- **`upload` is impossible on Fess 15.6 to 15.8.** Those servers route the upload only as PUT and cannot read a multipart PUT (the error is "File is required"); 15.9 added POST, and 14.x always used POST. `fessctl` therefore refuses an upload on 15.6 to 15.8 with an explanatory error. Use `download`, edit, and the Admin UI there. +- `upload` replaces the whole file, including entries added with `create`; `download` first when unsure. The file is in the engine's plain-text dictionary format; `download` one to see it (a synonym rule is a line such as `a,b=>c,d`). +- Entry IDs are positions in the file. Deleting an entry can shift the IDs of the ones after it, and a new entry may reuse an ID, so look an entry up again (by its content) before updating or deleting it after any change. +- Fess before 15.9 reports the ID of a new entry as `0`; `create` then prints no ID. Find it with `list`. +- `update` reads the entry first and sends it back with only the given options changed (the API's PUT replaces the whole entry). A repeatable option such as `--target` replaces all the values; it does not append. +- `list` and `get` of an unknown ``: `list` shows nothing (exit 0), `get` fails. `create` in an unknown dictionary fails on 15.9 but reports success without storing anything on earlier versions. +- Fess validates entries: synonym and mapping sources may not contain `,` or `=>` (15.9), a mapping source may not already exist, a kuromoji token may be registered once, and the number of words in `--segmentation` and `--reading` must match. +- Error messages of `download` and `upload` on a bad `` mention "protwords" for every type; that is the server's wording. + +## Examples +```bash +fessctl dict list --type synonym +fessctl dict synonym create c3lub255bS50eHQ= --source tv --source television --target tv +fessctl dict synonym list c3lub255bS50eHQ= +fessctl dict synonym update c3lub255bS50eHQ= 2 --target "tv set" +fessctl dict synonym delete c3lub255bS50eHQ= 2 + +fessctl dict kuromoji create amEva3Vyb21vamkudHh0 \ + --token 令和 --segmentation 令和 --reading レイワ --pos 名詞-固有名詞-一般 + +fessctl dict stopwords download ZW4vc3RvcHdvcmRzLnR4dA== --file stopwords.txt +fessctl dict stopwords upload ZW4vc3RvcHdvcmRzLnR4dA== stopwords.txt --yes +``` diff --git a/src/fessctl/api/client.py b/src/fessctl/api/client.py index a461af0..f442ff8 100644 --- a/src/fessctl/api/client.py +++ b/src/fessctl/api/client.py @@ -1,4 +1,5 @@ import json +import os from enum import Enum from pathlib import Path from typing import Dict, List, Optional @@ -16,6 +17,22 @@ class Action(Enum): GET = "get" START = "start" STOP = "stop" + UPLOAD = "upload" + + +# Name of the multipart form field that carries the file, per dictionary type. +DICT_UPLOAD_FIELDS = { + "synonym": "synonymFile", + "mapping": "charMappingFile", + "kuromoji": "kuromojiFile", + "stopwords": "stopwordsFile", + "protwords": "protwordsFile", + "stemmeroverride": "stemmerOverrideFile", +} + + +# Seconds to allow for uploading or downloading a dictionary file. +FILE_TRANSFER_TIMEOUT = 60.0 class FessAPIClientError(Exception): @@ -68,6 +85,7 @@ def send_request( json_data: dict | None = None, params: dict | None = None, is_admin: bool = True, + files: dict | None = None, ) -> dict: headers = self.admin_api_headers if is_admin else self.search_api_headers try: @@ -103,6 +121,14 @@ def send_request( response = httpx.get( url, headers=headers, params=params, timeout=self.timeout ) + elif action == Action.UPLOAD: + # httpx sets the multipart Content-Type (with its boundary) itself. + upload_headers = { + k: v for k, v in headers.items() if k != "Content-Type"} + response = httpx.post( + url, headers=upload_headers, files=files, params=params, + timeout=max(self.timeout, FILE_TRANSFER_TIMEOUT) + ) elif action == Action.START or action == Action.STOP: if self._major_version <= 14: response = httpx.post( @@ -185,6 +211,91 @@ def ping(self) -> dict: url = f"{self.base_url}/api/v1/health" return self.send_request(Action.GET, url, is_admin=False) + # Dictionary APIs + + def get_dicts(self) -> dict: + """ + Retrieves the dictionary files (type, id, path and timestamp). + """ + url = f"{self.base_url}/api/admin/dict" + return self.send_request(Action.GET, url) + + def list_dict_items(self, dict_type: str, dict_id: str, page: int = 1, size: int = 100) -> dict: + """ + Retrieves a page of the entries of a dictionary file. + """ + url = f"{self.base_url}/api/admin/dict/{dict_type}/settings/{dict_id}" + params = {"page": page, "size": size} + return self.send_request(Action.LIST, url, params=params) + + def get_dict_item(self, dict_type: str, dict_id: str, item_id: int) -> dict: + """ + Retrieves an entry of a dictionary file by ID. + """ + url = f"{self.base_url}/api/admin/dict/{dict_type}/setting/{dict_id}/{item_id}" + return self.send_request(Action.GET, url) + + def create_dict_item(self, dict_type: str, dict_id: str, config: dict) -> dict: + """ + Creates an entry in a dictionary file. + """ + url = f"{self.base_url}/api/admin/dict/{dict_type}/setting/{dict_id}" + return self.send_request(Action.CREATE, url, json_data=config) + + def update_dict_item(self, dict_type: str, dict_id: str, config: dict) -> dict: + """ + Replaces an entry of a dictionary file (config carries the entry's ID). + """ + url = f"{self.base_url}/api/admin/dict/{dict_type}/setting/{dict_id}" + return self.send_request(Action.EDIT, url, json_data=config) + + def delete_dict_item(self, dict_type: str, dict_id: str, item_id: int) -> dict: + """ + Deletes an entry of a dictionary file by ID. + """ + url = f"{self.base_url}/api/admin/dict/{dict_type}/setting/{dict_id}/{item_id}" + return self.send_request(Action.DELETE, url) + + def upload_dict(self, dict_type: str, dict_id: str, file_path: str) -> dict: + """ + Replaces a dictionary file with the content of a local file. + + Fess 15.6 to 15.8 only route the upload as PUT and cannot read a multipart + PUT, so no upload is possible there (fixed in 15.9, which accepts POST). + """ + if (15, 6) <= (self._major_version, self._minor_version) < (15, 9): + return {"response": { + "status": 1, + "message": "Dictionary upload is not supported by Fess 15.6 to 15.8 " + "(the server cannot read a multipart PUT); it needs Fess 15.9 or 14.x.", + }} + url = f"{self.base_url}/api/admin/dict/{dict_type}/upload/{dict_id}" + with open(file_path, "rb") as f: + files = {DICT_UPLOAD_FIELDS[dict_type]: ( + os.path.basename(file_path), f.read(), "application/octet-stream")} + return self.send_request(Action.UPLOAD, url, files=files) + + def download_dict(self, dict_type: str, dict_id: str) -> bytes: + """ + Downloads a dictionary file. Raises FessAPIClientError when Fess does not + return the file (it answers with a JSON error instead). + """ + url = f"{self.base_url}/api/admin/dict/{dict_type}/download/{dict_id}" + try: + response = httpx.get( + url, headers=self.admin_api_headers, + timeout=max(self.timeout, FILE_TRANSFER_TIMEOUT)) + except httpx.RequestError as e: + raise FessAPIClientError( + status_code=-1, content=f"Network error: {str(e)}") from e + if response.status_code != 200 or response.headers.get("content-type", "").startswith("application/json"): + try: + message = response.json().get("response", {}).get("message", "") + except ValueError: + message = response.text + raise FessAPIClientError(status_code=response.status_code, content=message) + return response.content + # Role APIs def create_role( diff --git a/src/fessctl/cli.py b/src/fessctl/cli.py index 882bdfb..9fe1aab 100644 --- a/src/fessctl/cli.py +++ b/src/fessctl/cli.py @@ -12,6 +12,7 @@ from fessctl.commands.boostdoc import boostdoc_app from fessctl.commands.crawlinginfo import crawlinginfo_app from fessctl.commands.dataconfig import dataconfig_app +from fessctl.commands.dict import dict_app # from fessctl.commands.documents import documents_app from fessctl.commands.duplicatehost import duplicatehost_app from fessctl.commands.elevateword import elevateword_app @@ -50,6 +51,7 @@ app.add_typer(boostdoc_app, name="boostdoc") app.add_typer(crawlinginfo_app, name="crawlinginfo") app.add_typer(dataconfig_app, name="dataconfig") +app.add_typer(dict_app, name="dict") # app.add_typer(documents_app, name="documents") app.add_typer(duplicatehost_app, name="duplicatehost") app.add_typer(elevateword_app, name="elevateword") diff --git a/src/fessctl/commands/dict.py b/src/fessctl/commands/dict.py new file mode 100644 index 0000000..6b64a6d --- /dev/null +++ b/src/fessctl/commands/dict.py @@ -0,0 +1,329 @@ +import inspect +import json +from pathlib import Path +from typing import List, NamedTuple, Optional + +import typer +import yaml + +from fessctl.api.client import FessAPIClient, FessAPIClientError +from fessctl.config.settings import Settings +from fessctl.utils import ( + format_detail_markdown, + format_list_markdown, + format_result_markdown, +) + +dict_app = typer.Typer() + + +class Field(NamedTuple): + api: str # field name in the API + option: str # CLI option name (without the leading dashes) + help: str + multi: bool = False # one value per line in the API; the option is repeatable + + +# The entry fields of each dictionary type, as the admin API defines them. The +# type names are the `type` values of `dict list` and the URL path segments. +DICT_FIELDS = { + "synonym": [ + Field("inputs", "source", "Source word; repeat for several (no commas or '=>')", True), + Field("outputs", "target", "Target word; repeat for several (no commas or '=>')", True), + ], + "mapping": [ + Field("inputs", "source", "Source character or string; repeat for several", True), + Field("output", "target", "Replacement string"), + ], + "kuromoji": [ + Field("token", "token", "Token"), + Field("segmentation", "segmentation", "Segmentation, words separated by spaces"), + Field("reading", "reading", "Reading, one per segment separated by spaces"), + Field("pos", "pos", "Part-of-speech"), + ], + "stopwords": [Field("input", "word", "Stop word")], + "protwords": [Field("input", "word", "Protected word")], + "stemmeroverride": [ + Field("input", "source", "Source word"), + Field("output", "target", "Stemmed form"), + ], +} + +OUTPUT_HELP = "Output format (text, json, yaml)" + + +def _status(result: dict) -> int: + return result.get("response", {}).get("status", 1) + + +def _message(result: dict) -> str: + return result.get("response", {}).get("message", "") + + +def _emit(result: dict, output: str, ok_message: str, fail_message: str, + dict_type: str, action: str, resource_id: str = "") -> None: + """Prints the outcome of a write command; exits non-zero when the API failed.""" + status = _status(result) + if output == "json": + typer.echo(json.dumps(result, indent=2)) + elif output == "yaml": + typer.echo(yaml.dump(result)) + elif status == 0: + typer.echo(format_result_markdown(True, ok_message, f"Dict.{dict_type}", action, resource_id)) + else: + typer.echo(format_result_markdown( + False, f"{fail_message} {_message(result)} Status code: {status}", f"Dict.{dict_type}", action)) + if status != 0: + raise typer.Exit(code=status) + + +def _param(name: str, annotation, default) -> inspect.Parameter: + return inspect.Parameter( + name, inspect.Parameter.KEYWORD_ONLY, default=default, annotation=annotation) + + +def _output_param() -> inspect.Parameter: + return _param("output", str, typer.Option("text", "--output", "-o", help=OUTPUT_HELP)) + + +def _dict_id_param() -> inspect.Parameter: + return _param("dict_id", str, typer.Argument(..., help="Dictionary ID (see `dict list`)")) + + +def _item_id_param() -> inspect.Parameter: + return _param("item_id", int, typer.Argument(..., help="Entry ID")) + + +def _field_param(field: Field, required: bool) -> inspect.Parameter: + annotation = List[str] if field.multi else str + if not required: + annotation = Optional[annotation] + return _param(field.option.replace("-", "_"), annotation, + typer.Option(... if required else None, f"--{field.option}", help=field.help)) + + +def _command(app: typer.Typer, name: str, func, params: list, doc: str) -> None: + """Registers func as a command whose Typer options are given by params. + + The options of create and update differ per dictionary type, so their + signatures are built from DICT_FIELDS instead of being written six times. + """ + func.__signature__ = inspect.Signature(params) + func.__annotations__ = {p.name: p.annotation for p in params} + func.__doc__ = doc + app.command(name)(func) + + +def _api_value(field: Field, value): + return "\n".join(value) if field.multi else value + + +def _register_type(dict_type: str) -> None: + fields = DICT_FIELDS[dict_type] + columns = [("ID", "id")] + [(f.api.upper(), f.api) for f in fields] + type_app = typer.Typer() + + def create(**kwargs): + for f in fields: + # Required options are not always enforced by Click, and Fess accepts a + # mapping without a target (reporting success but storing nothing). + if not kwargs[f.option.replace("-", "_")]: + raise typer.BadParameter("This option is required.", param_hint=f"--{f.option}") + client = FessAPIClient(Settings()) + config = {f.api: _api_value(f, kwargs[f.option.replace("-", "_")]) for f in fields} + result = client.create_dict_item(dict_type, kwargs["dict_id"], config) + # Fess before 15.9 reports the ID of a new entry as "0"; `list` shows the real one. + item_id = result.get("response", {}).get("id", "") + if item_id == "0": + item_id = "" + _emit(result, kwargs["output"], + f"{dict_type} entry {f'{item_id!r} ' if item_id else ''}created successfully.", + f"Failed to create {dict_type} entry.", dict_type, "create", item_id) + + _command(type_app, "create", create, + [_dict_id_param()] + [_field_param(f, True) for f in fields] + [_output_param()], + f"Create an entry in a {dict_type} dictionary.") + + def update(**kwargs): + client = FessAPIClient(Settings()) + dict_id, item_id = kwargs["dict_id"], kwargs["item_id"] + current = client.get_dict_item(dict_type, dict_id, item_id) + if _status(current) != 0: + typer.echo(format_result_markdown( + False, f"{dict_type} entry '{item_id}' not found. {_message(current)}", + f"Dict.{dict_type}", "update")) + raise typer.Exit(code=1) + + # The PUT replaces the whole entry, so start from the current one and + # override only the given fields. + config = current["response"]["setting"] + for f in fields: + value = kwargs[f.option.replace("-", "_")] + if value is not None: + config[f.api] = _api_value(f, value) + result = client.update_dict_item(dict_type, dict_id, config) + _emit(result, kwargs["output"], f"{dict_type} entry '{item_id}' updated successfully.", + f"Failed to update {dict_type} entry.", dict_type, "update", str(item_id)) + + _command(type_app, "update", update, + [_dict_id_param(), _item_id_param()] + [_field_param(f, False) for f in fields] + [_output_param()], + f"Update an entry of a {dict_type} dictionary; unspecified fields are kept.") + + def delete(**kwargs): + client = FessAPIClient(Settings()) + result = client.delete_dict_item(dict_type, kwargs["dict_id"], kwargs["item_id"]) + _emit(result, kwargs["output"], f"{dict_type} entry '{kwargs['item_id']}' deleted successfully.", + f"Failed to delete {dict_type} entry.", dict_type, "delete", str(kwargs["item_id"])) + + _command(type_app, "delete", delete, [_dict_id_param(), _item_id_param(), _output_param()], + f"Delete an entry of a {dict_type} dictionary by ID.") + + def get(**kwargs): + client = FessAPIClient(Settings()) + result = client.get_dict_item(dict_type, kwargs["dict_id"], kwargs["item_id"]) + status = _status(result) + output = kwargs["output"] + if output == "json": + typer.echo(json.dumps(result, indent=2)) + elif output == "yaml": + typer.echo(yaml.dump(result)) + elif status == 0: + setting = result.get("response", {}).get("setting", {}) + typer.echo(format_detail_markdown( + f"{dict_type} entry: {setting.get('id', '-')}", setting, [("id", "id")] + [(f.api, f.api) for f in fields])) + else: + typer.echo(format_result_markdown( + False, f"Failed to retrieve {dict_type} entry. {_message(result)} Status code: {status}", + f"Dict.{dict_type}", "get")) + if status != 0: + raise typer.Exit(code=status) + + _command(type_app, "get", get, [_dict_id_param(), _item_id_param(), _output_param()], + f"Retrieve an entry of a {dict_type} dictionary by ID.") + + def list_items(**kwargs): + client = FessAPIClient(Settings()) + result = client.list_dict_items(dict_type, kwargs["dict_id"], page=kwargs["page"], size=kwargs["size"]) + status = _status(result) + output = kwargs["output"] + if output == "json": + typer.echo(json.dumps(result, indent=2)) + elif output == "yaml": + typer.echo(yaml.dump(result)) + elif status == 0: + items = result.get("response", {}).get("settings", []) + if not items: + typer.echo(f"No {dict_type} entries found.") + else: + typer.echo(format_list_markdown(f"{dict_type} entries", items, columns)) + else: + typer.echo(format_result_markdown( + False, f"Failed to list {dict_type} entries. {_message(result)} Status code: {status}", + f"Dict.{dict_type}", "list")) + if status != 0: + raise typer.Exit(code=status) + + _command(type_app, "list", list_items, + [_dict_id_param(), + _param("page", int, typer.Option(1, "--page", "-p", help="Page number")), + _param("size", int, typer.Option(100, "--size", "-s", help="Page size")), + _output_param()], + f"List the entries of a {dict_type} dictionary. An unknown dictionary ID lists nothing.") + + def upload(**kwargs): + file: Path = kwargs["file"] + output = kwargs["output"] + if not file.is_file(): + typer.echo(format_result_markdown( + False, f"File '{file}' not found.", f"Dict.{dict_type}", "upload")) + raise typer.Exit(code=1) + if not kwargs["yes"]: + typer.confirm( + f"Replace the whole dictionary '{kwargs['dict_id']}' with '{file}'?", abort=True) + client = FessAPIClient(Settings()) + result = client.upload_dict(dict_type, kwargs["dict_id"], str(file)) + _emit(result, output, f"{dict_type} dictionary '{kwargs['dict_id']}' replaced successfully.", + f"Failed to upload {dict_type} dictionary.", dict_type, "upload", kwargs["dict_id"]) + + _command(type_app, "upload", upload, + [_dict_id_param(), + _param("file", Path, typer.Argument(..., help="Local file that replaces the dictionary")), + _param("yes", bool, typer.Option(False, "--yes", "-y", help="Skip the confirmation prompt")), + _output_param()], + f"Replace a whole {dict_type} dictionary file with a local file " + "(requires Fess 14.x or 15.9+; Fess 15.6 to 15.8 cannot receive uploads).") + + def download(**kwargs): + file: Optional[Path] = kwargs["file"] + client = FessAPIClient(Settings()) + try: + content = client.download_dict(dict_type, kwargs["dict_id"]) + except FessAPIClientError as e: + _emit({"response": {"status": 1, "message": e.content}}, kwargs["output"], "", + f"Failed to download {dict_type} dictionary.", dict_type, "download") + return + if file is None: + typer.echo(content, nl=False) + return + if file.exists() and not kwargs["force"]: + typer.echo(format_result_markdown( + False, f"File '{file}' already exists. Use --force to overwrite it.", + f"Dict.{dict_type}", "download")) + raise typer.Exit(code=1) + file.write_bytes(content) + _emit({"response": {"status": 0}}, kwargs["output"], + f"{dict_type} dictionary '{kwargs['dict_id']}' saved to '{file}' ({len(content)} bytes).", + "", dict_type, "download", kwargs["dict_id"]) + + _command(type_app, "download", download, + [_dict_id_param(), + _param("file", Optional[Path], typer.Option( + None, "--file", "-f", help="Write to this file instead of standard output")), + _param("force", bool, typer.Option(False, "--force", help="Overwrite --file if it exists")), + _output_param()], + f"Download a {dict_type} dictionary file to standard output or to --file.") + + dict_app.add_typer(type_app, name=dict_type, + help=f"Manage {dict_type} dictionaries: " + ", ".join(f"--{f.option}" for f in fields) + ".") + + +@dict_app.command("list") +def list_dicts( + dict_type: Optional[str] = typer.Option( + None, "--type", "-t", help="Only this type (synonym, mapping, kuromoji, stopwords, protwords, stemmeroverride)"), + output: str = typer.Option( + "text", "--output", "-o", help=OUTPUT_HELP), +): + """ + List the dictionary files; use their ID with the per-type subcommands. + """ + client = FessAPIClient(Settings()) + result = client.get_dicts() + status = _status(result) + response = result.get("response", {}) + if dict_type is not None and status == 0: + response["settings"] = [d for d in response.get("settings", []) if d.get("type") == dict_type] + if "total" in response: + response["total"] = len(response["settings"]) + + if output == "json": + typer.echo(json.dumps(result, indent=2)) + elif output == "yaml": + typer.echo(yaml.dump(result)) + elif status == 0: + dicts = response.get("settings", []) + if not dicts: + typer.echo("No Dictionaries found.") + else: + typer.echo(format_list_markdown("Dictionaries", dicts, [ + ("TYPE", "type"), ("ID", "id"), ("PATH", "path"), ("TIMESTAMP", "timestamp"), + ])) + else: + typer.echo(format_result_markdown( + False, f"Failed to list Dictionaries. {_message(result)} Status code: {status}", "Dict", "list")) + if status != 0: + raise typer.Exit(code=status) + + +for _dict_type in DICT_FIELDS: + _register_type(_dict_type) diff --git a/tests/commands/test_dict.py b/tests/commands/test_dict.py new file mode 100644 index 0000000..13c2c66 --- /dev/null +++ b/tests/commands/test_dict.py @@ -0,0 +1,251 @@ +import json +import uuid + +import pytest +from typer.testing import CliRunner + +from conftest import fess_version_at_least +from fessctl.commands.dict import dict_app + +# The dictionary file each type is exercised on (present in every supported Fess image). +DICT_PATHS = { + "synonym": "synonym.txt", + "mapping": "mapping.txt", + "kuromoji": "ja/kuromoji.txt", + "stopwords": "en/stopwords.txt", + "protwords": "en/protwords.txt", + "stemmeroverride": "en/stemmer_override.txt", +} + +# Per type: arguments for create, arguments for update, and the fields (API names) each +# is expected to leave in the entry. "{u}" is replaced with a value unique to the test. +ENTRY_CASES = { + "synonym": ( + ["--source", "{u}a", "--source", "{u}b", "--target", "{u}c"], + ["--target", "{u}d", "--target", "{u}e"], + {"inputs": "{u}a\n{u}b", "outputs": "{u}c"}, + {"inputs": "{u}a\n{u}b", "outputs": "{u}d\n{u}e"}, + ), + "mapping": ( + ["--source", "{u}a", "--source", "{u}b", "--target", "{u}c"], + ["--target", "{u}d"], + {"inputs": "{u}a\n{u}b", "output": "{u}c"}, + {"inputs": "{u}a\n{u}b", "output": "{u}d"}, + ), + "kuromoji": ( + ["--token", "{u}", "--segmentation", "{u}", "--reading", "エフ", "--pos", "カスタム名詞"], + ["--reading", "エフシー"], + {"token": "{u}", "segmentation": "{u}", "reading": "エフ", "pos": "カスタム名詞"}, + {"token": "{u}", "segmentation": "{u}", "reading": "エフシー", "pos": "カスタム名詞"}, + ), + "stopwords": ( + ["--word", "{u}"], ["--word", "{u}x"], {"input": "{u}"}, {"input": "{u}x"}, + ), + "protwords": ( + ["--word", "{u}"], ["--word", "{u}x"], {"input": "{u}"}, {"input": "{u}x"}, + ), + "stemmeroverride": ( + ["--source", "{u}a", "--target", "{u}b"], + ["--target", "{u}c"], + {"input": "{u}a", "output": "{u}b"}, + {"input": "{u}a", "output": "{u}c"}, + ), +} + + +@pytest.fixture(scope="module") +def runner(): + """ + Provides a CliRunner instance for invoking commands. + """ + return CliRunner() + + +def fill(value, u): + """Replaces "{u}" in an argument list or an expected-fields dict.""" + if isinstance(value, dict): + return {k: v.replace("{u}", u) for k, v in value.items()} + return [v.replace("{u}", u) for v in value] + + +def invoke_json(runner, args): + result = runner.invoke(dict_app, [*args, "--output", "json"]) + return result, json.loads(result.stdout) + + +@pytest.fixture(scope="module") +def dict_ids(runner, fess_service): + """ + Maps each dictionary type to the ID of its test file, found with `dict list`. + """ + result, resp = invoke_json(runner, ["list"]) + assert result.exit_code == 0, f"List failed: {result.stdout}" + paths = {(d["type"], d["path"]): d["id"] for d in resp["response"]["settings"]} + missing = [t for t, p in DICT_PATHS.items() if (t, p) not in paths] + assert not missing, f"The Fess under test has no dictionary file for {missing}" + return {t: paths[(t, p)] for t, p in DICT_PATHS.items()} + + +def find_entries(runner, dict_type, dict_id, needle): + """ + Returns the entries whose fields contain needle. Entry IDs are positions in the + file and shift when an earlier entry is deleted, so entries are always looked + up by their (unique) content. + """ + result, resp = invoke_json(runner, [dict_type, "list", dict_id, "--size", "5000"]) + assert result.exit_code == 0, f"List failed: {result.stdout}" + return [e for e in resp["response"]["settings"] if needle in json.dumps(e, ensure_ascii=False)] + + +def test_dict_list(runner, dict_ids): + result, resp = invoke_json(runner, ["list"]) + assert result.exit_code == 0, f"List failed: {result.stdout}" + assert resp["response"]["status"] == 0 + assert all({"id", "type", "path"} <= d.keys() for d in resp["response"]["settings"]) + + result, resp = invoke_json(runner, ["list", "--type", "kuromoji"]) + assert result.exit_code == 0 + assert [d["path"] for d in resp["response"]["settings"]] == ["ja/kuromoji.txt"] + + result = runner.invoke(dict_app, ["list", "--type", "synonym"]) + assert result.exit_code == 0 + assert "synonym.txt" in result.stdout + + +@pytest.mark.parametrize("dict_type", sorted(ENTRY_CASES)) +def test_entry_crud_flow(runner, dict_ids, dict_type): + """ + Tests the full Create, Read, Update, Delete flow of each dictionary type. + """ + dict_id = dict_ids[dict_type] + u = f"fctl{uuid.uuid4().hex[:8]}" + create_args, update_args, after_create, after_update = (fill(x, u) for x in ENTRY_CASES[dict_type]) + try: + # 1) Create. Fess before 15.9 answers with id "0", so find the entry by its content. + result, resp = invoke_json(runner, [dict_type, "create", dict_id, *create_args]) + assert result.exit_code == 0, f"Create failed: {result.stdout}" + assert resp["response"]["status"] == 0 + entries = find_entries(runner, dict_type, dict_id, u) + assert len(entries) == 1, f"Created entry not found in the list: {entries}" + item_id = entries[0]["id"] + if resp["response"].get("id") not in (None, "", "0"): + assert resp["response"]["id"] == str(item_id) + + # 2) Get + result, resp = invoke_json(runner, [dict_type, "get", dict_id, str(item_id)]) + assert result.exit_code == 0, f"Get failed: {result.stdout}" + setting = resp["response"]["setting"] + assert setting["id"] == item_id + assert {k: setting[k] for k in after_create} == after_create + + # 3) Update one field; the others must be kept + result, resp = invoke_json(runner, [dict_type, "update", dict_id, str(item_id), *update_args]) + assert result.exit_code == 0, f"Update failed: {result.stdout}" + assert resp["response"]["status"] == 0 + result, resp = invoke_json(runner, [dict_type, "get", dict_id, str(item_id)]) + setting = resp["response"]["setting"] + assert {k: setting[k] for k in after_update} == after_update + + # 4) The text list shows it + result = runner.invoke(dict_app, [dict_type, "list", dict_id, "--size", "5000"]) + assert result.exit_code == 0 + assert u in result.stdout + + # 5) Delete + result, resp = invoke_json(runner, [dict_type, "delete", dict_id, str(item_id)]) + assert result.exit_code == 0, f"Delete failed: {result.stdout}" + assert find_entries(runner, dict_type, dict_id, u) == [] + finally: + # Entry IDs shift on delete, so a leftover is looked up again by its content. + for entry in find_entries(runner, dict_type, dict_id, u): + runner.invoke(dict_app, [dict_type, "delete", dict_id, str(entry["id"])]) + + +def test_entry_errors(runner, dict_ids): + dict_id = dict_ids["synonym"] + + result = runner.invoke(dict_app, ["synonym", "get", dict_id, "99999999"]) + assert result.exit_code != 0 + assert "failed to retrieve synonym entry" in result.stdout.lower() + + result = runner.invoke(dict_app, ["synonym", "update", dict_id, "99999999", "--source", "fctl"]) + assert result.exit_code != 0 + assert "not found" in result.stdout.lower() + + result = runner.invoke(dict_app, ["synonym", "delete", dict_id, "99999999"]) + assert result.exit_code != 0 + assert "failed to delete synonym entry" in result.stdout.lower() + + # An unknown dictionary ID lists nothing in every version + result = runner.invoke(dict_app, ["synonym", "list", "nosuchdict"]) + assert result.exit_code == 0 + assert "No synonym entries found." in result.stdout + + # A source word may not contain a comma. The check exists since Fess 15.9; older + # versions fail on it with a server error, so only the failure is common. + u = f"fctl{uuid.uuid4().hex[:8]}" + try: + result = runner.invoke(dict_app, ["synonym", "create", dict_id, "--source", f"{u},x", "--target", "y"]) + assert result.exit_code != 0 + finally: + for entry in find_entries(runner, "synonym", dict_id, u): + runner.invoke(dict_app, ["synonym", "delete", dict_id, str(entry["id"])]) + + # Fess before 15.9 reports success when creating in an unknown dictionary + if fess_version_at_least(15, 9): + result = runner.invoke(dict_app, ["synonym", "create", "nosuchdict", "--source", "fctl", "--target", "y"]) + assert result.exit_code != 0 + + +def test_download_error(runner, dict_ids): + result = runner.invoke(dict_app, ["synonym", "download", "nosuchdict"]) + assert result.exit_code != 0 + assert "failed to download synonym dictionary" in result.stdout.lower() + + +def test_download_and_upload_flow(runner, dict_ids, tmp_path): + """ + Downloads a dictionary file, replaces it with a modified copy, and restores it. + Fess 15.6 to 15.8 cannot receive uploads at all (the server cannot read a multipart PUT). + """ + dict_id = dict_ids["protwords"] + original = tmp_path / "original.txt" + result = runner.invoke(dict_app, ["protwords", "download", dict_id, "--file", str(original)]) + assert result.exit_code == 0, f"Download failed: {result.stdout}" + + # An existing file is not overwritten without --force + result = runner.invoke(dict_app, ["protwords", "download", dict_id, "--file", str(original)]) + assert result.exit_code != 0 + assert "already exists" in result.stdout + + result = runner.invoke(dict_app, ["protwords", "download", dict_id]) + assert result.exit_code == 0 + assert result.stdout_bytes == original.read_bytes() + + u = f"fctl{uuid.uuid4().hex[:8]}" + modified = tmp_path / "modified.txt" + modified.write_bytes(original.read_bytes() + f"{u}\n".encode()) + + if fess_version_at_least(15, 6) and not fess_version_at_least(15, 9): + result = runner.invoke(dict_app, ["protwords", "upload", dict_id, str(modified), "--yes"]) + assert result.exit_code != 0 + assert "not supported" in result.stdout + assert runner.invoke(dict_app, ["protwords", "download", dict_id]).stdout_bytes == original.read_bytes() + return + + try: + result, resp = invoke_json(runner, ["protwords", "upload", dict_id, str(modified), "--yes"]) + assert result.exit_code == 0, f"Upload failed: {result.stdout}" + assert resp["response"]["status"] == 0 + downloaded = runner.invoke(dict_app, ["protwords", "download", dict_id]) + assert downloaded.stdout_bytes == modified.read_bytes() + assert len(find_entries(runner, "protwords", dict_id, u)) == 1 + finally: + result = runner.invoke(dict_app, ["protwords", "upload", dict_id, str(original), "--yes"]) + assert result.exit_code == 0, f"Restoring the dictionary failed: {result.stdout}" + assert runner.invoke(dict_app, ["protwords", "download", dict_id]).stdout_bytes == original.read_bytes() + assert find_entries(runner, "protwords", dict_id, u) == [] + + # Fess rejects an upload to an unknown dictionary + result = runner.invoke(dict_app, ["protwords", "upload", "nosuchdict", str(original), "--yes"]) + assert result.exit_code != 0 diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 416d359..a79b1a5 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,6 +1,7 @@ """ Unit tests for fessctl.api.client module. """ +import inspect import json from types import SimpleNamespace from unittest.mock import Mock, patch, MagicMock @@ -315,7 +316,11 @@ def test_get_carries_no_body(self, client, http): @pytest.mark.parametrize("method", LIST_METHODS) def test_every_list_method_puts_paging_in_body(self, client, http, method): - getattr(client, method)(page=2, size=1) + list_method = getattr(client, method) + # Some list methods also take a resource selector (e.g. a dictionary ID). + required = [p for p in inspect.signature(list_method).parameters.values() + if p.default is p.empty] + list_method(*["x"] * len(required), page=2, size=1) http.get.assert_not_called() call_kwargs = http.request.call_args[1] diff --git a/tests/unit/test_dict.py b/tests/unit/test_dict.py new file mode 100644 index 0000000..d529cda --- /dev/null +++ b/tests/unit/test_dict.py @@ -0,0 +1,340 @@ +""" +Unit tests for the dict command and its client methods. +""" +from unittest.mock import Mock, patch + +import httpx +import pytest +from typer.testing import CliRunner + +from fessctl.api.client import FessAPIClient, FessAPIClientError +from fessctl.commands.dict import DICT_FIELDS, dict_app +from fessctl.config.settings import Settings + +DICT_TYPES = sorted(DICT_FIELDS) + + +def make_client(version: str) -> FessAPIClient: + settings = Mock(spec=Settings) + settings.fess_endpoint = "http://localhost:8080" + settings.access_token = "test-token" + settings.fess_version = version + return FessAPIClient(settings) + + +def ok_response(body=None): + response = Mock() + response.json.return_value = body or {"response": {"status": 0}} + return response + + +@pytest.fixture +def runner(): + return CliRunner() + + +class TestClient: + def test_get_dicts(self): + with patch("httpx.get", return_value=ok_response()) as get: + make_client("15.9.0").get_dicts() + assert get.call_args[0][0] == "http://localhost:8080/api/admin/dict" + + def test_list_dict_items_sends_paging_in_body_and_query(self): + with patch("httpx.request", return_value=ok_response()) as request: + make_client("15.9.0").list_dict_items("synonym", "ID1", page=2, size=1) + assert request.call_args[0] == ("GET", "http://localhost:8080/api/admin/dict/synonym/settings/ID1") + assert request.call_args[1]["json"] == {"page": 2, "size": 1} + assert request.call_args[1]["params"] == {"page": 2, "size": 1} + + def test_get_dict_item(self): + with patch("httpx.get", return_value=ok_response()) as get: + make_client("15.9.0").get_dict_item("kuromoji", "ID1", 7) + assert get.call_args[0][0] == "http://localhost:8080/api/admin/dict/kuromoji/setting/ID1/7" + + def test_delete_dict_item(self): + with patch("httpx.delete", return_value=ok_response()) as delete: + make_client("15.9.0").delete_dict_item("stopwords", "ID1", 7) + assert delete.call_args[0][0] == "http://localhost:8080/api/admin/dict/stopwords/setting/ID1/7" + + @pytest.mark.parametrize("version,create,update", [ + ("15.9.0", "post", "put"), + ("15.6.1", "post", "put"), + ("14.19.2", "put", "post"), + ]) + def test_create_and_update_methods_follow_the_fess_version(self, version, create, update): + client = make_client(version) + with patch(f"httpx.{create}", return_value=ok_response()) as created: + client.create_dict_item("protwords", "ID1", {"input": "a"}) + assert created.call_args[0][0] == "http://localhost:8080/api/admin/dict/protwords/setting/ID1" + assert created.call_args[1]["json"] == {"input": "a"} + with patch(f"httpx.{update}", return_value=ok_response()) as updated: + client.update_dict_item("protwords", "ID1", {"id": 1, "input": "b"}) + assert updated.call_args[1]["json"] == {"id": 1, "input": "b"} + + @pytest.mark.parametrize("version", ["15.9.0", "14.19.2"]) + @pytest.mark.parametrize("dict_type,field", [ + ("synonym", "synonymFile"), ("mapping", "charMappingFile"), ("kuromoji", "kuromojiFile"), + ("stopwords", "stopwordsFile"), ("protwords", "protwordsFile"), + ("stemmeroverride", "stemmerOverrideFile"), + ]) + def test_upload_posts_multipart_with_the_types_field(self, version, dict_type, field, tmp_path): + path = tmp_path / "words.txt" + path.write_bytes(b"a\nb\n") + with patch("httpx.post", return_value=ok_response()) as post: + make_client(version).upload_dict(dict_type, "ID1", str(path)) + assert post.call_args[0][0] == f"http://localhost:8080/api/admin/dict/{dict_type}/upload/ID1" + assert post.call_args[1]["files"] == {field: ("words.txt", b"a\nb\n", "application/octet-stream")} + assert "Content-Type" not in post.call_args[1]["headers"] + assert post.call_args[1]["headers"]["Authorization"] == "Bearer test-token" + + @pytest.mark.parametrize("version", ["15.6.1", "15.7.0", "15.8.0"]) + def test_upload_is_refused_where_the_server_cannot_receive_it(self, version, tmp_path): + path = tmp_path / "words.txt" + path.write_bytes(b"a\n") + with patch("httpx.post") as post, patch("httpx.put") as put: + result = make_client(version).upload_dict("synonym", "ID1", str(path)) + assert result["response"]["status"] == 1 + assert "15.9" in result["response"]["message"] + post.assert_not_called() + put.assert_not_called() + + def test_download_returns_the_content(self): + response = Mock(status_code=200, content=b"a\nb\n", headers={"content-type": "application/octet-stream"}) + with patch("httpx.get", return_value=response) as get: + content = make_client("15.9.0").download_dict("synonym", "ID1") + assert content == b"a\nb\n" + assert get.call_args[0][0] == "http://localhost:8080/api/admin/dict/synonym/download/ID1" + + def test_download_error_raises_with_the_message(self): + response = Mock(status_code=400, headers={"content-type": "application/json;charset=UTF-8"}) + response.json.return_value = {"response": {"status": 1, "message": "Failed to download."}} + with patch("httpx.get", return_value=response): + with pytest.raises(FessAPIClientError) as error: + make_client("15.9.0").download_dict("synonym", "nosuch") + assert error.value.content == "Failed to download." + + def test_download_network_error_raises(self): + with patch("httpx.get", side_effect=httpx.ConnectError("refused")): + with pytest.raises(FessAPIClientError) as error: + make_client("15.9.0").download_dict("synonym", "ID1") + assert error.value.status_code == -1 + + +def patched_client(): + """Patches FessAPIClient in the command module; returns (patcher, mock client).""" + patcher = patch("fessctl.commands.dict.FessAPIClient") + mock_client = Mock() + patcher.start().return_value = mock_client + return patcher, mock_client + + +@pytest.fixture +def client(): + patcher, mock_client = patched_client() + yield mock_client + patcher.stop() + + +class TestDictList: + def test_list_text_output_and_type_filter(self, client, runner): + client.get_dicts.return_value = {"response": {"status": 0, "settings": [ + {"id": "SY", "type": "synonym", "path": "synonym.txt", "timestamp": "2026-10-16T12:00:00"}, + {"id": "KU", "type": "kuromoji", "path": "ja/kuromoji.txt", "timestamp": "2026-10-16T12:00:01"}, + ]}} + + result = runner.invoke(dict_app, ["list"]) + assert result.exit_code == 0 + assert "| synonym | SY | synonym.txt | 2026-10-16T12:00:00 |" in result.stdout + assert "| kuromoji | KU |" in result.stdout + + result = runner.invoke(dict_app, ["list", "--type", "kuromoji"]) + assert "| kuromoji | KU |" in result.stdout + assert "synonym" not in result.stdout + + result = runner.invoke(dict_app, ["list", "--type", "mapping"]) + assert "No Dictionaries found." in result.stdout + + def test_list_failure_exits_with_status(self, client, runner): + client.get_dicts.return_value = {"response": {"status": 1, "message": "boom"}} + result = runner.invoke(dict_app, ["list"]) + assert result.exit_code == 1 + assert "boom" in result.stdout + + +# Command-line arguments that fill every field of each type, and the API body they must produce. +CREATE_CASES = { + "synonym": (["--source", "a", "--source", "b", "--target", "c"], {"inputs": "a\nb", "outputs": "c"}), + "mapping": (["--source", "a", "--source", "b", "--target", "c"], {"inputs": "a\nb", "output": "c"}), + "kuromoji": (["--token", "t", "--segmentation", "s", "--reading", "r", "--pos", "p"], + {"token": "t", "segmentation": "s", "reading": "r", "pos": "p"}), + "stopwords": (["--word", "w"], {"input": "w"}), + "protwords": (["--word", "w"], {"input": "w"}), + "stemmeroverride": (["--source", "a", "--target", "c"], {"input": "a", "output": "c"}), +} + + +class TestEntries: + def test_every_type_has_a_create_case(self): + assert sorted(CREATE_CASES) == DICT_TYPES + + @pytest.mark.parametrize("dict_type", DICT_TYPES) + def test_create_sends_the_types_fields(self, client, runner, dict_type): + args, body = CREATE_CASES[dict_type] + client.create_dict_item.return_value = {"response": {"status": 0, "id": "7"}} + + result = runner.invoke(dict_app, [dict_type, "create", "ID1", *args]) + + assert result.exit_code == 0, result.stdout + assert "created successfully" in result.stdout + client.create_dict_item.assert_called_once_with(dict_type, "ID1", body) + + def test_create_without_id_in_the_response_still_succeeds(self, client, runner): + client.create_dict_item.return_value = {"response": {"status": 0, "id": "0"}} + result = runner.invoke(dict_app, ["stopwords", "create", "ID1", "--word", "w"]) + assert result.exit_code == 0 + assert "stopwords entry created successfully." in result.stdout + assert "'0'" not in result.stdout + + def test_create_requires_every_field(self, client, runner): + result = runner.invoke(dict_app, ["mapping", "create", "ID1", "--source", "a"]) + assert result.exit_code == 2 + client.create_dict_item.assert_not_called() + + def test_create_failure_exits_with_status(self, client, runner): + client.create_dict_item.return_value = {"response": {"status": 1, "message": "is invalid"}} + result = runner.invoke(dict_app, ["synonym", "create", "ID1", "--source", "a,b", "--target", "c"]) + assert result.exit_code == 1 + assert "is invalid" in result.stdout + + def test_update_keeps_unspecified_fields(self, client, runner): + client.get_dict_item.return_value = {"response": {"status": 0, "setting": { + "id": 3, "dict_id": "ID1", "inputs": "a\nb", "outputs": "c"}}} + client.update_dict_item.return_value = {"response": {"status": 0, "id": "3"}} + + result = runner.invoke(dict_app, ["synonym", "update", "ID1", "3", "--target", "x", "--target", "y"]) + + assert result.exit_code == 0, result.stdout + client.get_dict_item.assert_called_once_with("synonym", "ID1", 3) + client.update_dict_item.assert_called_once_with( + "synonym", "ID1", {"id": 3, "dict_id": "ID1", "inputs": "a\nb", "outputs": "x\ny"}) + + def test_update_of_a_missing_entry_fails_without_writing(self, client, runner): + client.get_dict_item.return_value = {"response": {"status": 1, "message": "gone"}} + result = runner.invoke(dict_app, ["kuromoji", "update", "ID1", "3", "--pos", "p"]) + assert result.exit_code == 1 + assert "not found" in result.stdout + client.update_dict_item.assert_not_called() + + def test_get_text_output(self, client, runner): + client.get_dict_item.return_value = {"response": {"status": 0, "setting": { + "id": 3, "dict_id": "ID1", "token": "t", "segmentation": "s", "reading": "r", "pos": "p"}}} + result = runner.invoke(dict_app, ["kuromoji", "get", "ID1", "3"]) + assert result.exit_code == 0 + for row in ("| id | 3 |", "| token | t |", "| segmentation | s |", "| reading | r |", "| pos | p |"): + assert row in result.stdout + + def test_get_failure_exits_with_status(self, client, runner): + client.get_dict_item.return_value = {"response": {"status": 1, "message": "gone"}} + result = runner.invoke(dict_app, ["synonym", "get", "ID1", "9"]) + assert result.exit_code == 1 + assert "Failed to retrieve synonym entry" in result.stdout + + def test_delete(self, client, runner): + client.delete_dict_item.return_value = {"response": {"status": 0}} + result = runner.invoke(dict_app, ["protwords", "delete", "ID1", "3"]) + assert result.exit_code == 0 + client.delete_dict_item.assert_called_once_with("protwords", "ID1", 3) + + def test_item_id_must_be_a_number(self, client, runner): + result = runner.invoke(dict_app, ["protwords", "delete", "ID1", "abc"]) + assert result.exit_code == 2 + client.delete_dict_item.assert_not_called() + + def test_list_passes_paging_and_shows_the_fields(self, client, runner): + client.list_dict_items.return_value = {"response": {"status": 0, "settings": [ + {"id": 1, "dict_id": "ID1", "inputs": "a\nb", "outputs": "c"}]}} + result = runner.invoke(dict_app, ["synonym", "list", "ID1", "--page", "2", "--size", "5"]) + assert result.exit_code == 0 + client.list_dict_items.assert_called_once_with("synonym", "ID1", page=2, size=5) + assert "| 1 | a
b | c |" in result.stdout + + def test_list_empty(self, client, runner): + client.list_dict_items.return_value = {"response": {"status": 0, "settings": []}} + result = runner.invoke(dict_app, ["mapping", "list", "ID1"]) + assert "No mapping entries found." in result.stdout + + def test_json_output_is_the_raw_response(self, client, runner): + client.list_dict_items.return_value = {"response": {"status": 0, "settings": [], "total": 0}} + result = runner.invoke(dict_app, ["mapping", "list", "ID1", "-o", "json"]) + assert '"total": 0' in result.stdout + + +class TestUpload: + def test_upload_confirms_then_uploads(self, client, runner, tmp_path): + path = tmp_path / "words.txt" + path.write_text("a\n") + client.upload_dict.return_value = {"response": {"status": 0}} + + declined = runner.invoke(dict_app, ["stopwords", "upload", "ID1", str(path)], input="n\n") + assert declined.exit_code != 0 + client.upload_dict.assert_not_called() + + result = runner.invoke(dict_app, ["stopwords", "upload", "ID1", str(path)], input="y\n") + assert result.exit_code == 0 + client.upload_dict.assert_called_once_with("stopwords", "ID1", str(path)) + + def test_upload_yes_skips_the_prompt(self, client, runner, tmp_path): + path = tmp_path / "words.txt" + path.write_text("a\n") + client.upload_dict.return_value = {"response": {"status": 0}} + result = runner.invoke(dict_app, ["stopwords", "upload", "ID1", str(path), "--yes"]) + assert result.exit_code == 0 + assert "replaced successfully" in result.stdout + + def test_upload_missing_file(self, client, runner, tmp_path): + result = runner.invoke(dict_app, ["stopwords", "upload", "ID1", str(tmp_path / "nope.txt"), "-y"]) + assert result.exit_code == 1 + assert "not found" in result.stdout + client.upload_dict.assert_not_called() + + def test_upload_failure_exits_with_status(self, client, runner, tmp_path): + path = tmp_path / "words.txt" + path.write_text("a\n") + client.upload_dict.return_value = {"response": {"status": 1, "message": "no upload here"}} + result = runner.invoke(dict_app, ["stopwords", "upload", "ID1", str(path), "-y"]) + assert result.exit_code == 1 + assert "no upload here" in result.stdout + + +class TestDownload: + def test_download_to_standard_output(self, client, runner): + client.download_dict.return_value = b"a\nb\n" + result = runner.invoke(dict_app, ["synonym", "download", "ID1"]) + assert result.exit_code == 0 + assert result.stdout_bytes == b"a\nb\n" + + def test_download_to_file_refuses_to_overwrite(self, client, runner, tmp_path): + client.download_dict.return_value = b"new" + path = tmp_path / "out.txt" + + result = runner.invoke(dict_app, ["synonym", "download", "ID1", "--file", str(path)]) + assert result.exit_code == 0 + assert path.read_bytes() == b"new" + + client.download_dict.return_value = b"newer" + result = runner.invoke(dict_app, ["synonym", "download", "ID1", "--file", str(path)]) + assert result.exit_code == 1 + assert "already exists" in result.stdout + assert path.read_bytes() == b"new" + + result = runner.invoke(dict_app, ["synonym", "download", "ID1", "--file", str(path), "--force"]) + assert result.exit_code == 0 + assert path.read_bytes() == b"newer" + + def test_download_failure_exits_non_zero(self, client, runner, tmp_path): + client.download_dict.side_effect = FessAPIClientError(400, "Failed to download a protwords file.") + path = tmp_path / "out.txt" + result = runner.invoke(dict_app, ["synonym", "download", "nosuch", "--file", str(path)]) + assert result.exit_code == 1 + assert "Failed to download a protwords file." in result.stdout + assert not path.exists()