Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1931,8 +1931,8 @@ def _read_integration_json(project_root: Path) -> dict[str, Any]:
return {}
try:
data = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
console.print(f"[red]Error:[/red] {path} contains invalid JSON.")
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
console.print(f"[red]Error:[/red] {path} contains invalid JSON or is not valid UTF-8.")
console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
console.print(f"[dim]Details:[/dim] {exc}")
raise typer.Exit(1)
Expand Down
99 changes: 98 additions & 1 deletion src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,12 @@

import yaml

from ..integration_state import (
INTEGRATION_JSON,
INTEGRATION_STATE_SCHEMA,
default_integration_key,
normalize_integration_state,
)
from .base import RunStatus, StepContext, StepResult, StepStatus


Expand Down Expand Up @@ -143,6 +149,35 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
f"Must be 'string', 'number', or 'boolean'."
)

# Validate the default eagerly so authoring mistakes (e.g. a
# default not in the declared enum, or a non-numeric default for
# a number input) surface at install/validation time instead of
# at workflow-execution time. ``"auto"`` for the integration
# input is a runtime-resolved sentinel, so only the
# enum-membership check is exempted for that exact case — the
# declared type is still enforced (e.g. ``type: number`` paired
# with ``default: "auto"`` is still rejected).
if "default" in input_def:
default_value = input_def["default"]
is_auto_integration = (
input_name == "integration" and default_value == "auto"
)
validation_input_def: dict[str, Any] = input_def
if is_auto_integration and "enum" in input_def:
validation_input_def = {
key: value
for key, value in input_def.items()
if key != "enum"
}
try:
WorkflowEngine._coerce_input(
input_name, default_value, validation_input_def
)
except ValueError as exc:
errors.append(
f"Input {input_name!r} has invalid default: {exc}"
)

# -- Steps ------------------------------------------------------------
if not isinstance(definition.steps, list):
errors.append("'steps' must be a list.")
Expand Down Expand Up @@ -715,12 +750,74 @@ def _resolve_inputs(
name, provided[name], input_def
)
elif "default" in input_def:
resolved[name] = input_def["default"]
default_value = self._resolve_default(name, input_def["default"])
# If the integration default could not be resolved against
# project state and falls back to the literal ``"auto"``
# sentinel, exempt it from enum-membership coercion so a
# workflow that lists specific integrations in ``enum`` does
# not crash at runtime — declared type is still enforced.
coerce_input_def = input_def
if (
name == "integration"
and default_value == "auto"
and "enum" in input_def
):
coerce_input_def = {
key: value
for key, value in input_def.items()
if key != "enum"
}
resolved[name] = self._coerce_input(
name, default_value, coerce_input_def
)
elif input_def.get("required", False):
Comment thread
mnriem marked this conversation as resolved.
msg = f"Required input {name!r} not provided."
raise ValueError(msg)
return resolved

def _resolve_default(self, name: str, default: Any) -> Any:
"""Resolve special default sentinels against project state.

For the ``integration`` input, ``"auto"`` resolves to the integration
recorded in ``.specify/integration.json`` so workflows dispatch to the
AI the project was actually initialized with, instead of a hardcoded
value baked into the workflow YAML.
"""
if name == "integration" and default == "auto":
resolved = self._load_project_integration()
if resolved is not None:
return resolved
return default

def _load_project_integration(self) -> str | None:
"""Read the default integration key from ``.specify/integration.json``.

Honors the same schema guard as ``_read_integration_json`` (rejects
files whose ``integration_state_schema`` is newer than this CLI
supports) and reads the canonical normalized state, so modern
installs that record ``default_integration`` / ``installed_integrations``
resolve correctly under ``integration: auto``. Returns None when the
file is missing, malformed, or written by a newer CLI; callers are
expected to fall back to a literal default.
"""
path = self.project_root / INTEGRATION_JSON
if not path.is_file():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError, UnicodeDecodeError):
return None
Comment on lines +807 to +809
if not isinstance(data, dict):
return None
schema = data.get("integration_state_schema")
if (
isinstance(schema, int)
and not isinstance(schema, bool)
and schema > INTEGRATION_STATE_SCHEMA
):
return None
return default_integration_key(normalize_integration_state(data))

@staticmethod
def _coerce_input(
name: str, value: Any, input_def: dict[str, Any]
Expand Down
24 changes: 24 additions & 0 deletions tests/integrations/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -1191,6 +1191,30 @@ def fail_search(self, **kwargs):
assert "contains invalid JSON" in normalized_output
assert "integration.json" in normalized_output

def test_search_rejects_non_utf8_integration_json_before_catalog_lookup(
self, tmp_path, monkeypatch
):
"""A non-UTF8 ``integration.json`` must surface a clear error and
avoid falling through to the catalog lookup, mirroring the malformed-JSON
case but for the ``UnicodeDecodeError`` branch in ``_read_integration_json``."""
project = self._make_project(tmp_path)
# 0xFF is invalid as the leading byte of any UTF-8 sequence, so
# ``Path.read_text(encoding="utf-8")`` raises ``UnicodeDecodeError``.
(project / ".specify" / "integration.json").write_bytes(b"\xff\xfe\x00\x00")

from specify_cli.integrations.catalog import IntegrationCatalog

def fail_search(self, **kwargs):
raise AssertionError("catalog search should not be called")

monkeypatch.setattr(IntegrationCatalog, "search", fail_search)

result = self._invoke(["integration", "search"], project)
normalized_output = _normalize_cli_output(result.output)
assert result.exit_code == 1
assert "not valid UTF-8" in normalized_output
assert "integration.json" in normalized_output

def test_search_filters_by_tag(self, tmp_path, monkeypatch):
project = self._make_project(tmp_path)
self._patch_catalog(monkeypatch)
Expand Down
Loading