diff --git a/lang/qet_en.ts b/lang/qet_en.ts index 5b308acbb..e76b18c19 100644 --- a/lang/qet_en.ts +++ b/lang/qet_en.ts @@ -865,7 +865,7 @@ Note: these options DO NOT allow or block auto numberings, only their update pol Dimensions du folio - Dimensions of folio + Folio size @@ -996,8 +996,8 @@ Note: these options DO NOT allow or block auto numberings, only their update pol Modifier la couleur de %n conducteur(s) undo caption - - + Change the colour of %n conductor + Change the colour of %n conductors @@ -3045,12 +3045,12 @@ The element's display name is edited separately in the element properties.< Renvoi de folio suivant - Reference folio following + Next folio reference Renvoi de folio précédent - Previous reference folio + Previous folio reference @@ -3281,7 +3281,7 @@ The element's display name is edited separately in the element properties.< Réf. - Réf. + Ref. @@ -3629,7 +3629,7 @@ The element's display name is edited separately in the element properties.< Titre du folio - Title of folio + Folio title @@ -4040,16 +4040,16 @@ By importing this file, you confirm that: %n élément(s), répartie(s) - %n element, part - %n elements, parts + %n element, spread + %n elements, spread dans %n dossier(s). - in %n folder. - in %n folders. + across %n folder. + across %n folders. @@ -4215,7 +4215,7 @@ By importing this file, you confirm that: Remonter ce folio - Move up this folio + Move this folio up @@ -4225,17 +4225,17 @@ By importing this file, you confirm that: Remonter ce folio x10 - Move up this folio x10 + Move this folio up x10 Remonter ce folio x100 - Move up this folio x100 + Move this folio up x100 Remonter ce folio au debut - Move up this folio to the beginning + Move this folio to the beginning @@ -4381,7 +4381,7 @@ By importing this file, you confirm that: Titre du folio - Title of folio + Folio title @@ -4477,7 +4477,7 @@ By importing this file, you confirm that: Exporter entièrement le folio - Export fully folio + Export the whole folio @@ -4770,7 +4770,7 @@ that you create. Text and number inputs are Référence croisé - Cross reference + Cross-reference @@ -5609,8 +5609,8 @@ Any setting other than “No rounding” may cause rendering errors in the proje <center>ATTENTION :</center> il manque %1 lignes afin d'afficher l'intégralité des informations - <center>ATTENTION :</center> - lines are missing %1 to display all the informations + <center>WARNING:</center> + %1 more rows are needed to display all the information @@ -6195,13 +6195,13 @@ Please use the advanced editor for this. N° de folio - N° of folio + Folio no. Label de folio - Label folio + Folio label @@ -6209,7 +6209,7 @@ Please use the advanced editor for this. Titre de folio - Title of folio + Folio title @@ -6297,12 +6297,12 @@ Please use the advanced editor for this. Report de folio - Folio referencing + Folio reference Référence croisée (esclave) - Cross Reference (slave) + Cross-reference (slave) @@ -6364,18 +6364,18 @@ Please use the advanced editor for this. N° de folio - N° of folio + Folio no. Label de folio - Label folio + Folio label Titre de folio - Title of folio + Folio title @@ -6583,12 +6583,12 @@ Do you still want to link this slave contact? Reports de folio - Folio referencings + Folio references Références croisées - Cross References + Cross-references @@ -6816,7 +6816,7 @@ Do you still want to link this slave contact? N° folio - N° scheme + Folio no. @@ -7216,7 +7216,7 @@ Voltage / Protocol : %1 Numéro : %1 -Numéro : %1 +Number: %1 @@ -7265,7 +7265,7 @@ Conductor section : %1 Veuillez saisir une formule compatible pour ce potentiel. Les variables suivantes sont incompatibles : %sequf_ %seqtf_ %seqhf_ %id %F %M %LM - The new potential formula contains variables incompatible with the folio reports. + The new potential formula contains variables incompatible with folio references. Please enter a compatible formula for this potential. The following variables are incompatible: %sequf_ %seqtf_ %seqhf_ %id %F %M %LM @@ -8562,7 +8562,7 @@ Available options: Édite les propriétés du folio (dimensions, informations du cartouche, propriétés des conducteurs...) status bar tip - Edits the properties of the folio (size, title block informations, conductor properties...) + Edits the properties of the folio (size, title block information, conductor properties...) @@ -9172,12 +9172,12 @@ Hold Ctrl while moving to place freely. Ajoute une courbe de Bézier sur le folio actuel - Adds a Bézier curve on the current sheet + Adds a Bézier curve to the current folio Ajoute un plan de bornier sur le folio actuel - Add a terminal plan on the current sheet + Adds a terminal plan to the current folio @@ -9273,8 +9273,8 @@ Unbridge and/or remove the levels from the affected terminals so that they can b %n objet(s) remis sur la grille - %n item placed on the grid - %n item placed on the grid + %n item put back on the grid + %n items put back on the grid @@ -9387,8 +9387,8 @@ Unbridge and/or remove the levels from the affected terminals so that they can b %n élément(s) redessiné(s). - %n element redesigned. - %n elements redesigned. + %n element redrawn. + %n elements redrawn. @@ -9403,8 +9403,8 @@ Unbridge and/or remove the levels from the affected terminals so that they can b %n élément(s) non redessiné(s) : leur taille, leur point de saisie ou leurs bornes ont changé (borne ajoutée, supprimée ou déplacée). - %n element not redrawn: their size, grip or boundaries have changed (a boundary has been added, removed or moved). - %n elements not redrawn: their size, grip or boundaries have changed (a boundary has been added, removed or moved). + %n element not redrawn: its size, grip point or terminals have changed (a terminal was added, removed or moved). + %n elements not redrawn: their size, grip point or terminals have changed (a terminal was added, removed or moved). @@ -9858,7 +9858,7 @@ Enable scripts? This setting can be changed in Configure QElectroTech > Gener <br><b>Erreur</b> :<br>Les reports de folio doivent posséder une seul borne.<br><b>Solution</b> :<br>Verifier que l'élément ne possède qu'une seul borne - <br> <b> Error </ b>: <br> folio referencings must have a single terminal <br> <b> Solution </ b> :<br> Check that the element has only one terminal + <br><b>Error</b>:<br>Folio references must have a single terminal.<br><b>Solution</b>:<br>Check that the element has only one terminal @@ -9871,8 +9871,8 @@ Enable scripts? This setting can be changed in Configure QElectroTech > Gener %n erreur(s) errors - %n erreur - %n erreurs + %n error + %n errors @@ -11138,8 +11138,8 @@ What do you wish to do ? %n forme(s) Sentence fragment used in an automatically generated list of different objects, e.g. objects moved at the same time, which will be combined into a sentence. - %n forme - %n formes + %n shape + %n shapes @@ -11252,7 +11252,7 @@ What do you wish to do ? Folio sans titre - Folio Untitled + Untitled folio @@ -11645,8 +11645,8 @@ the translated name of this folder could not be read, so its folder name is disp Ajouter %n conducteur(s) add a numbers of conductor one or more - add %n conductor - add %n conductors + Add %n conductor + Add %n conductors @@ -13849,7 +13849,7 @@ E.g. associating the name "volta" with the value "1745" will Label de report de folio - Label of folio referencing + Folio reference label @@ -13861,14 +13861,14 @@ Créer votre propre texte en vous aidant des variables suivantes : %LM : la localisation %l : le numéro de ligne %c : le numéro de colonne - You can define a custom label for folio reports. -Create your own text by using the following variables: -% f: the folio position in the project -% F: the folio number -% M: the installation -% LM: the location -% l: the line number -% c: the column number + You can define a custom label for folio references. +Create your own text using the following variables: +%f: the folio position in the project +%F: the folio number +%M: the installation +%LM: the location +%l: the row number +%c: the column number @@ -14031,7 +14031,7 @@ Create your own text by using the following variables: Eléments report de folio - Folio referencings elements + Folio reference elements @@ -17384,7 +17384,7 @@ The other fields are not used. Référence croisé - Cross reference + Cross-reference @@ -17673,7 +17673,7 @@ The other fields are not used. Informations des cartouches - Title block informations + Title block information @@ -18547,13 +18547,13 @@ associer le nom "volta" et la valeur "1745" remplacera %{vol %c : le numéro de colonne %M: Installation %LM: Localisation - Create your own text by helping you of the following variables : -%f : the folio number -% F: folio label -% l : the line number -% c : column number -% M: Plant -% LM: Location + Create your own text using the following variables: +%f: the folio number +%F: the folio label +%l: the row number +%c: the column number +%M: the installation +%LM: the location diff --git a/misc/qet-mcp/README.md b/misc/qet-mcp/README.md index ed0613021..fd3207026 100644 --- a/misc/qet-mcp/README.md +++ b/misc/qet-mcp/README.md @@ -69,6 +69,7 @@ Register it with an MCP client, for example: "args": ["/path/to/qelectrotech/misc/qet-mcp/qet_mcp.py"], "env": { "QET_MCP_WORKSPACE": "/home/you/drawings", + "QET_BINARY": "/usr/bin/qelectrotech", "QET_ENABLE_SCRIPTING": "1" } } @@ -76,6 +77,11 @@ Register it with an MCP client, for example: } ``` +`QET_BINARY` is the QElectroTech the tools launch. Leave it out when +`qelectrotech` is on your `PATH`, or when the server is installed with +QElectroTech (as `/share/qelectrotech/mcp/qet_mcp.py`, which also +finds the installed element collection). + ## Using it from the Claude app A web chat in a browser cannot start a program on your computer, so it @@ -106,9 +112,9 @@ and it uses the same account as the website. Windows path is written `\\`. 4. Quit the app completely and start it again. The tools appear under the chat box's tools menu. -5. In the chat, say where your `qelectrotech` executable is. `qet_edit` and - `qet_export` take it as an argument on every call; the other tools do not - need it. +5. If `qelectrotech` is not on your `PATH`, add `"QET_BINARY"` to the + `env` block with the full path to the executable + (`C:\\Program Files\\...\\qelectrotech.exe`, written with `\\`). Only files under `QET_MCP_WORKSPACE` can be read or written (see [What the server is allowed to touch](#what-the-server-is-allowed-to-touch)). @@ -179,11 +185,22 @@ Set the workspace to the folder your drawings live in. A path outside it is refused with an error naming what was allowed; symlinks are resolved first, so a link planted inside the workspace is judged by where it points. -Two arguments are deliberately **not** confined: `binary` (the -`qelectrotech` executable) and `elements_dir` (the element collection). -Those are configuration, chosen once by whoever runs the server, and both -normally live in `/usr` or a build tree — outside any sensible workspace. -Confining them would reject the ordinary case while stopping nothing. +**The client does not choose what program runs.** The tools that launch +QElectroTech use the one the server found (`QET_BINARY`, the install it +ships in, or `PATH`). A call may still name `binary`, but only as that same +file or one listed by whoever configured the server: + +| | | +|---|---| +| `QET_BINARY` | the QElectroTech to launch | +| `QET_MCP_BINARIES` | other executables a call may name, separated like `QET_MCP_WORKSPACE` (for comparing two builds) | +| `QET_MCP_ALLOW_ANY_BINARY=1` | turns the check off: a call can then run any program | +| `QET_MCP_ELEMENTS` | element collections a call may name as `elements_dir` besides the workspace and the installed one | + +Anything else is refused, even a file inside the workspace: being there +makes it readable, not runnable. Before this rule any executable a call +named was run, with the call's own paths as arguments, so text inside a +project could steer an assistant into starting another program. `QET_MCP_ALLOW_ANY_PATH=1` is equivalent to granting the client local filesystem access with this process's privileges. It exists so that is a @@ -221,7 +238,6 @@ the answer a screenshot gave wrongly. ```json {"name": "qet_edit", "arguments": { - "binary": "/path/to/qelectrotech", "project": "in.qet", "output": "out.qet", "elements_dir": "/path/to/qelectrotech/elements", "operations": [ @@ -277,7 +293,7 @@ to emit. ```json {"name": "qet_query", "arguments": { - "binary": "/path/to/qelectrotech", "project": "industrial.qet", + "project": "industrial.qet", "sql": "SELECT label, COUNT(*) AS n FROM element_nomenclature_view WHERE label <> '' GROUP BY label HAVING n > 1 ORDER BY n DESC"}} ``` @@ -359,6 +375,10 @@ Python, plus the hang guard on `addConductor` and the database refresh in an index would shift. A folio saved without a uuid shows it empty: QElectroTech gives it one on load and writes it on the next save, so it appears after a first `qet_edit`. Needs `qet.folioIndex()` in the build. + The `"$id"` of an `add_folio` or `insert_folio` works the same way: it + keeps naming that folio after a later `insert_folio` or `remove_folio` in + the same run (on a build without `qet.folioUuid()`, it is the index the + folio had when it was made, as before). - **A conductor can be named by its uuid** (`qet_conductors` reports it): `set_conductor`, `move_conductor_segment` and `delete_conductor` take `"conductor": "{uuid}"` in place of `element` + `terminal`, which works @@ -420,7 +440,8 @@ Python, plus the hang guard on `addConductor` and the database refresh in changes nothing. `addElement` and the move/delete verbs shipped with the scripting API; `addConductor`, `rotateElement`, `setElementLabel`, `setElementInfo` and `setFolioTitle` are newer. -- **`elements_dir` is not optional for `common://` paths.** The sandboxed +- **`elements_dir` is not optional for `common://` paths** unless the + server is installed with QElectroTech, which fills it in. The sandboxed run has its own empty HOME, so QElectroTech falls back to the compiled-in collection path, which on a machine that never ran `make install` does not exist. The only symptom is `addElement` reporting that a file plainly diff --git a/misc/qet-mcp/qet_mcp.py b/misc/qet-mcp/qet_mcp.py index 62675672f..153056b66 100755 --- a/misc/qet-mcp/qet_mcp.py +++ b/misc/qet-mcp/qet_mcp.py @@ -1463,6 +1463,10 @@ SEARCH_REPLACE_KINDS = ["element_info", "conductor", "text"] FOLIO_PROPERTIES = ["title", "author", "filename", "plant", "locmach", "indexrev", "folio", "template"] +# The ops that make a folio: their "$id" is an index the next insert_folio +# or remove_folio can shift, so the script also keeps the folio's uuid. +FOLIO_MAKING_OPS = ("add_folio", "insert_folio") + # The ops that address one conductor by element + terminal, and so also # take "conductor": "{uuid}" (qet_conductors reports each one's uuid). CONDUCTOR_UUID_OPS = ("set_conductor", "move_conductor_segment", "delete_conductor") @@ -1502,6 +1506,9 @@ def _build_script(operations: list, output: str) -> str: JavaScript exception. """ refs: set[str] = set() + # "$name"s made by an op that creates a folio: held as the folio's uuid + # too, since a later insert_folio or remove_folio shifts its index. + folio_refs: set[str] = set() # Resolvers a uuid reference needs; required only when one is used, so # an index-only edit still runs on a build that predates them. uuid_methods: set[str] = set() @@ -1511,6 +1518,7 @@ def _build_script(operations: list, output: str) -> str: lines = [ "// generated by qet-mcp; do not edit", "var R = {};", # $name -> value from an earlier op + "var F = {};", # $name -> uuid of a folio an op made "var missing = [];", "var need = @NEED@;", "for (var i = 0; i < need.length; i++) {", @@ -1583,6 +1591,11 @@ def _build_script(operations: list, output: str) -> str: raise ValueError( f"operation {op_index} refers to {value!r}, which no earlier " f"operation defined (set \"id\": {name!r} on the op that creates it)") + if kind == "folio" and name in folio_refs: + # the folio's index now, not when it was made; the stored + # index on a build that cannot report folio uuids + ref = _js(name) + return f"(F[{ref}] ? qet.folioIndex(F[{ref}]) : R[{ref}])" return f"R[{_js(name)}]" if kind == "num": if not isinstance(value, (int, float)) or isinstance(value, bool): @@ -1785,6 +1798,10 @@ def _build_script(operations: list, output: str) -> str: if ident is not None: lines.append(f" R[{_js(ident)}] = v{i};") refs.add(ident) + if name in FOLIO_MAKING_OPS: + lines.append(f" F[{_js(ident)}] = (typeof qet.folioUuid === 'function' " + f"&& v{i} >= 0) ? qet.folioUuid(v{i}) : '';") + folio_refs.add(ident) lines.append( f" qet.log({_js(_MARKER)} + JSON.stringify(" f"{{kind: 'op', index: {i}, op: {_js(name)}, " @@ -2655,7 +2672,7 @@ TOOLS = [ "inputSchema": { "type": "object", "properties": { - "binary": {"type": "string", "description": "path to the qelectrotech executable"}, + "binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"}, "project": {"type": "string"}, "format": {"type": "string", "enum": sorted(EXPORT_FORMATS)}, "output": {"type": "string"}, @@ -2664,7 +2681,7 @@ TOOLS = [ "without this an existing file is never clobbered"}, "timeout": {"type": "integer", "default": 180}, }, - "required": ["binary", "project", "format", "output"], + "required": ["project", "format", "output"], }, "handler": lambda a: tool_export(a["binary"], a["project"], a["format"], a["output"], a.get("timeout", 180)), @@ -2682,7 +2699,7 @@ TOOLS = [ "inputSchema": { "type": "object", "properties": { - "binary": {"type": "string", "description": "path to the qelectrotech executable"}, + "binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"}, "project": {"type": "string", "description": "the .qet to start from; not modified"}, "output": {"type": "string", "description": "where to write the edited project"}, "overwrite": {"type": "boolean", "default": False, @@ -2702,7 +2719,8 @@ TOOLS = [ "qet_project_info call 1 is \"folio\": 0 here. A folio can " "be given as its uuid instead (qet_project_info lists them), " "which still names the same folio after an earlier op adds or " - "removes one. A terminal (\"terminal\", \"from_terminal\", " + "removes one; so does the \"$id\" of an add_folio or " + "insert_folio. A terminal (\"terminal\", \"from_terminal\", " "\"to_terminal\") is given by its index -- qet_element_info " "lists terminals in index order, top to bottom then left to " "right -- or by its uuid, which qet_element_info also lists and " @@ -2900,7 +2918,7 @@ TOOLS = [ }, "timeout": {"type": "integer", "default": 180}, }, - "required": ["binary", "project", "output", "operations"], + "required": ["project", "output", "operations"], }, "handler": lambda a: tool_edit(a["binary"], a["project"], a["operations"], a["output"], a.get("elements_dir"), @@ -2918,7 +2936,7 @@ TOOLS = [ "inputSchema": { "type": "object", "properties": { - "binary": {"type": "string", "description": "path to the qelectrotech executable"}, + "binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"}, "project": {"type": "string", "description": "the .qet to query; never modified"}, "sql": {"type": "string", "description": "a single SELECT or WITH...SELECT. " @@ -2926,7 +2944,7 @@ TOOLS = [ "elements_dir": {"type": "string"}, "timeout": {"type": "integer", "default": 180}, }, - "required": ["binary", "project"], + "required": ["project"], }, "handler": lambda a: tool_query(a["binary"], a["project"], a.get("sql", ""), a.get("elements_dir"), a.get("timeout", 180)), @@ -2953,7 +2971,7 @@ TOOLS = [ "inputSchema": { "type": "object", "properties": { - "binary": {"type": "string", "description": "path to the qelectrotech executable"}, + "binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"}, "project": {"type": "string", "description": "the .qet to check; never modified"}, "folio": {"type": "integer", "description": "check one folio only; omit for the whole project. An index " @@ -2964,7 +2982,7 @@ TOOLS = [ "elements_dir": {"type": "string"}, "timeout": {"type": "integer", "default": 180}, }, - "required": ["binary", "project"], + "required": ["project"], }, "handler": lambda a: tool_continuity(a["binary"], a["project"], a.get("folio"), a.get("elements_dir"), a.get("timeout", 180)), @@ -2980,7 +2998,7 @@ TOOLS = [ "inputSchema": { "type": "object", "properties": { - "binary": {"type": "string", "description": "path to the qelectrotech executable"}, + "binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"}, "output": {"type": "string", "description": "where to write the new .qet"}, "title": {"type": "string", "description": "the project title"}, "folios": {"description": "how many empty folios, or a list of folio titles", @@ -2992,7 +3010,7 @@ TOOLS = [ "elements_dir": {"type": "string"}, "timeout": {"type": "integer", "default": 180}, }, - "required": ["binary", "output", "title"], + "required": ["output", "title"], }, "handler": lambda a: tool_project_new( a["binary"], a["output"], a["title"], a.get("folios", 1), a.get("author", ""), @@ -3041,7 +3059,7 @@ TOOLS = [ "inputSchema": { "type": "object", "properties": { - "binary": {"type": "string", "description": "path to the qelectrotech executable"}, + "binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"}, "project": {"type": "string"}, "checks": {"type": "array", "items": {"type": "string", "enum": sorted(CHECKS)}, "description": "which checks to run; omit for all"}, @@ -3050,7 +3068,7 @@ TOOLS = [ "elements_dir": {"type": "string"}, "timeout": {"type": "integer", "default": 180}, }, - "required": ["binary", "project"], + "required": ["project"], }, "handler": lambda a: tool_check(a["binary"], a["project"], a.get("checks"), a.get("sample", 10), a.get("elements_dir"), @@ -3130,17 +3148,21 @@ _BY_NAME = {t["name"]: t for t in TOOLS} # sandboxed HOME each QElectroTech launch gets isolates *settings*, not the # filesystem. # -# So data paths are confined to a workspace. Two kinds of path are treated -# differently, deliberately: +# So data paths are confined to a workspace, and the program the server +# launches is not the client's to choose: # # data chosen by the client per call -- the projects, directories, -# images and outputs below. Confined. -# configuration chosen once by whoever runs the server -- "binary" (the -# qelectrotech executable) and "elements_dir" (the element -# collection). Both normally live in /usr or a build tree, -# i.e. outside any sane workspace, so confining them would -# reject the ordinary case while stopping nothing: they are -# not where a model gets to point the server at /etc. +# images and outputs below. Confined to the workspace. +# executable "binary". Resolved by the server itself (resolve_binary()); +# a client may name it only when it is that same file or one +# whoever configured the server listed in QET_MCP_BINARIES. +# It once counted as configuration and went unchecked, but it +# is a per-call argument: a model steered by text in a +# project could run any program on the machine with it. +# collection "elements_dir". Normally outside the workspace (in /usr or +# a build tree), so allowed there, in the collection of the +# resolved install, or in a directory listed in +# QET_MCP_ELEMENTS. # # Enforced here, at the dispatcher, because this is the trust boundary -- # the point where model-supplied arguments enter. Calling the tool_* helpers @@ -3164,6 +3186,10 @@ _DATA_PATHS = { "qet_element_build": {"write": ("output",)}, } +# Tools that launch QElectroTech, and so take "binary" and "elements_dir". +_LAUNCHES_QET = {"qet_export", "qet_edit", "qet_query", "qet_continuity", + "qet_check", "qet_project_new"} + # qet_edit operations that name a file of their own. _DATA_PATH_OPS = {"add_image": "file", "add_pdf_page": "file"} @@ -3202,6 +3228,106 @@ def _within_workspace(path: Path, roots: list) -> bool: return False +def _env_paths(name: str) -> list: + """An os.pathsep-separated list of paths from the environment, resolved.""" + out = [] + for part in os.environ.get(name, "").split(os.pathsep): + if part.strip(): + try: + out.append(Path(part).expanduser().resolve()) + except OSError: + continue + return out + + +def _installed_prefix() -> Path | None: + """The install prefix when this script is /share/qelectrotech/mcp/.""" + here = Path(__file__).resolve().parent + if here.name == "mcp" and here.parent.name == "qelectrotech" \ + and here.parent.parent.name == "share": + return here.parent.parent.parent + return None + + +def resolve_binary() -> Path | None: + """The QElectroTech this server launches, found without asking the client. + + QET_BINARY first, then the install this script ships in, then + qelectrotech on PATH. None when there is none; the tools that launch + QElectroTech then say how to set it. + """ + env = os.environ.get("QET_BINARY", "").strip() + if env: + return Path(env).expanduser().resolve() + prefix = _installed_prefix() + if prefix is not None: + for name in ("qelectrotech", "qelectrotech.exe"): + cand = prefix / "bin" / name + if cand.is_file(): + return cand.resolve() + found = shutil.which("qelectrotech") + return Path(found).resolve() if found else None + + +def default_elements_dir() -> Path | None: + """The element collection of the install this script ships in, if any.""" + prefix = _installed_prefix() + if prefix is not None: + coll = prefix / "share" / "qelectrotech" / "elements" + if coll.is_dir(): + return coll.resolve() + return None + + +def _check_binary(arguments: dict) -> None: + """Fill in "binary", or refuse one that is not the server's own choice.""" + if os.environ.get("QET_MCP_ALLOW_ANY_BINARY") == "1" and arguments.get("binary"): + return + default = resolve_binary() + raw = arguments.get("binary") + if not raw: + if default is None: + raise ValueError( + "no QElectroTech found: set QET_BINARY to the qelectrotech " + "executable in the environment this server is started in") + arguments["binary"] = str(default) + return + if not isinstance(raw, str): + raise ValueError("'binary' must be a path") + given = Path(raw).expanduser().resolve() + allowed = ([default] if default else []) + _env_paths("QET_MCP_BINARIES") + if given not in allowed: + raise ValueError( + f"'binary' is not an allowed QElectroTech: {given}. Leave it out " + "to use " + (str(default) if default else "QET_BINARY") + + "; whoever configured this server can list others in " + "QET_MCP_BINARIES, or set QET_MCP_ALLOW_ANY_BINARY=1 to " + "disable this check (which lets the client run any program).") + arguments["binary"] = str(given) + + +def _check_elements_dir(arguments: dict, roots: list) -> None: + """Fill in "elements_dir" from the install, or confine a given one.""" + raw = arguments.get("elements_dir") + if not raw: + default = default_elements_dir() + if default is not None: + arguments["elements_dir"] = str(default) + return + if not isinstance(raw, str): + raise ValueError("'elements_dir' must be a path") + given = Path(raw).expanduser().resolve() + extra = _env_paths("QET_MCP_ELEMENTS") + default = default_elements_dir() + if default is not None: + extra.append(default) + if roots and not _within_workspace(given, roots + extra): + raise ValueError( + f"'elements_dir' is outside the workspace: {given}. Leave it out " + "to use the installed collection, or list the directory in " + "QET_MCP_ELEMENTS.") + + def _check_path(raw, arg: str, mode: str, roots: list) -> Path: """Resolve one path and refuse it if it leaves the workspace. @@ -3225,12 +3351,20 @@ def _check_path(raw, arg: str, mode: str, roots: list) -> Path: def enforce_path_policy(tool_name: str, arguments: dict) -> None: - """Apply the workspace and overwrite policy to one tool call.""" + """Apply the workspace, executable and overwrite policy to one tool call. + + For a tool that launches QElectroTech this also fills in "binary" and, + when the install has one, "elements_dir", so a client need not know them. + """ spec = _DATA_PATHS.get(tool_name) if spec is None: return roots = workspace_roots() + if tool_name in _LAUNCHES_QET: + _check_binary(arguments) + _check_elements_dir(arguments, roots) + for arg in spec.get("read", ()): if arg in arguments: _check_path(arguments[arg], arg, "read", roots) diff --git a/misc/qet-mcp/test_qet_mcp.py b/misc/qet-mcp/test_qet_mcp.py index 3c1815ac0..cb2efca91 100644 --- a/misc/qet-mcp/test_qet_mcp.py +++ b/misc/qet-mcp/test_qet_mcp.py @@ -30,6 +30,7 @@ import subprocess import sys import tempfile import unittest +from unittest import mock import xml.etree.ElementTree as ET from pathlib import Path @@ -74,6 +75,16 @@ PLC_MASTER = "common://plc_master_test.elmt" PLC_SLAVE = "common://plc_slave_test.elmt" +def fake_qet(bindir: Path, name: str = "qelectrotech") -> Path: + """An executable file standing in for QElectroTech. The policy checks + which file it is, never runs it.""" + bindir.mkdir(parents=True, exist_ok=True) + exe = bindir / name + exe.write_text("#!/bin/sh\nexit 0\n") + exe.chmod(0o755) + return exe + + def png(path: Path) -> None: """A real 64x32 PNG from the standard library, so no imaging dependency.""" import struct @@ -184,7 +195,8 @@ class EditValidation(unittest.TestCase): s = self.build([{"op": "add_folio", "id": "f"}, {"op": "add_element", "id": "k", "folio": "$f", "path": "p", "x": 0, "y": 0}, {"op": "delete_element_text", "folio": "$f", "element": "$k", "index": U}]) - self.assertIn(f'qet.deleteElementText(R["f"], R["k"], qet.elementTextIndex(R["f"], R["k"], "{U}"))', s) + f = '(F["f"] ? qet.folioIndex(F["f"]) : R["f"])' # a folio "$name": see test_folio_ref_follows_the_folio + self.assertIn(f'qet.deleteElementText({f}, R["k"], qet.elementTextIndex({f}, R["k"], "{U}"))', s) # the lookups are required only when a uuid is used self.assertIn('"tableIndex"', self.build([{"op": "delete_table", "folio": 0, "table": U}])) self.assertNotIn('"tableIndex"', self.build([{"op": "delete_table", "folio": 0, "table": 0}])) @@ -214,6 +226,27 @@ class EditValidation(unittest.TestCase): with self.assertRaisesRegex(ValueError, "folio index, its uuid"): self.build([{"op": "set_folio", "folio": "first", "property": "author", "value": "a"}]) + def test_folio_ref_follows_the_folio(self): + """A "$name" made by add_folio or insert_folio is looked up by the + folio's uuid when used as a folio, since a later insert or removal + shifts its index; the stored index is the fallback on a build that + cannot report folio uuids. Used as anything else, it is unchanged.""" + s = self.build([{"op": "add_folio", "id": "f"}, + {"op": "insert_folio", "id": "g", "position": 0}, + {"op": "set_folio_title", "folio": "$f", "title": "t"}, + {"op": "set_folio_title", "folio": "$g", "title": "u"}]) + self.assertIn("F[\"f\"] = (typeof qet.folioUuid === 'function' && v0 >= 0) " + "? qet.folioUuid(v0) : '';", s) + self.assertIn('qet.setFolioTitle((F["f"] ? qet.folioIndex(F["f"]) : R["f"]), "t")', s) + self.assertIn('qet.setFolioTitle((F["g"] ? qet.folioIndex(F["g"]) : R["g"]), "u")', s) + # not required: an edit still runs on a build without folio uuids + self.assertNotIn('"folioUuid"', s) + # a "$name" from any other op is untouched + s = self.build([{"op": "add_text", "id": "t", "folio": 0, "text": "x", "x": 0, "y": 0}, + {"op": "delete_text", "folio": 0, "index": "$t"}]) + self.assertIn('R["t"]', s) + self.assertNotIn("F[", s.split("if (missing.length === 0) {")[1]) + def test_conductor_by_uuid(self): """A conductor named by uuid is turned into one of its ends at run time; the lookup is required only then, and "conductor" cannot be @@ -267,7 +300,8 @@ class EditValidation(unittest.TestCase): s = self.build([{"op": "add_folio", "id": "f"}, {"op": "add_element", "id": "a", "folio": "$f", "path": "x.elmt", "x": 0, "y": 0}, {"op": "delete_conductor", "folio": "$f", "element": "$a", "terminal": T}]) - self.assertIn(f'qet.deleteConductor(R["f"], R["a"], qetMcpTerminal(2, "terminal", R["f"], R["a"], "{T}"))', s) + f = '(F["f"] ? qet.folioIndex(F["f"]) : R["f"])' # a folio "$name": see test_folio_ref_follows_the_folio + self.assertIn(f'qet.deleteConductor({f}, R["a"], qetMcpTerminal(2, "terminal", {f}, R["a"], "{T}"))', s) s = self.build([{"op": "delete_conductor", "folio": V, "element": E, "terminal": T}]) self.assertIn(f'qetMcpTerminal(0, "terminal", qet.folioIndex("{V}"), "{E}", "{T}")', s) # an index is unchanged and needs no lookup @@ -458,6 +492,80 @@ class EditValidation(unittest.TestCase): need = json.loads(re.search(r"var need = (\[.*?\]);", script).group(1)) self.assertFalse({"shapeIndex", "textIndex", "imageIndex"} & set(need)) + def test_malformed_arguments_are_refused(self): + """Each argument kind rejects what it cannot take, before any launch. + Found by the mutation audit: forcing any of these checks off went + unnoticed.""" + f = {"op": "add_folio", "id": "f"} + cases = [ + ("indices", {"op": "group_terminals", "strip": 0, "indices": []}), + ("indices", {"op": "group_terminals", "strip": 0, "indices": [1, "2"]}), + ("indices", {"op": "group_terminals", "strip": 0, "indices": [True]}), + ("indices", {"op": "bridge_terminals", "strip": 0, "indices": 3}), + ("true or false", {"op": "add_polygon", "folio": 0, "closed": 1, + "points": [{"x": 0, "y": 0}, {"x": 1, "y": 1}]}), + ("at least 2", {"op": "add_polygon", "folio": 0, "closed": True, + "points": [{"x": 0, "y": 0}]}), + ("at least 2", {"op": "add_polygon", "folio": 0, "closed": True, "points": "xy"}), + ("entries must be", {"op": "add_polygon", "folio": 0, "closed": True, + "points": [{"x": 0, "y": 0}, {"x": "1", "y": 1}]}), + ("entries must be", {"op": "add_polygon", "folio": 0, "closed": True, + "points": [{"x": 0, "y": 0}, {"x": True, "y": 1}]}), + ("entries must be", {"op": "add_polygon", "folio": 0, "closed": True, + "points": [{"x": 0, "y": 0}, [1, 1]]}), + ("at least 2 nodes", {"op": "add_path", "folio": 0, "closed": False, + "nodes": [{"x": 0, "y": 0}]}), + ("corner, smooth or symmetric", {"op": "add_path", "folio": 0, "closed": False, + "nodes": [{"x": 0, "y": 0}, + {"x": 1, "y": 1, "kind": "sharp"}]}), + ("inHandle", {"op": "add_path", "folio": 0, "closed": False, + "nodes": [{"x": 0, "y": 0}, {"x": 1, "y": 1, "inHandle": [0, 0]}]}), + ("outHandle", {"op": "add_path", "folio": 0, "closed": False, + "nodes": [{"x": 0, "y": 0}, {"x": 1, "y": 1, "outHandle": {"x": 0}}]}), + ("unknown kind", {"op": "search_and_replace", "kind": "folio", "field": "x", + "pattern": "a", "replacement": "b", "regex": False, + "case_sensitive": False}), + ("unknown conductor field", {"op": "search_and_replace", "kind": "conductor", + "field": "label", "pattern": "a", "replacement": "b", + "regex": False, "case_sensitive": False}), + ("non-empty", {"op": "search_and_replace", "kind": "element_info", "field": "", + "pattern": "a", "replacement": "b", "regex": False, + "case_sensitive": False}), + ("not both", {"op": "delete_conductor", "folio": 0, + "conductor": "{11111111-2222-4333-8444-555555555555}", + "element": "{11111111-2222-4333-8444-555555555555}"}), + ("not both", {"op": "delete_conductor", "folio": 0, + "conductor": "{11111111-2222-4333-8444-555555555555}", "terminal": 0}), + ("not an object", "add_folio"), + ] + for message, op in cases: + with self.subTest(op=op): + with self.assertRaisesRegex(ValueError, message): + self.build([f, op]) + + def test_valid_shapes_of_those_arguments_pass(self): + """The other side: the same kinds accept what they should, so the + refusals above are not everything failing.""" + s = self.build([ + {"op": "add_polygon", "folio": 0, "closed": False, + "points": [{"x": 0, "y": 0}, {"x": 1.5, "y": -2}]}, + {"op": "add_path", "folio": 0, "closed": True, + "nodes": [{"x": 0, "y": 0, "kind": "smooth", "inHandle": {"x": 1, "y": 1}}, + {"x": 5, "y": 5, "outHandle": {"x": 2, "y": 2}}]}, + {"op": "add_polygon", "folio": 0, "closed": True, + "points": [{"x": 0, "y": 0, "kind": "anything"}, {"x": 1, "y": 1}]}, + {"op": "search_and_replace", "kind": "conductor", "field": "num", "pattern": "a", + "replacement": "b", "regex": True, "case_sensitive": False}, + {"op": "search_and_replace", "kind": "text", "field": "", "pattern": "a", + "replacement": "b", "regex": False, "case_sensitive": True}]) + self.assertIn("qet.addPolygon(0, [{", s) + self.assertIn("qet.searchAndReplace(\"conductor\", \"num\", \"a\", \"b\", true, false)", s) + self.assertIn("qet.searchAndReplace(\"text\", \"\", \"a\", \"b\", false, true)", s) + + def test_no_id_stores_nothing(self): + s = self.build([{"op": "add_folio"}]) + self.assertNotIn("R[", s.split("if (missing.length === 0) {")[1]) + def test_a_string_index_that_is_not_a_uuid_is_refused(self): for bad in ("3", "{nope}", "11111111-1111"): with self.subTest(index=bad): @@ -521,7 +629,8 @@ class EditValidation(unittest.TestCase): script = self.build([{"op": "add_folio", "id": "f"}, {"op": "set_folio_title", "folio": "$f", "title": nasty}]) # the literal must be valid JSON, so JavaScript reads exactly what was sent - literal = re.search(r'setFolioTitle\(R\["f"\], (".*?")\)', script, re.S).group(1) + literal = re.search(r'setFolioTitle\(\(F\["f"\] \? qet\.folioIndex\(F\["f"\]\) : R\["f"\]\), (".*?")\)', + script, re.S).group(1) self.assertEqual(json.loads(literal), nasty) def test_edit_refuses_in_place_and_empty(self): @@ -566,6 +675,37 @@ class ResultParsing(unittest.TestCase): self.assertEqual(parsed["operations"], []) + def test_capabilities_notes_and_save(self): + """Every field the script reports reaches the result, attached where + it belongs. Found by the mutation audit: only the binary tests read + these back.""" + text = "\n".join([ + self.line(kind="capabilities", missing=["addFolio"]), + self.line(kind="op", index=0, op="a", id=None, result=True), + self.line(kind="op", index=1, op="b", id=None, result=False), + self.line(kind="op_note", index=1, note="why"), + self.line(kind="save", result=True, stopped_early=True), + self.line(kind="something_newer", result=False)]) # not a save + parsed = m._parse_script_output(text) + self.assertEqual(set(parsed), {"missing_methods", "operations", "saved", "stopped_early"}) + self.assertEqual(parsed["missing_methods"], ["addFolio"]) + self.assertTrue(parsed["saved"]) + self.assertTrue(parsed["stopped_early"]) + self.assertNotIn("note", parsed["operations"][0]) + self.assertEqual(parsed["operations"][1]["note"], "why") + # no capabilities line: unknown, not "nothing missing" + self.assertIsNone(m._parse_script_output(self.line(kind="save", result=False))["missing_methods"]) + self.assertEqual(m._parse_script_output(self.line(kind="capabilities", missing=None)) + ["missing_methods"], []) + + def test_a_line_without_the_marker_is_ignored(self): + """Even one that would parse if read from where the marker would be.""" + almost = "x" * (len(m._MARKER) - 1) + json.dumps({"kind": "save", "result": True}) + parsed = m._parse_script_output(almost) + self.assertIsNone(parsed["saved"]) + self.assertFalse(parsed["stopped_early"]) + + class TerminalOrder(unittest.TestCase): """QElectroTech indexes terminals top-to-bottom then left-to-right, not in file order. Getting it wrong wires the wrong end of a coil with no error; @@ -709,6 +849,391 @@ class ElementBuild(unittest.TestCase): self.assertEqual(r["verified"]["names"]["en"], 'Coil "A" & ') +class ElementFileExact(unittest.TestCase): + """Exact answers for the tools that write and read .elmt files and need + no QElectroTech. Found by a mutation audit: the existing tests checked a + few fields, so dropping any other one from a result, or getting a + header number off by one, went unnoticed.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + m._ELEMENT_INDEX.clear() + + def tearDown(self): + self.tmp.cleanup() + + def test_fmt(self): + self.assertEqual([m._fmt(v) for v in (True, False, 3, 2.0, 2.5, -0.25, "x")], + ["true", "false", "3", "2", "2.5", "-0.25", "x"]) + + def test_part_extent_per_kind(self): + ext = m._part_extent + self.assertEqual(ext("line", {"x1": 1, "y1": 2, "x2": 3, "y2": 4}), [(1, 2), (3, 4)]) + for kind in ("rect", "ellipse", "arc"): + self.assertEqual(ext(kind, {"x": -5, "y": 1, "width": 10, "height": 4}), + [(-5, 1), (5, 5)], kind) + self.assertEqual(ext("circle", {"x": 2, "y": 3, "diameter": 6}), [(2, 3), (8, 9)]) + self.assertEqual(ext("polygon", {"points": [[0, 1], [2, 3], [4, 5]]}), + [(0, 1), (2, 3), (4, 5)]) + self.assertEqual(ext("text", {"x": 7, "y": 8, "text": "a"}), [(7, 8)]) + self.assertEqual(ext("dynamic_text", {"x": 7}), []) + + def test_element_geometry(self): + """Box, hotspot and size, worked out by hand: points reach x -10..20 + and y -15..5; a 5 unit margin, rounded out to tens.""" + g = m._element_geometry( + [{"type": "line", "x1": 0, "y1": 0, "x2": 20, "y2": 0}, + {"type": "rect", "x": -10, "y": -5, "width": 10, "height": 10}], + [{"x": 0, "y": -15, "orientation": "n"}]) + self.assertEqual(g, {"width": 50, "height": 30, "hotspot_x": 20, "hotspot_y": 20, + "bbox": [-10, -15, 20, 5]}) + # a terminal alone is enough; nothing at all is refused + self.assertEqual(m._element_geometry([], [{"x": 0, "y": 0, "orientation": "n"}]), + {"width": 20, "height": 20, "hotspot_x": 10, "hotspot_y": 10, "bbox": [0, 0, 0, 0]}) + with self.assertRaisesRegex(ValueError, "at least one part or terminal"): + m._element_geometry([], []) + + def test_part_element(self): + E = lambda part: dict(m._part_element(part, "{u}").attrib) + self.assertEqual(E({"type": "polygon", "points": [[0, 1], [2.5, 3]]}), + {"uuid": "{u}", "x1": "0", "y1": "1", "x2": "2.5", "y2": "3", + "closed": "true", "antialias": "true", "style": m.DEFAULT_STYLE}) + self.assertEqual(E({"type": "polygon", "points": [[0, 1], [2, 3]], "closed": False, + "antialias": False, "style": "x"})["closed"], "false") + self.assertEqual(E({"type": "text", "x": 1, "y": 2, "text": "K1"}), + {"uuid": "{u}", "x": "1", "y": "2", "text": "K1", "rotation": "0", + "font": "Sans Serif,9,-1,5,50,0,0,0,0,0", "color": "#000000"}) + self.assertEqual(E({"type": "text", "x": 1, "y": 2, "text": "K1", "size": 12, + "rotation": 90, "color": "red"})["font"], + "Sans Serif,12,-1,5,50,0,0,0,0,0") + rect = E({"type": "rect", "x": 0, "y": 0, "width": 4, "height": 2, "antialias": False}) + self.assertEqual(rect, {"uuid": "{u}", "x": "0", "y": "0", "width": "4", "height": "2", + "antialias": "false", "style": m.DEFAULT_STYLE}) + + def test_build_writes_and_reports_exactly(self): + out = self.root / "k" / "coil.elmt" + U = "{11111111-2222-4333-8444-555555555555}" + r = m.tool_element_build( + str(out), {"fr": "Bobine", "en": "Coil"}, + [{"type": "line", "x1": 0, "y1": -10, "x2": 0, "y2": 10}], + terminals=[{"x": 0, "y": 20, "orientation": "s", "name": "A2"}, + {"x": 0, "y": -20, "orientation": "n", "name": "A1", "type": "Inner"}], + link_type="master", informations={"type": "coil"}, uuid=U) + self.assertEqual(set(r), {"ok", "output", "bytes", "width", "height", "hotspot_x", + "hotspot_y", "bbox", "terminal_index_order", "part_uuids", + "verified"}) + self.assertTrue(r["ok"]) + self.assertEqual(r["output"], str(out)) + self.assertEqual(r["bytes"], out.stat().st_size) + self.assertEqual(r["terminal_index_order"], ["A1", "A2"]) + root = ET.parse(out).getroot() + self.assertEqual(dict(root.attrib), { + "version": "0.100.0", "type": "element", "link_type": "master", + "width": str(r["width"]), "height": str(r["height"]), + "hotspot_x": str(r["hotspot_x"]), "hotspot_y": str(r["hotspot_y"])}) + self.assertEqual(root.find("uuid").get("uuid"), U) + self.assertEqual([(n.get("lang"), n.text) for n in root.iter("name")], + [("en", "Coil"), ("fr", "Bobine")]) + self.assertEqual([(k.get("name"), k.text) for k in root.iter("kindInformation")], + [("type", "coil")]) + t2, t1 = root.findall("description/terminal") + self.assertEqual({k: v for k, v in t1.attrib.items() if k != "uuid"}, + {"x": "0", "y": "-20", "orientation": "n", "type": "Inner", "name": "A1"}) + self.assertEqual(t2.get("type"), "Generic") + self.assertRegex(t1.get("uuid"), m._UUID_RE) + self.assertNotEqual(t1.get("uuid"), t2.get("uuid")) + + def test_build_refusals(self): + out = str(self.root / "x.elmt") + line = [{"type": "line", "x1": 0, "y1": 0, "x2": 1, "y2": 0}] + term = [{"x": 0, "y": 0, "orientation": "n"}] + for message, kwargs in [ + ("names must be a non-empty", dict(names={}, parts=line, terminals=term)), + ("unknown link_type", dict(names={"en": "a"}, parts=line, terminals=term, link_type="x")), + ("parts must be a list", dict(names={"en": "a"}, parts={}, terminals=term)), + ("terminal 0 is not an object", dict(names={"en": "a"}, parts=line, terminals=[1])), + ("terminal 0 is missing 'orientation'", dict(names={"en": "a"}, parts=line, + terminals=[{"x": 0, "y": 0}])), + ("orientation is one of", dict(names={"en": "a"}, parts=line, + terminals=[{"x": 0, "y": 0, "orientation": "up"}])), + ("cannot be connected", dict(names={"en": "a"}, parts=line, terminals=[])), + ("same uuid", dict(names={"en": "a"}, terminals=term, parts=[ + dict(line[0], uuid="{11111111-2222-4333-8444-555555555555}"), + dict(line[0], uuid="{11111111-2222-4333-8444-555555555555}")]))]: + with self.subTest(message=message): + with self.assertRaisesRegex(ValueError, message): + m.tool_element_build(out, **kwargs) + + def test_element_info_exactly(self): + p = self.root / "e.elmt" + p.write_text( + '' + 'E' + ' label ' + '' + '' + '' + '', + encoding="utf-8") + r = m.tool_element_info(str(p)) + self.assertEqual(set(r), {"file", "type", "link_type", "width", "height", "names", + "terminal_count", "terminals", "terminal_order", + "info_fields", "parts", "part_list"}) + self.assertEqual((r["file"], r["type"], r["link_type"], r["width"], r["height"]), + (str(p), "element", "simple", "20", "40")) + self.assertEqual(r["terminal_count"], 3) + self.assertEqual(r["terminals"][0], {"index": 0, "x": "0", "y": "-10", "orientation": "n", + "name": "1", "type": "", "uuid": ""}) + self.assertEqual(r["terminals"][2], {"index": 2, "x": "0", "y": "10", "orientation": "s", + "name": "2", "type": "Generic", "uuid": "{B}"}) + self.assertEqual(r["info_fields"], ["label"]) + self.assertEqual(r["parts"], {"line": 1, "terminal": 3, "arc": 1}) + self.assertEqual(r["part_list"], [{"type": "line", "uuid": "{L}"}, {"type": "arc", "uuid": ""}]) + self.assertNotIn("undefined", r["terminal_order"]) + p.write_text(p.read_text().replace('x="5" y="-10"', 'x="0" y="-10"'), encoding="utf-8") + self.assertIn("undefined", m.tool_element_info(str(p))["terminal_order"]) + + def test_index_entry_exactly(self): + f = self.root / "a" / "k.elmt" + f.parent.mkdir() + f.write_text('' + 'Bobine' + ' coil ' + 'x' + '' + '', + encoding="utf-8") + (self.root / "a" / "broken.elmt").write_text("", encoding="utf-8") + items = m._index_collection(self.root) + self.assertEqual(len(items), 1) + it = {k: v for k, v in items[0].items() if k != "haystack"} + self.assertEqual(it, {"path": "common://a/k.elmt", "file": str(f), "name": "Bobine", + "names": {"fr": "Bobine"}, "link_type": "master", "kind": "coil", + "terminals": 2, "terminal_names": ["A2", "A1"], + "terminal_order_ambiguous": True, "width": "30", "height": "50"}) + # the index is cached until the collection changes + self.assertIs(m._index_collection(self.root), items) + self.assertEqual(m._collection_signature(self.root)[0], 3) + + def test_search_ranking_and_limit(self): + for rel, name in (("a/1.elmt", "Coil latching"), ("a/2.elmt", "Coil"), + ("a/3.elmt", "Remanence coil")): + f = self.root / rel + f.parent.mkdir(exist_ok=True) + f.write_text(f'' + f'{name}' + f'', + encoding="utf-8") + r = m.tool_element_search(str(self.root), "coil") + self.assertEqual([e["name"] for e in r["results"]], + ["Coil", "Coil latching", "Remanence coil"]) + r = m.tool_element_search(str(self.root), "coil", limit=1) + self.assertEqual((r["total_matches"], r["returned"]), (3, 1)) + with self.assertRaisesRegex(ValueError, "limit must be >= 1"): + m.tool_element_search(str(self.root), "coil", limit=0) + self.assertEqual(m.tool_element_search(str(self.root), "coil", limit=1)["results"][0]["name"], + "Coil") + self.assertEqual(set(r), {"query", "total_matches", "returned", "indexed", "results"}) + self.assertEqual((r["query"], r["indexed"]), ("coil", 3)) + self.assertEqual(r["results"][0]["languages"], ["en"]) + + def test_search_puts_the_exact_name_then_names_starting_with_it(self): + for n, name in enumerate(("Relay coil", "Coil relay", "Coil relay X", "A coil relay")): + f = self.root / f"{n}.elmt" + f.write_text(f'' + f'{name}' + f'', + encoding="utf-8") + r = m.tool_element_search(str(self.root), "coil relay") + self.assertEqual([e["name"] for e in r["results"]], + ["Coil relay", "Coil relay X", "Relay coil", "A coil relay"]) + + def test_search_returns_25_by_default(self): + for n in range(26): + (self.root / f"{n:02}.elmt").write_text( + f'Coil {n}' + f'' + f'', encoding="utf-8") + r = m.tool_element_search(str(self.root), "coil") + self.assertEqual((r["total_matches"], r["returned"]), (26, 25)) + + def test_search_exact_name_wins_a_tie(self): + """Same length, same first word: the exact name comes first, not + the one whose path sorts first.""" + for rel, name in (("a.elmt", "Coil-a"), ("z.elmt", "Coil a")): + (self.root / rel).write_text( + f'{name}' + f'' + f'', encoding="utf-8") + r = m.tool_element_search(str(self.root), "coil a") + self.assertEqual([e["name"] for e in r["results"]], ["Coil a", "Coil-a"]) + + def test_validate_part_refusals(self): + ok = lambda part: m._validate_part(0, part) + self.assertEqual(ok({"type": "polygon", "points": [[0, 0], [1, 1]]}), "polygon") + for message, part in [ + ("not an object", ["line"]), + ("at least two points", {"type": "polygon", "points": [[0, 0]]}), + ("at least two points", {"type": "polygon", "points": "ab"}), + ("each point is", {"type": "polygon", "points": [[0, 0], [1, 1, 1]]}), + ("each point is", {"type": "polygon", "points": [[0, 0], 5]}), + ("uuid must look like", {"type": "line", "x1": 0, "y1": 0, "x2": 1, "y2": 1, + "uuid": "L1"})]: + with self.subTest(message=message): + with self.assertRaisesRegex(ValueError, message): + ok(part) + + def test_terminal_order_with_unreadable_coordinates(self): + T = lambda **a: ET.Element("terminal", {k: str(v) for k, v in a.items()}) + ordered, _ = m._terminals_in_index_order([T(x=0, y=0.5, name="b"), T(x=0, y="bad", name="a")]) + self.assertEqual([t.get("name") for t in ordered], ["a", "b"]) + ordered, _ = m._terminals_in_index_order([T(x=0.5, y=0, name="b"), T(x="bad", y=0, name="a")]) + self.assertEqual([t.get("name") for t in ordered], ["a", "b"]) + # a missing coordinate counts as 0 + ordered, _ = m._terminals_in_index_order([T(x=0, y=0.5, name="b"), T(x=0, name="a")]) + self.assertEqual([t.get("name") for t in ordered], ["a", "b"]) + ordered, _ = m._terminals_in_index_order([T(x=0.5, y=0, name="b"), T(y=0, name="a")]) + self.assertEqual([t.get("name") for t in ordered], ["a", "b"]) + + +class CheckAndContinuityAnswers(unittest.TestCase): + """qet_check and qet_continuity turn QElectroTech's log lines into their + answer. With _run_qet replaced by a stub that returns chosen lines, the + whole answer can be checked exactly, without QElectroTech. Found by a + mutation audit: the binary tests look at a finding or two, so a wrong + summary count, a dropped field or a wrong "passed" went unnoticed.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.qet = Path(self.tmp.name) / "p.qet" + self.qet.write_text('', + encoding="utf-8") + + def tearDown(self): + self.tmp.cleanup() + + def stub(self, lines, **extra): + out = "\n".join(["noise", m._MARKER + "{not json"] + + [m._MARKER + json.dumps(l) for l in lines]) + return mock.patch.object(m, "_run_qet", lambda *a, **k: {"stdout": out, "stderr": "", **extra}) + + def test_check_answer_exactly(self): + rows = [{"label": f"K{i}"} for i in range(12)] + lines = [ + {"kind": "check", "name": "duplicate_master_labels", "rows": rows, "error": ""}, + {"kind": "check", "name": "unlabelled_masters", "rows": [{"x": 1}], "error": ""}, + {"kind": "check", "name": "unnumbered_conductors", "rows": [{"n": 1}, {"n": 2}], "error": ""}, + {"kind": "check", "name": "duplicate_simple_labels", "rows": [], "error": ""}, + {"kind": "check", "name": "empty_folios", "rows": None, "error": "bad SQL"}, + {"kind": "other", "name": "masters_without_manufacturer_reference", "rows": [1]}, + ] + with self.stub(lines): + r = m.tool_check("qet", str(self.qet)) # sample defaults to 10 + C = m.CHECKS + self.assertEqual(r, { + "ok": False, + "summary": {"errors": 1, "warnings": 1, "info": 1, "passed": 1, "check_failures": 2}, + "findings": [ + {"check": "duplicate_master_labels", "severity": "error", "count": 12, + "note": C["duplicate_master_labels"]["note"], "rows": rows[:10]}, + {"check": "unlabelled_masters", "severity": "warning", "count": 1, + "note": C["unlabelled_masters"]["note"], "rows": [{"x": 1}]}, + {"check": "unnumbered_conductors", "severity": "info", "count": 2, + "note": C["unnumbered_conductors"]["note"], "rows": [{"n": 1}, {"n": 2}]}], + "passed": ["duplicate_simple_labels"], + "check_failures": [ + {"check": "empty_folios", "error": "bad SQL"}, + {"check": "masters_without_manufacturer_reference", "error": "no result came back"}]}) + + def test_check_sorts_by_severity_then_name(self): + lines = [{"kind": "check", "name": "unlabelled_masters", "rows": [1], "error": ""}, + {"kind": "check", "name": "empty_folios", "rows": [1], "error": ""}] + with self.stub(lines): + r = m.tool_check("qet", str(self.qet), checks=["empty_folios", "unlabelled_masters"]) + self.assertEqual([f["check"] for f in r["findings"]], ["unlabelled_masters", "empty_folios"]) + + def test_check_failure_alone_is_not_ok(self): + lines = [{"kind": "check", "name": "empty_folios", "rows": None, "error": "bad SQL"}] + with self.stub(lines): + r = m.tool_check("qet", str(self.qet), checks=["empty_folios"]) + self.assertFalse(r["ok"]) + + def test_check_ignores_a_line_without_the_marker(self): + almost = "x" * (len(m._MARKER) - 1) + json.dumps( + {"kind": "check", "name": "empty_folios", "rows": [1], "error": ""}) + with mock.patch.object(m, "_run_qet", lambda *a, **k: {"stdout": almost, "stderr": ""}): + r = m.tool_check("qet", str(self.qet), checks=["empty_folios"]) + self.assertEqual(r["check_failures"], [{"check": "empty_folios", "error": "no result came back"}]) + + def test_check_ok_without_errors_and_sample_zero(self): + lines = [{"kind": "check", "name": "unlabelled_masters", "rows": [{"x": 1}], "error": ""}] + with self.stub(lines): + r = m.tool_check("qet", str(self.qet), checks=["unlabelled_masters"], sample=0) + self.assertTrue(r["ok"]) # a warning is not a failure + self.assertEqual(r["findings"][0]["rows"], []) + self.assertEqual(r["findings"][0]["count"], 1) + with self.assertRaisesRegex(ValueError, "sample must be >= 0"): + m.tool_check("qet", str(self.qet), sample=-1) + with self.assertRaisesRegex(ValueError, "no such project"): + m.tool_check("qet", str(self.qet) + ".missing") + + def test_check_carries_the_launch_hint(self): + with self.stub([], hint="why it did not start"): + r = m.tool_check("qet", str(self.qet), checks=["empty_folios"]) + self.assertFalse(r["ok"]) + self.assertEqual(r["hint"], "why it did not start") + + def test_continuity_answer_exactly(self): + found = [{"severity": "error", "folio": 1, "what": "a"}, + {"severity": "warning", "folio": 0, "what": "b"}, + {"severity": "info", "folio": "?", "what": "c"}, + {"severity": "info", "what": "d"}, + {"severity": "info", "folio": 1, "what": "e"}] + with self.stub([{"kind": "continuity", "findings": found}, + {"kind": "other", "findings": []}]): + r = m.tool_continuity("qet", str(self.qet), folio=1) + self.assertEqual((r["finding_count"], r["errors"], r["warnings"], r["info"]), (5, 1, 1, 3)) + self.assertEqual([f.get("folio_number") for f in r["findings"]], [2, 1, None, None, 2]) + self.assertNotIn(m._MARKER, r["stdout"]) + self.assertIn("noise", r["stdout"]) + + def test_continuity_passes_the_folio_to_the_script(self): + seen = [] + def fake(binary, args, **kw): + seen.append(kw["script"]) + return {"stdout": m._MARKER + json.dumps({"kind": "continuity", "findings": []}), + "stderr": ""} + with mock.patch.object(m, "_run_qet", fake): + m.tool_continuity("qet", str(self.qet)) + m.tool_continuity("qet", str(self.qet), folio=1) + self.assertIn("qet.checkContinuity(-1)", seen[0]) + self.assertIn("qet.checkContinuity(1)", seen[1]) + with self.assertRaisesRegex(ValueError, "no such project"): + m.tool_continuity("qet", str(self.qet) + ".missing") + almost = "x" * (len(m._MARKER) - 1) + json.dumps({"kind": "continuity", "findings": []}) + with mock.patch.object(m, "_run_qet", lambda *a, **k: {"stdout": almost, "stderr": ""}): + self.assertFalse(m.tool_continuity("qet", str(self.qet))["ok"]) + + def test_continuity_without_findings_is_not_ok(self): + with self.stub([]): + r = m.tool_continuity("qet", str(self.qet)) + self.assertFalse(r["ok"]) + self.assertIn("predate qet.checkContinuity()", r["hint"]) + with self.stub([], hint="launch failed"): + self.assertEqual(m.tool_continuity("qet", str(self.qet))["hint"], "launch failed") + + def test_continuity_folio_bounds(self): + for bad in (-1, 2): + with self.subTest(folio=bad): + with self.assertRaisesRegex(ValueError, f"folio {bad} does not exist"): + m.tool_continuity("qet", str(self.qet), folio=bad) + with self.stub([{"kind": "continuity", "findings": []}]): + self.assertEqual(m.tool_continuity("qet", str(self.qet), folio=0)["finding_count"], 0) + + class ElementSearch(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() @@ -962,6 +1487,32 @@ class ReadToolContracts(unittest.TestCase): self.assertEqual(numbered, "1:{E}@0,-4,2--{E}@6,0,1") self.assertEqual(by_uuid, numbered) + def test_conductor_key_same_in_both_forms_in_a_column(self): + """Terminals one above the other share their x: a uuid end must be + matched on x, y and orientation together, not on any one of them. + Found by the mutation audit: an "or" in place of "and" went + unnoticed while every test's terminals differed in every coordinate.""" + root = ET.fromstring( + '' + '' + '' + '' + '' + '' + '' + '' + '' + '' + '' + '' + '' + '' + '') + numbered, by_uuid = [m._conductor_row(i, c, ix)["key"] + for i, c, ix in m._conductors(root)] + self.assertEqual(numbered, "1:{E}@0,24,0--{E}@30,24,0") + self.assertEqual(by_uuid, numbered) + def test_conductor_row_without_an_index(self): c = ET.fromstring('') self.assertEqual(m._conductor_row(3, c)["key"], "3:#7--#8") @@ -1767,6 +2318,8 @@ class PathPolicy(unittest.TestCase): self._saved = dict(os.environ) os.environ["QET_MCP_WORKSPACE"] = str(self.root) os.environ.pop("QET_MCP_ALLOW_ANY_PATH", None) + self.qet = fake_qet(Path(self.tmp.name) / "bin") + os.environ["QET_BINARY"] = str(self.qet) def tearDown(self): os.environ.clear() @@ -1800,7 +2353,7 @@ class PathPolicy(unittest.TestCase): def test_write_outside_the_workspace_is_refused(self): with self.assertRaisesRegex(ValueError, "outside the workspace"): m.enforce_path_policy("qet_export", { - "binary": "/usr/bin/qelectrotech", + "binary": str(self.qet), "project": str(self.root / "ok.qet"), "format": "pdf", "output": str(self.outside / "exfiltrated.pdf")}) @@ -1808,7 +2361,7 @@ class PathPolicy(unittest.TestCase): def test_existing_output_is_not_clobbered_without_overwrite(self): target = self.root / "existing.qet" target.write_text("precious") - args = {"binary": "/usr/bin/qelectrotech", "output": str(target), "title": "T"} + args = {"binary": str(self.qet), "output": str(target), "title": "T"} with self.assertRaisesRegex(ValueError, "already exists"): m.enforce_path_policy("qet_project_new", args) # ... and goes through once the caller says so explicitly @@ -1817,12 +2370,12 @@ class PathPolicy(unittest.TestCase): def test_new_output_needs_no_overwrite_flag(self): m.enforce_path_policy("qet_project_new", { - "binary": "/usr/bin/qelectrotech", + "binary": str(self.qet), "output": str(self.root / "brand_new.qet"), "title": "T"}) def test_operation_level_file_paths_are_checked(self): """add_image/add_pdf_page carry their own path, one level down.""" - base = {"binary": "/usr/bin/qelectrotech", + base = {"binary": str(self.qet), "project": str(self.root / "ok.qet"), "output": str(self.root / "out.qet")} outside_png = str(self.outside / "anything.png") @@ -1833,17 +2386,6 @@ class PathPolicy(unittest.TestCase): with self.assertRaisesRegex(ValueError, "outside the workspace"): m.enforce_path_policy("qet_edit", dict(base, operations=[op])) - def test_configuration_paths_are_exempt(self): - """binary and elements_dir are the operator's choice, not the model's. - - Both normally live in /usr or a build tree, so confining them would - reject the ordinary case while stopping nothing. - """ - m.enforce_path_policy("qet_query", { - "binary": "/usr/bin/qelectrotech", - "project": str(self.root / "ok.qet"), - "elements_dir": "/usr/share/qelectrotech/elements"}) - def test_several_roots_may_be_allowed(self): os.environ["QET_MCP_WORKSPACE"] = os.pathsep.join( [str(self.root), str(self.outside)]) @@ -1880,8 +2422,8 @@ class PathPolicy(unittest.TestCase): def test_every_data_path_argument_is_guarded(self): """The other direction: a tool whose schema takes a data path must be in the policy, or that path is read or written with no workspace - check at all -- and nothing fails. binary and elements_dir are - configuration, deliberately not confined (see the README).""" + check at all -- and nothing fails. binary and elements_dir have + their own rule (BinaryPolicy).""" pathish = {"path", "project", "before", "after", "output", "directory"} for t in m.TOOLS: with self.subTest(tool=t["name"]): @@ -1902,6 +2444,125 @@ class PathPolicy(unittest.TestCase): f"{name} has no {arg!r} argument to guard") +class BinaryPolicy(unittest.TestCase): + """The program the server launches is not the client's to choose (F063). + + "binary" used to be exempt from every check as "configuration", but it + is a per-call argument: any executable file it named was run, with the + client's own paths as arguments. + """ + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + base = Path(self.tmp.name) + self.root = base / "workspace" + self.root.mkdir() + (self.root / "ok.qet").write_text("") + self.qet = fake_qet(base / "bin") + self.other = fake_qet(base / "other") + self._saved = dict(os.environ) + os.environ["QET_MCP_WORKSPACE"] = str(self.root) + os.environ["QET_BINARY"] = str(self.qet) + # A PATH with no qelectrotech on it, so only what a test sets counts. + os.environ["PATH"] = str(base / "empty") + for var in ("QET_MCP_ALLOW_ANY_PATH", "QET_MCP_ALLOW_ANY_BINARY", + "QET_MCP_BINARIES", "QET_MCP_ELEMENTS"): + os.environ.pop(var, None) + + def tearDown(self): + os.environ.clear() + os.environ.update(self._saved) + self.tmp.cleanup() + + def call(self, **extra): + args = dict({"project": str(self.root / "ok.qet")}, **extra) + m.enforce_path_policy("qet_query", args) + return args + + def test_left_out_it_is_filled_in(self): + self.assertEqual(self.call()["binary"], str(self.qet.resolve())) + + def test_the_configured_one_is_accepted(self): + self.assertEqual(self.call(binary=str(self.qet))["binary"], str(self.qet.resolve())) + + def test_any_other_program_is_refused(self): + with self.assertRaisesRegex(ValueError, "not an allowed QElectroTech"): + self.call(binary=str(self.other)) + + def test_a_program_inside_the_workspace_is_refused_too(self): + """Being inside the workspace makes a file readable, not runnable.""" + planted = fake_qet(self.root, "run.sh") + with self.assertRaisesRegex(ValueError, "not an allowed QElectroTech"): + self.call(binary=str(planted)) + + def test_a_symlink_is_judged_by_its_target(self): + link = self.root / "qelectrotech" + link.symlink_to(self.other) + with self.assertRaisesRegex(ValueError, "not an allowed QElectroTech"): + self.call(binary=str(link)) + good = self.root / "also-qet" + good.symlink_to(self.qet) + self.call(binary=str(good)) + + def test_others_can_be_listed(self): + os.environ["QET_MCP_BINARIES"] = str(self.other) + self.assertEqual(self.call(binary=str(self.other))["binary"], str(self.other.resolve())) + + def test_the_check_can_be_switched_off(self): + os.environ["QET_MCP_ALLOW_ANY_BINARY"] = "1" + self.assertEqual(self.call(binary=str(self.other))["binary"], str(self.other)) + + def test_found_on_path(self): + del os.environ["QET_BINARY"] + os.environ["PATH"] = str(self.qet.parent) + self.assertEqual(self.call()["binary"], str(self.qet.resolve())) + + def test_none_found_says_what_to_set(self): + del os.environ["QET_BINARY"] + with self.assertRaisesRegex(ValueError, "set QET_BINARY"): + self.call() + + def test_every_tool_that_takes_binary_is_checked(self): + """A tool whose schema offers "binary" but that the policy skips + would run whatever it was given.""" + takes = {t["name"] for t in m.TOOLS + if "binary" in t["inputSchema"].get("properties", {})} + self.assertEqual(takes, m._LAUNCHES_QET) + for name in takes: + self.assertNotIn("binary", m._BY_NAME[name]["inputSchema"].get("required", []), + f"{name} still requires the client to name a binary") + + def test_elements_dir_outside_the_workspace_is_refused(self): + with self.assertRaisesRegex(ValueError, "'elements_dir' is outside"): + self.call(elements_dir=self.tmp.name) + + def test_elements_dir_can_be_listed(self): + coll = Path(self.tmp.name) / "collection" + coll.mkdir() + os.environ["QET_MCP_ELEMENTS"] = str(coll) + self.call(elements_dir=str(coll / "10_electric")) + + def test_an_install_finds_its_own_qet_and_collection(self): + """Installed as /share/qelectrotech/mcp/qet_mcp.py, the + server needs no configuration at all.""" + import importlib.util + prefix = Path(self.tmp.name) / "prefix" + mcp = prefix / "share" / "qelectrotech" / "mcp" + mcp.mkdir(parents=True) + (prefix / "share" / "qelectrotech" / "elements").mkdir() + shutil.copy2(HERE / "qet_mcp.py", mcp / "qet_mcp.py") + exe = fake_qet(prefix / "bin") + del os.environ["QET_BINARY"] + spec = importlib.util.spec_from_file_location("qet_mcp_installed", mcp / "qet_mcp.py") + inst = importlib.util.module_from_spec(spec) + spec.loader.exec_module(inst) + args = {"project": str(self.root / "ok.qet")} + inst.enforce_path_policy("qet_query", args) + self.assertEqual(args["binary"], str(exe.resolve())) + self.assertEqual(args["elements_dir"], + str((prefix / "share" / "qelectrotech" / "elements").resolve())) + + class LaunchExecutable(unittest.TestCase): """Windows cannot run a lone copy of QElectroTech (F065): its DLLs sit beside the original. Everywhere else the private copy stays.""" @@ -2075,6 +2736,24 @@ class PathPolicyOverStdio(unittest.TestCase): self.assertTrue(result.get("isError"), result) self.assertIn("outside the workspace", result["content"][0]["text"]) + def test_a_tool_call_cannot_choose_the_program_to_run(self): + """F063's reproduction: a script in the workspace, named as binary.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + (root / "a.qet").write_text('') + planted = fake_qet(root, "argv.py") + reply = self.rpc({"jsonrpc": "2.0", "id": 9, "method": "tools/call", + "params": {"name": "qet_export", + "arguments": {"binary": str(planted), + "project": str(root / "a.qet"), + "format": "pdf", + "output": str(root / "o.pdf")}}}, + {"QET_MCP_WORKSPACE": str(root), + "QET_BINARY": str(fake_qet(root / "bin"))}) + result = reply["result"] + self.assertTrue(result.get("isError"), result) + self.assertIn("not an allowed QElectroTech", result["content"][0]["text"]) + def test_the_same_call_succeeds_inside_the_workspace(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) / "ws" @@ -2394,6 +3073,19 @@ class Integration(unittest.TestCase): note = r["operations"][1].get("note", "") self.assertIn("from_terminal: no terminal", note) self.assertIn("to_terminal: no terminal", note) + def test_folio_ref_survives_an_insert_before_it(self): + """add_folio names a folio "$f"; inserting another at position 0 + moves it from index 1 to 2. "$f" must still name it.""" + base = self.sb.new() + r = self.ok(self.sb.edit(base, [ + {"op": "add_folio", "id": "f"}, + {"op": "insert_folio", "position": 0}, + {"op": "set_folio_title", "folio": "$f", "title": "MINE"}])) + self.assertEqual(r["operations"][0]["result"], 1) + titles = [f["title"] for f in m.tool_project_info(r["output"])["folios"]] + self.assertEqual(len(titles), 3) + self.assertEqual(titles[2], "MINE", titles) + self.assertNotIn("MINE", titles[:2]) def test_noop_edit_has_no_conductor_churn(self): """Re-saving renumbers the file's terminal ids; the diff must not