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 anton/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -345,9 +345,9 @@ def _ensure_terms_consent(console: Console, settings) -> None:
env_path.parent.mkdir(parents=True, exist_ok=True)

# Append if file exists, otherwise create
existing = env_path.read_text() if env_path.is_file() else ""
existing = env_path.read_text(encoding="utf-8") if env_path.is_file() else ""
if "ANTON_TERMS_CONSENT" not in existing:
with env_path.open("a") as f:
with env_path.open("a", encoding="utf-8") as f:
if existing and not existing.endswith("\n"):
f.write("\n")
f.write("ANTON_TERMS_CONSENT=true\n")
Expand Down
17 changes: 9 additions & 8 deletions anton/workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -120,13 +120,14 @@ def _initialize_once(self, *, create_anton_md: bool = True) -> list[str]:
# Create anton.md if it doesn't exist
if create_anton_md and not self._anton_md.is_file():
self._anton_md.write_text(
ANTON_MD_TEMPLATE.format(date=datetime.now().strftime("%Y-%m-%d"))
ANTON_MD_TEMPLATE.format(date=datetime.now().strftime("%Y-%m-%d")),
encoding="utf-8",
)
actions.append(f"Created {self._anton_md}")

# Create .env if it doesn't exist
if not self._env_file.is_file():
self._env_file.write_text("# Anton environment variables\n")
self._env_file.write_text("# Anton environment variables\n", encoding="utf-8")
actions.append(f"Created {self._env_file}")

# Visible artifacts directory at the workspace root. Replaces
Expand All @@ -150,7 +151,7 @@ def read_anton_md(self) -> str | None:
"""Read anton.md content. Returns None if it doesn't exist."""
if not self._anton_md.is_file():
return None
return self._anton_md.read_text()
return self._anton_md.read_text(encoding="utf-8")

def anton_md_modified_since_last_read(self) -> bool:
"""Check if anton.md has been modified since last read_anton_md_tracked()."""
Expand Down Expand Up @@ -187,7 +188,7 @@ def load_env(self) -> dict[str, str]:
result: dict[str, str] = {}
if not self._env_file.is_file():
return result
for line in self._env_file.read_text().splitlines():
for line in self._env_file.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
Expand Down Expand Up @@ -217,7 +218,7 @@ def set_secret(self, key: str, value: str) -> None:
lines: list[str] = []
replaced = False
if self._env_file.is_file():
for line in self._env_file.read_text().splitlines():
for line in self._env_file.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#") and "=" in stripped:
existing_key = stripped.partition("=")[0].strip()
Expand All @@ -230,7 +231,7 @@ def set_secret(self, key: str, value: str) -> None:
if not replaced:
lines.append(f"{key}={value}")

self._env_file.write_text("\n".join(lines) + "\n")
self._env_file.write_text("\n".join(lines) + "\n", encoding="utf-8")

# Also set in current process environment
os.environ[key] = value
Expand All @@ -245,7 +246,7 @@ def remove_secret(self, key: str) -> bool:

lines: list[str] = []
found = False
for line in self._env_file.read_text().splitlines():
for line in self._env_file.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#") and "=" in stripped:
existing_key = stripped.partition("=")[0].strip()
Expand All @@ -255,7 +256,7 @@ def remove_secret(self, key: str) -> bool:
lines.append(line)

if found:
self._env_file.write_text("\n".join(lines) + "\n")
self._env_file.write_text("\n".join(lines) + "\n", encoding="utf-8")
os.environ.pop(key, None)

return found
Expand Down
74 changes: 74 additions & 0 deletions tests/test_workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -247,3 +247,77 @@ def test_remove_secret_pops_environ(self, ws, tmp_path):
assert os.environ.get("ANTON_TEST_REMOVE_XYZ") == "val"
ws.remove_secret("ANTON_TEST_REMOVE_XYZ")
assert os.environ.get("ANTON_TEST_REMOVE_XYZ") is None


class TestSecretVaultEncoding:
# AntonSettings reads .anton/.env with env_file_encoding="utf-8"
# (anton/config/settings.py). The vault must therefore be written as
# UTF-8, not in the host locale: on a Windows code page (cp1252) a bare
# write_text() stores a non-ASCII secret as bytes the settings loader
# cannot decode, so every later AntonSettings() raises UnicodeDecodeError.

def test_non_ascii_secret_is_stored_as_utf8(self, ws):
ws.initialize()
ws.set_secret("ANTON_MINDS_MIND_NAME", "café_mind")

raw = ws.env_path.read_bytes()
assert "café_mind" in raw.decode("utf-8")

def test_secret_outside_the_host_code_page_is_storable(self, ws):
# 密钥 has no cp1252 representation, so a bare write_text() raises
# UnicodeEncodeError and the secret is never stored at all.
ws.initialize()
ws.set_secret("ANTON_MINDS_DATASOURCE", "密钥")
assert ws.get_secret("ANTON_MINDS_DATASOURCE") == "密钥"

def test_settings_can_load_a_vault_holding_a_non_ascii_secret(self, ws):
from anton.config.settings import AntonSettings

ws.initialize()
ws.set_secret("ANTON_MINDS_MIND_NAME", "café_mind")
assert AntonSettings(_env_file=str(ws.env_path)).minds_mind_name == "café_mind"

def test_anton_md_is_read_as_utf8(self, ws):
ws.initialize()
ws.anton_md_path.write_text("café", encoding="utf-8")
assert ws.read_anton_md() == "café"

def test_workspace_file_io_never_relies_on_the_host_locale(self, tmp_path):
# PEP 597: under `-X warn_default_encoding` every locale-default text
# open() emits an EncodingWarning. Asserting that none names
# workspace.py keeps this regression visible on UTF-8 CI too, where
# the behavioural tests above pass whether or not the bug is present.
import subprocess
import sys

import anton.workspace as workspace_mod

repo_root = Path(workspace_mod.__file__).resolve().parent.parent
probe = tmp_path / "probe.py"
probe.write_text(
"import tempfile, warnings\n"
"from pathlib import Path\n"
"from anton.workspace import Workspace\n"
"with warnings.catch_warnings(record=True) as caught:\n"
" warnings.simplefilter('always')\n"
" ws = Workspace(Path(tempfile.mkdtemp()))\n"
" ws.initialize()\n"
" ws.set_secret('K', 'v')\n"
" ws.load_env()\n"
" ws.read_anton_md()\n"
" ws.remove_secret('K')\n"
"for w in caught:\n"
" if w.category is EncodingWarning:\n"
" print(f'{w.filename}:{w.lineno}')\n",
encoding="utf-8",
)
proc = subprocess.run(
[sys.executable, "-X", "warn_default_encoding", str(probe)],
capture_output=True,
text=True,
env={**os.environ, "PYTHONPATH": str(repo_root)},
)

assert proc.returncode == 0, proc.stderr
offenders = [ln for ln in proc.stdout.splitlines() if "workspace.py" in ln]
assert offenders == [], f"locale-default text I/O in workspace.py: {offenders}"
Loading