feat(infra): xong R02 — Settings Facade, versioning, chuyển khoá sang keyring
R02-T03 Typed Settings Facade
Khắp nơi đang viết ctx.config.routing.get("switch_mode", "off"). Gõ sai một
chữ thì lặng lẽ nhận mặc định, không ai biết cho tới lúc tính năng "không
hiểu sao không chạy". ProviderSettings / RoutingSettings / SecuritySettings
làm sai tên là lỗi ngay, và kiểu ghi rõ nên đọc là biết confirm_timeout_sec
tính bằng giây.
Là KHUNG NHÌN lên dict sống, không phải dataclass sao chép — sửa qua đây là
sửa vào cấu hình, save() là xuống đĩa, khỏi sinh chuyện đồng bộ hai chiều.
Có raw() để ai thiếu thuộc tính thì dùng tạm, đừng vòng lại config.data.
Bắt cả trường hợp giá trị là null: file cũ hay để null, đọc ra None rồi đem
so sánh số là vỡ.
R02-T06 Schema versioning + phục hồi
config.json hôm nay không có số phiên bản, nên mọi thay đổi hình dạng phải
đoán — _migrate_connectors() đoán "có khoá office nghĩa là file cũ". Giờ:
thiếu schema_version thì coi là v1, mỗi bước là một hàm chạy tuần tự, sao
lưu trước khi nâng, và file mới hơn app thì dùng nguyên trạng chứ không đoán
ngược.
R02-T05 Chuyển API key sang kho bí mật
Là bước v1→v2. Người dùng cập nhật app, mở lên, khoá cũ tự vào keyring và
biến khỏi đĩa — có test cho đúng cảnh đó.
Hai chỗ cố tình không làm:
- Máy chưa có keyring: KHÔNG chuyển, giữ nguyên v1. Thà để khoá trong file
còn hơn xoá đi rồi người dùng mất khoá mà không hiểu vì sao.
- Giá trị "ollama" là bù nhìn (Ollama đòi có api_key nhưng bỏ qua nội
dung), đẩy vào keyring chỉ tổ rác.
Hai chuỗi test trông giống khoá thật bị CASAN Check 1 bắt — đánh dấu
"# casan: allow" kèm lý do, đúng lối thoát đã thiết kế cho cả đội.
150 test xanh (129 + 21 mới). CASAN Check 1 sạch. File mới đều dưới 200 dòng.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
a7e369e46c
commit
ab0d26761f
@@ -22,6 +22,7 @@ from typing import Any, Dict
|
||||
|
||||
from ..persistence.json.atomic_json_file import AtomicJsonFile
|
||||
from ..secrets.secret_store import SecretStore, provider_key
|
||||
from .schema_migration import CURRENT_VERSION, migrate
|
||||
|
||||
|
||||
class JsonConfigRepository:
|
||||
@@ -51,7 +52,14 @@ class JsonConfigRepository:
|
||||
merged = copy.deepcopy(self._defaults)
|
||||
stored = self._file.read(default=None)
|
||||
if isinstance(stored, dict):
|
||||
# Nâng cấp TRƯỚC khi trộn với mặc định: bước v1→v2 gỡ api_key khỏi
|
||||
# đĩa, mà mặc định thì không có khoá nào để gỡ.
|
||||
stored, changed = migrate(stored, secrets=self._secrets,
|
||||
path=self._file.path)
|
||||
merged = _deep_merge(merged, stored)
|
||||
if changed:
|
||||
self.data = merged
|
||||
self.save() # ghi ngay, để lần sau khỏi chuyển lại
|
||||
merged = self._env_overrides(merged)
|
||||
# Trạng thái mở khoá ms365 chỉ tồn tại lúc chạy — mỗi lần mở app đều
|
||||
# bắt đầu ở trạng thái khoá, không tin giá trị đọc từ đĩa.
|
||||
@@ -170,6 +178,7 @@ class JsonConfigRepository:
|
||||
to_write = copy.deepcopy(self.data)
|
||||
to_write["ms365"]["unlocked"] = False
|
||||
to_write.pop("_project_history_dir", None)
|
||||
to_write["schema_version"] = CURRENT_VERSION
|
||||
self._file.write(to_write)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Đánh số phiên bản và chuyển đổi cấu hình — R02-T06.
|
||||
|
||||
Hôm nay ``config.json`` không có số phiên bản. Nghĩa là không có cách nào biết
|
||||
file trên đĩa thuộc thời nào, và mọi thay đổi hình dạng phải xử lý bằng cách
|
||||
đoán — ``config.py::_migrate_connectors()`` chính là một ví dụ: nó đoán "có
|
||||
khoá ``office`` nghĩa là file cũ".
|
||||
|
||||
Ở đây đặt luật rõ:
|
||||
|
||||
* File có ``schema_version``. Thiếu ⇒ coi là **1** (mọi file đang tồn tại).
|
||||
* Mỗi bước nâng cấp là một hàm ``v1 -> v2``, chạy tuần tự, không nhảy cóc.
|
||||
* **Sao lưu trước khi nâng cấp.** Người dùng lùi về bản app cũ thì bản cũ đọc
|
||||
file mới có thể hỏng — phải còn đường về.
|
||||
* Chỉ nâng, không hạ. File mới hơn app thì báo và dùng nguyên trạng, không cố
|
||||
đoán ngược.
|
||||
|
||||
Bước v1→v2 đầu tiên đi kèm R02-T05: gỡ ``api_key`` khỏi đĩa, đẩy vào
|
||||
``SecretStore``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import logging
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict
|
||||
|
||||
from ..secrets.secret_store import SecretStore, provider_key
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
#: Phiên bản app hiện đang ghi ra.
|
||||
CURRENT_VERSION = 2
|
||||
|
||||
#: Thiếu ``schema_version`` ⇒ file có từ trước khi đánh số.
|
||||
ASSUMED_VERSION = 1
|
||||
|
||||
|
||||
def read_version(data: Dict[str, Any]) -> int:
|
||||
try:
|
||||
return int(data.get("schema_version", ASSUMED_VERSION))
|
||||
except (TypeError, ValueError):
|
||||
return ASSUMED_VERSION
|
||||
|
||||
|
||||
def _v1_to_v2(data: Dict[str, Any], secrets: SecretStore | None) -> Dict[str, Any]:
|
||||
"""Chuyển API key từ file sang kho bí mật — R02-T05.
|
||||
|
||||
Không có kho bí mật thì **không chuyển**: thà để khoá nằm nguyên trong file
|
||||
còn hơn xoá đi rồi người dùng mất khoá mà không hiểu vì sao. File giữ
|
||||
nguyên phiên bản 1, lần chạy sau trên máy có keyring sẽ chuyển.
|
||||
"""
|
||||
if secrets is None or not getattr(secrets, "available", True):
|
||||
log.info("bỏ qua v1→v2: máy này chưa có kho bí mật dùng được")
|
||||
return data
|
||||
|
||||
out = copy.deepcopy(data)
|
||||
moved = []
|
||||
for name, conf in (out.get("providers") or {}).items():
|
||||
if not isinstance(conf, dict):
|
||||
continue
|
||||
key = (conf.get("api_key") or "").strip()
|
||||
# "ollama" là giá trị bù nhìn — Ollama đòi có api_key nhưng bỏ qua nội
|
||||
# dung. Đẩy nó vào keyring chỉ tổ rác.
|
||||
if not key or key == "ollama":
|
||||
continue
|
||||
secrets.set(provider_key(name), key)
|
||||
conf["api_key"] = ""
|
||||
moved.append(name)
|
||||
|
||||
out["schema_version"] = 2
|
||||
if moved:
|
||||
log.info("đã chuyển API key sang kho bí mật: %s", ", ".join(moved))
|
||||
return out
|
||||
|
||||
|
||||
#: {phiên bản nguồn: hàm nâng lên phiên bản kế tiếp}
|
||||
STEPS: Dict[int, Callable[[Dict[str, Any], SecretStore | None], Dict[str, Any]]] = {
|
||||
1: _v1_to_v2,
|
||||
}
|
||||
|
||||
|
||||
def backup(path: Path) -> Path | None:
|
||||
"""Chép file trước khi nâng cấp. Trả về đường dẫn bản sao."""
|
||||
if not path.exists():
|
||||
return None
|
||||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
target = path.with_suffix(path.suffix + f".v{stamp}.bak")
|
||||
try:
|
||||
shutil.copy2(path, target)
|
||||
return target
|
||||
except OSError as exc:
|
||||
log.warning("không sao lưu được %s: %s", path, exc)
|
||||
return None
|
||||
|
||||
|
||||
def migrate(data: Dict[str, Any], *, secrets: SecretStore | None = None,
|
||||
path: Path | None = None) -> tuple[Dict[str, Any], bool]:
|
||||
"""Nâng ``data`` lên :data:`CURRENT_VERSION`.
|
||||
|
||||
Trả về ``(dữ_liệu, có_đổi_không)``. ``có_đổi_không`` là False thì chỗ gọi
|
||||
khỏi phải ghi lại đĩa.
|
||||
"""
|
||||
version = read_version(data)
|
||||
|
||||
if version > CURRENT_VERSION:
|
||||
# App cũ gặp file mới. Đoán ngược là cách nhanh nhất để mất dữ liệu.
|
||||
log.warning("config phiên bản %s mới hơn app (%s) — dùng nguyên trạng",
|
||||
version, CURRENT_VERSION)
|
||||
return data, False
|
||||
|
||||
if version == CURRENT_VERSION:
|
||||
return data, False
|
||||
|
||||
if path is not None:
|
||||
backup(path)
|
||||
|
||||
changed = False
|
||||
while version < CURRENT_VERSION:
|
||||
step = STEPS.get(version)
|
||||
if step is None:
|
||||
log.warning("thiếu bước nâng cấp từ phiên bản %s — dừng", version)
|
||||
break
|
||||
data = step(data, secrets)
|
||||
new_version = read_version(data)
|
||||
if new_version <= version:
|
||||
# Bước không nâng được phiên bản (ví dụ v1→v2 bỏ qua vì chưa có
|
||||
# keyring). Dừng, đừng lặp vô hạn.
|
||||
break
|
||||
version = new_version
|
||||
changed = True
|
||||
|
||||
return data, changed
|
||||
@@ -0,0 +1,178 @@
|
||||
"""Khung nhìn có kiểu cho từng nhóm cấu hình — R02-T03.
|
||||
|
||||
Vấn đề đang có: khắp nơi viết ``ctx.config.routing.get("switch_mode", "off")``.
|
||||
Gõ sai một chữ thì lặng lẽ nhận giá trị mặc định, không ai biết cho tới khi
|
||||
tính năng "không hiểu sao không chạy". Đếm được **156 lời gọi ``ctx.config.*``
|
||||
trong 29 file** kiểu đó.
|
||||
|
||||
Ở đây mỗi nhóm cấu hình có một lớp: gõ sai tên thuộc tính là lỗi ngay, và kiểu
|
||||
dữ liệu ghi rõ ràng nên đọc code là biết ``confirm_timeout_sec`` là số giây
|
||||
chứ không phải mili giây.
|
||||
|
||||
Cố ý KHÔNG dùng dataclass đông cứng: đây là *khung nhìn* lên dict cấu hình
|
||||
sống, sửa qua đây là sửa vào dict rồi ``save()`` là xuống đĩa. Sao chép thành
|
||||
dataclass thì lại sinh chuyện đồng bộ hai chiều.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
|
||||
class _View:
|
||||
"""Khung nhìn lên một nhánh của dict cấu hình."""
|
||||
|
||||
def __init__(self, data: Dict[str, Any]):
|
||||
self._d = data
|
||||
|
||||
def _get(self, key: str, default: Any) -> Any:
|
||||
value = self._d.get(key, default)
|
||||
return default if value is None else value
|
||||
|
||||
def raw(self) -> Dict[str, Any]:
|
||||
"""Dict gốc — dùng khi cần đọc khoá chưa được đưa vào khung nhìn.
|
||||
|
||||
Có mặt để không ai bị kẹt: thiếu thuộc tính thì dùng tạm ``raw()`` rồi
|
||||
mở issue bổ sung, chứ đừng vòng lại ``ctx.config.data``.
|
||||
"""
|
||||
return self._d
|
||||
|
||||
|
||||
class ProviderSettings(_View):
|
||||
"""Một provider: đi đâu, model nào, khoá nào.
|
||||
|
||||
``api_key`` ở đây là thứ ``JsonConfigRepository.provider_conf()`` đã ghép
|
||||
sẵn từ kho bí mật — xem đường A trong ``GammaTeam_decisions.md``.
|
||||
"""
|
||||
|
||||
@property
|
||||
def base_url(self) -> str:
|
||||
return str(self._get("base_url", ""))
|
||||
|
||||
@property
|
||||
def model(self) -> str:
|
||||
return str(self._get("model", ""))
|
||||
|
||||
@property
|
||||
def api_key(self) -> str:
|
||||
return str(self._get("api_key", ""))
|
||||
|
||||
@property
|
||||
def configured(self) -> bool:
|
||||
"""Đủ thông tin để gọi được chưa.
|
||||
|
||||
Ollama chạy cục bộ nên không cần khoá — đó là lý do điều kiện là
|
||||
"có base_url và model", không phải "có api_key".
|
||||
"""
|
||||
return bool(self.base_url and self.model)
|
||||
|
||||
|
||||
class RoutingSettings(_View):
|
||||
"""Định tuyến model tự động (``core/routing/``)."""
|
||||
|
||||
@property
|
||||
def switch_mode(self) -> str:
|
||||
"""``"off"`` | ``"auto"`` | ``"manual"``."""
|
||||
return str(self._get("switch_mode", "off"))
|
||||
|
||||
@switch_mode.setter
|
||||
def switch_mode(self, value: str) -> None:
|
||||
self._d["switch_mode"] = value
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
return self.switch_mode != "off"
|
||||
|
||||
@property
|
||||
def policy(self) -> str:
|
||||
"""``"balanced"`` | ``"cheap"`` | ``"quality"``…"""
|
||||
return str(self._get("policy", "balanced"))
|
||||
|
||||
@property
|
||||
def min_score_gain(self) -> float:
|
||||
"""Phải hơn model hiện tại bao nhiêu điểm mới đáng đổi."""
|
||||
return float(self._get("min_score_gain", 0.05))
|
||||
|
||||
@property
|
||||
def confirm_timeout_sec(self) -> int:
|
||||
"""GIÂY, không phải mili giây — đọc tên là biết, khỏi phải mò."""
|
||||
return int(self._get("confirm_timeout_sec", 60))
|
||||
|
||||
@property
|
||||
def reassess_interval_hours(self) -> int:
|
||||
return int(self._get("reassess_interval_hours", 24))
|
||||
|
||||
@property
|
||||
def per_provider_concurrency(self) -> int:
|
||||
return int(self._get("per_provider_concurrency", 2))
|
||||
|
||||
@property
|
||||
def judge_provider(self) -> str:
|
||||
return str(self._get("judge_provider", ""))
|
||||
|
||||
@property
|
||||
def judge_model(self) -> str:
|
||||
return str(self._get("judge_model", ""))
|
||||
|
||||
|
||||
class SecuritySettings(_View):
|
||||
"""Chính sách an toàn cho agent (``core/agent_security.py``)."""
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
return bool(self._get("enabled", True))
|
||||
|
||||
@property
|
||||
def validate_prompt(self) -> bool:
|
||||
return bool(self._get("validate_prompt", True))
|
||||
|
||||
@property
|
||||
def validate_attachments(self) -> bool:
|
||||
return bool(self._get("validate_attachments", True))
|
||||
|
||||
@property
|
||||
def validate_commands(self) -> bool:
|
||||
return bool(self._get("validate_commands", True))
|
||||
|
||||
@property
|
||||
def command_ai_check(self) -> bool:
|
||||
return bool(self._get("command_ai_check", False))
|
||||
|
||||
@property
|
||||
def cowork_confirm_commands(self) -> bool:
|
||||
"""Có hỏi trước khi chạy lệnh không.
|
||||
|
||||
Ứng với ``PolicyOutcome.ASK`` trong
|
||||
``domain/security/tool_policy.py``.
|
||||
"""
|
||||
return bool(self._get("cowork_confirm_commands", True))
|
||||
|
||||
@property
|
||||
def rules_onedrive_url(self) -> str:
|
||||
return str(self._get("rules_onedrive_url", ""))
|
||||
|
||||
@property
|
||||
def admin_email(self) -> str:
|
||||
return str(self._get("admin_email", ""))
|
||||
|
||||
|
||||
class Settings:
|
||||
"""Cửa vào duy nhất cho các nhóm cấu hình có kiểu.
|
||||
|
||||
>>> s = Settings(repo)
|
||||
>>> if s.routing.enabled and s.provider().configured:
|
||||
... ...
|
||||
"""
|
||||
|
||||
def __init__(self, repo):
|
||||
self._repo = repo
|
||||
|
||||
def provider(self, name: str | None = None) -> ProviderSettings:
|
||||
return ProviderSettings(self._repo.provider_conf(name))
|
||||
|
||||
@property
|
||||
def routing(self) -> RoutingSettings:
|
||||
return RoutingSettings(self._repo.routing)
|
||||
|
||||
@property
|
||||
def security(self) -> SecuritySettings:
|
||||
return SecuritySettings(self._repo.agent_security)
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Đánh số phiên bản + chuyển API key — R02-T06 và R02-T05."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from cowork_local.infrastructure.config.json_config_repository import (
|
||||
JsonConfigRepository,
|
||||
)
|
||||
from cowork_local.infrastructure.config.schema_migration import (
|
||||
CURRENT_VERSION, migrate, read_version,
|
||||
)
|
||||
from cowork_local.tests.fakes.fake_config import FakeSecretStore
|
||||
|
||||
DEFAULTS = {
|
||||
"active_provider": "openai",
|
||||
"providers": {"openai": {"base_url": "u", "model": "m", "api_key": ""},
|
||||
"ollama": {"base_url": "u", "model": "m", "api_key": "ollama"}},
|
||||
"theme": "dark", "language": "vi", "ms365": {},
|
||||
}
|
||||
|
||||
|
||||
def _repo(tmp_path, secrets=None):
|
||||
return JsonConfigRepository(tmp_path / "config.json", secrets=secrets,
|
||||
defaults=DEFAULTS, env_overrides=lambda d: d)
|
||||
|
||||
|
||||
def test_thieu_so_phien_ban_thi_coi_la_v1():
|
||||
assert read_version({}) == 1
|
||||
assert read_version({"schema_version": 2}) == 2
|
||||
assert read_version({"schema_version": "hỏng"}) == 1
|
||||
|
||||
|
||||
def test_v1_sang_v2_chuyen_khoa_vao_kho_bi_mat():
|
||||
secrets = FakeSecretStore()
|
||||
data = {"providers": {"openai": {"api_key": "sk-cu-nam-trong-file"}}} # casan: allow - du lieu test
|
||||
|
||||
out, changed = migrate(data, secrets=secrets)
|
||||
|
||||
assert changed is True
|
||||
assert out["schema_version"] == 2
|
||||
assert out["providers"]["openai"]["api_key"] == ""
|
||||
assert secrets.get("provider:openai") == "sk-cu-nam-trong-file"
|
||||
|
||||
|
||||
def test_khong_day_gia_tri_bu_nhin_cua_ollama_vao_kho():
|
||||
"""Ollama đòi có api_key nhưng bỏ qua nội dung — đẩy vào keyring chỉ tổ rác."""
|
||||
secrets = FakeSecretStore()
|
||||
out, _ = migrate({"providers": {"ollama": {"api_key": "ollama"}}}, secrets=secrets)
|
||||
assert secrets.get("provider:ollama") is None
|
||||
assert out["providers"]["ollama"]["api_key"] == "ollama"
|
||||
|
||||
|
||||
def test_khong_co_kho_bi_mat_thi_KHONG_chuyen():
|
||||
"""Thà để khoá nằm nguyên trong file còn hơn xoá đi rồi người dùng mất
|
||||
khoá mà không hiểu vì sao."""
|
||||
data = {"providers": {"openai": {"api_key": "sk-quy-gia"}}}
|
||||
out, changed = migrate(data, secrets=None)
|
||||
|
||||
assert changed is False
|
||||
assert out["providers"]["openai"]["api_key"] == "sk-quy-gia"
|
||||
assert read_version(out) == 1 # giữ v1, lần sau có keyring sẽ chuyển
|
||||
|
||||
|
||||
def test_da_v2_thi_khong_lam_gi_them():
|
||||
out, changed = migrate({"schema_version": 2}, secrets=FakeSecretStore())
|
||||
assert changed is False
|
||||
|
||||
|
||||
def test_file_moi_hon_app_thi_dung_nguyen_trang():
|
||||
"""App cũ gặp file mới. Đoán ngược là cách nhanh nhất để mất dữ liệu."""
|
||||
data = {"schema_version": 99, "thu_gi_do_tuong_lai": True}
|
||||
out, changed = migrate(data, secrets=FakeSecretStore())
|
||||
assert changed is False
|
||||
assert out == data
|
||||
|
||||
|
||||
def test_sao_luu_truoc_khi_nang_cap(tmp_path):
|
||||
path = tmp_path / "config.json"
|
||||
path.write_text(json.dumps({"providers": {"openai": {"api_key": "sk-x"}}}),
|
||||
encoding="utf-8")
|
||||
|
||||
migrate(json.loads(path.read_text(encoding="utf-8")),
|
||||
secrets=FakeSecretStore(), path=path)
|
||||
|
||||
backups = list(tmp_path.glob("*.bak"))
|
||||
assert len(backups) == 1, "phải có bản sao lưu để còn đường lùi"
|
||||
assert "sk-x" in backups[0].read_text(encoding="utf-8")
|
||||
|
||||
|
||||
# ---- nối vào repository ----------------------------------------------------
|
||||
|
||||
def test_repository_tu_chuyen_khoa_khi_mo_file_cu(tmp_path):
|
||||
"""Cảnh thật: người dùng cập nhật app, mở lên, khoá cũ tự vào keyring."""
|
||||
(tmp_path / "config.json").write_text(
|
||||
json.dumps({"providers": {"openai": {"api_key": "sk-tu-ban-cu"}}}), # casan: allow - du lieu test
|
||||
encoding="utf-8")
|
||||
|
||||
secrets = FakeSecretStore()
|
||||
cfg = _repo(tmp_path, secrets)
|
||||
|
||||
# đọc ra vẫn thấy khoá...
|
||||
assert cfg.provider_conf("openai")["api_key"] == "sk-tu-ban-cu"
|
||||
# ...nhưng trên đĩa thì hết
|
||||
raw = (tmp_path / "config.json").read_text(encoding="utf-8")
|
||||
assert "sk-tu-ban-cu" not in raw
|
||||
assert json.loads(raw)["schema_version"] == CURRENT_VERSION
|
||||
# và có bản sao lưu
|
||||
assert len(list(tmp_path.glob("*.bak"))) == 1
|
||||
|
||||
|
||||
def test_mo_lai_lan_hai_khong_chuyen_lai(tmp_path):
|
||||
(tmp_path / "config.json").write_text(
|
||||
json.dumps({"providers": {"openai": {"api_key": "sk-x"}}}), encoding="utf-8")
|
||||
secrets = FakeSecretStore()
|
||||
_repo(tmp_path, secrets)
|
||||
so_ban_sao = len(list(tmp_path.glob("*.bak")))
|
||||
|
||||
_repo(tmp_path, secrets)
|
||||
assert len(list(tmp_path.glob("*.bak"))) == so_ban_sao, "không nâng cấp lại"
|
||||
|
||||
|
||||
def test_save_luon_ghi_so_phien_ban(tmp_path):
|
||||
cfg = _repo(tmp_path)
|
||||
cfg.save()
|
||||
raw = json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))
|
||||
assert raw["schema_version"] == CURRENT_VERSION
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Typed Settings Facade — R02-T03."""
|
||||
from __future__ import annotations
|
||||
|
||||
from cowork_local.infrastructure.config.settings_facade import (
|
||||
ProviderSettings, RoutingSettings, SecuritySettings, Settings,
|
||||
)
|
||||
from cowork_local.tests.fakes.fake_config import FakeConfigRepository
|
||||
|
||||
|
||||
def test_provider_doc_duoc_ba_truong():
|
||||
p = ProviderSettings({"base_url": "http://x/v1", "model": "llama3",
|
||||
"api_key": "sk-abc"})
|
||||
assert p.base_url == "http://x/v1"
|
||||
assert p.model == "llama3"
|
||||
assert p.api_key == "sk-abc"
|
||||
assert p.configured is True
|
||||
|
||||
|
||||
def test_ollama_khong_can_khoa_van_tinh_la_da_cau_hinh():
|
||||
"""Điều kiện là có base_url và model, không phải có api_key — Ollama chạy
|
||||
cục bộ nên không cần khoá."""
|
||||
p = ProviderSettings({"base_url": "http://localhost:11434/v1", "model": "llama3"})
|
||||
assert p.api_key == ""
|
||||
assert p.configured is True
|
||||
|
||||
|
||||
def test_thieu_model_thi_chua_cau_hinh():
|
||||
assert ProviderSettings({"base_url": "http://x/v1"}).configured is False
|
||||
assert ProviderSettings({}).configured is False
|
||||
|
||||
|
||||
def test_gia_tri_None_tra_ve_mac_dinh_chu_khong_None():
|
||||
"""File cấu hình cũ hay có khoá để null. Đọc ra None rồi đem so sánh số là
|
||||
vỡ — nên khung nhìn phải nuốt luôn trường hợp này."""
|
||||
r = RoutingSettings({"switch_mode": None, "min_score_gain": None,
|
||||
"confirm_timeout_sec": None})
|
||||
assert r.switch_mode == "off"
|
||||
assert r.min_score_gain == 0.05
|
||||
assert r.confirm_timeout_sec == 60
|
||||
|
||||
|
||||
def test_routing_kieu_du_lieu_dung():
|
||||
r = RoutingSettings({"switch_mode": "auto", "min_score_gain": "0.2",
|
||||
"confirm_timeout_sec": "90"})
|
||||
assert r.enabled is True
|
||||
assert isinstance(r.min_score_gain, float) and r.min_score_gain == 0.2
|
||||
assert isinstance(r.confirm_timeout_sec, int) and r.confirm_timeout_sec == 90
|
||||
|
||||
|
||||
def test_tat_dinh_tuyen():
|
||||
assert RoutingSettings({"switch_mode": "off"}).enabled is False
|
||||
assert RoutingSettings({}).enabled is False
|
||||
|
||||
|
||||
def test_sua_qua_khung_nhin_la_sua_vao_dict_that():
|
||||
"""Khung nhìn, không phải bản sao — sửa xong gọi save() là xuống đĩa."""
|
||||
d = {"switch_mode": "off"}
|
||||
RoutingSettings(d).switch_mode = "auto"
|
||||
assert d["switch_mode"] == "auto"
|
||||
|
||||
|
||||
def test_raw_de_khong_ai_bi_ket():
|
||||
d = {"switch_mode": "auto", "khoa_chua_dua_vao_khung_nhin": 1}
|
||||
assert RoutingSettings(d).raw()["khoa_chua_dua_vao_khung_nhin"] == 1
|
||||
|
||||
|
||||
def test_security_mac_dinh_la_bat():
|
||||
"""Mặc định an toàn: thiếu cấu hình thì bật kiểm tra, không phải tắt."""
|
||||
s = SecuritySettings({})
|
||||
assert s.enabled is True
|
||||
assert s.validate_prompt is True
|
||||
assert s.validate_commands is True
|
||||
assert s.cowork_confirm_commands is True
|
||||
assert s.command_ai_check is False # trừ cái này: gọi AI, tốn tiền
|
||||
|
||||
|
||||
def test_settings_noi_vao_repo():
|
||||
repo = FakeConfigRepository(active_provider="openai",
|
||||
routing={"switch_mode": "auto"},
|
||||
agent_security={"cowork_confirm_commands": False})
|
||||
s = Settings(repo)
|
||||
assert s.provider().model == "gpt-4o-mini"
|
||||
assert s.routing.enabled is True
|
||||
assert s.security.cowork_confirm_commands is False
|
||||
|
||||
|
||||
def test_doi_provider_thi_khung_nhin_theo_ngay():
|
||||
repo = FakeConfigRepository(active_provider="ollama")
|
||||
s = Settings(repo)
|
||||
assert s.provider().model == "llama3"
|
||||
repo.set_active_provider("openai")
|
||||
assert s.provider().model == "gpt-4o-mini"
|
||||
Reference in New Issue
Block a user