mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-09-21 18:36:47 +00:00
fix(cli): read json checkpoints as utf-8 (#7491)
JsonProvider writes checkpoints with encoding="utf-8" and the runtime serialises non-ASCII text verbatim, but the checkpoint CLI still opened them with the platform default encoding. On Windows (cp1252) `crewai checkpoint info` raised UnicodeDecodeError and `list` showed a 0-byte entry for any checkpoint containing non-ASCII text. Open the files as UTF-8 and add regression tests for the three readers.
This commit is contained in:
@@ -246,7 +246,7 @@ def _list_json(location: str) -> list[dict[str, Any]]:
|
||||
):
|
||||
name = os.path.basename(path)
|
||||
try:
|
||||
with open(path) as f:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
raw = f.read()
|
||||
meta = _parse_checkpoint_json(raw, source=name)
|
||||
meta["name"] = name
|
||||
@@ -267,7 +267,7 @@ def _info_json_latest(location: str) -> dict[str, Any] | None:
|
||||
if not files:
|
||||
return None
|
||||
path = files[0]
|
||||
with open(path) as f:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
raw = f.read()
|
||||
meta = _parse_checkpoint_json(raw, source=os.path.basename(path))
|
||||
meta["name"] = os.path.basename(path)
|
||||
@@ -278,7 +278,7 @@ def _info_json_latest(location: str) -> dict[str, Any] | None:
|
||||
|
||||
|
||||
def _info_json_file(path: str) -> dict[str, Any]:
|
||||
with open(path) as f:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
raw = f.read()
|
||||
meta = _parse_checkpoint_json(raw, source=os.path.basename(path))
|
||||
meta["name"] = os.path.basename(path)
|
||||
|
||||
@@ -13,6 +13,9 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from crewai_cli.checkpoint_cli import (
|
||||
_info_json_file,
|
||||
_info_json_latest,
|
||||
_list_json,
|
||||
_parse_checkpoint_json,
|
||||
_parse_duration,
|
||||
_prune_json,
|
||||
@@ -23,6 +26,7 @@ from crewai_cli.checkpoint_cli import (
|
||||
prune_checkpoints,
|
||||
resume_checkpoint,
|
||||
)
|
||||
from crewai.state.provider.json_provider import JsonProvider
|
||||
|
||||
|
||||
def _make_checkpoint_data(
|
||||
@@ -198,6 +202,46 @@ class TestResolveCheckpoint:
|
||||
assert _resolve_checkpoint("/nonexistent/path", None) is None
|
||||
|
||||
|
||||
class TestNonAsciiJsonCheckpoint:
|
||||
"""JSON checkpoints are UTF-8 on disk; the CLI readers must decode them as such.
|
||||
|
||||
``JsonProvider`` writes checkpoints with ``encoding="utf-8"`` and the runtime
|
||||
serialises non-ASCII text verbatim, so readers that rely on the platform
|
||||
default encoding break on Windows (cp1252) for any non-ASCII checkpoint.
|
||||
"""
|
||||
|
||||
# "Đ" (U+0110) encodes to 0xC4 0x90; 0x90 is undefined in cp1252, so a
|
||||
# locale-dependent read fails loudly instead of silently producing mojibake.
|
||||
_NAME = "Đội ngũ phân tích"
|
||||
|
||||
def _write_checkpoint(self, base_dir: str) -> str:
|
||||
"""Write a UTF-8 checkpoint whose entity name is non-ASCII, as the runtime does."""
|
||||
data = json.loads(_make_checkpoint_data(name=self._NAME))
|
||||
raw = json.dumps(data, ensure_ascii=False)
|
||||
return JsonProvider().checkpoint(raw, base_dir, branch="main")
|
||||
|
||||
def test_info_json_file_reads_utf8(self, tmp_path: Any) -> None:
|
||||
"""``_info_json_file`` decodes a non-ASCII checkpoint on any platform."""
|
||||
path = self._write_checkpoint(str(tmp_path))
|
||||
meta = _info_json_file(path)
|
||||
assert meta["entities"][0]["name"] == self._NAME
|
||||
|
||||
def test_info_json_latest_reads_utf8(self, tmp_path: Any) -> None:
|
||||
"""``_info_json_latest`` decodes the newest non-ASCII checkpoint."""
|
||||
self._write_checkpoint(str(tmp_path))
|
||||
meta = _info_json_latest(str(tmp_path))
|
||||
assert meta is not None
|
||||
assert meta["entities"][0]["name"] == self._NAME
|
||||
|
||||
def test_list_json_reads_utf8(self, tmp_path: Any) -> None:
|
||||
"""``_list_json`` lists a non-ASCII checkpoint with its real size and entities."""
|
||||
self._write_checkpoint(str(tmp_path))
|
||||
results = _list_json(str(tmp_path))
|
||||
assert len(results) == 1
|
||||
assert results[0]["size"] > 0
|
||||
assert results[0]["entities"][0]["name"] == self._NAME
|
||||
|
||||
|
||||
class TestTaskListFromMeta:
|
||||
def test_flattens_tasks(self) -> None:
|
||||
data = _make_checkpoint_data(tasks_completed=2, tasks_total=3)
|
||||
|
||||
Reference in New Issue
Block a user