From aeb220ba00267bbe9c864c6e7126a49d203a9942 Mon Sep 17 00:00:00 2001 From: Shinsuke Sugaya Date: Fri, 16 Oct 2026 21:45:21 +0900 Subject: [PATCH] feat: add the dict command for the dictionary admin API Add fessctl dict list (the dictionary files) and, for each of synonym, mapping, kuromoji, stopwords, protwords and stemmeroverride, the subcommands list, get, create, update, delete, upload and download. The six types share one implementation; each keeps the entry fields of its admin API form. update reads the entry first and sends back the unchanged fields. upload asks for confirmation unless --yes; download writes to standard output or to --file (not overwritten without --force). The API exists in 14.19.2, 15.6.1, 15.7.0, 15.8.0 and 15.9.0. Uploads are POST multipart in 14.x and 15.9; Fess 15.6 to 15.8 only route a PUT that cannot read a multipart body, so upload is refused there. list_dict_items takes a dictionary type and ID, so the shared client paging test now supplies placeholders for required parameters. --- skills/fessctl/SKILL.md | 3 +- skills/fessctl/references/features/dict.md | 66 ++++ src/fessctl/api/client.py | 111 +++++++ src/fessctl/cli.py | 2 + src/fessctl/commands/dict.py | 329 ++++++++++++++++++++ tests/commands/test_dict.py | 251 +++++++++++++++ tests/unit/test_client.py | 7 +- tests/unit/test_dict.py | 340 +++++++++++++++++++++ 8 files changed, 1107 insertions(+), 2 deletions(-) create mode 100644 skills/fessctl/references/features/dict.md create mode 100644 src/fessctl/commands/dict.py create mode 100644 tests/commands/test_dict.py create mode 100644 tests/unit/test_dict.py diff --git a/skills/fessctl/SKILL.md b/skills/fessctl/SKILL.md index a24fc0b..02b8a9e 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 25 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 26 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 0d837c3..ce4193b 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 typing import Dict, List, Optional @@ -15,6 +16,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): @@ -67,6 +84,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: @@ -102,6 +120,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( @@ -145,6 +171,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 9fbe1df..f6b6a2e 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()