diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..fba4dfc --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,32 @@ +name: CLI and ModelTrace integration tests +on: [push, pull_request] +permissions: + contents: read +jobs: + test: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + python: ['3.10', '3.12', '3.14'] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v4 + with: + path: switcher + - uses: actions/checkout@v4 + with: + repository: FastR-D/ModelTrace + path: fingerprint-library + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + - run: python -m pip install ./fingerprint-library ./switcher build + - run: python -m unittest discover -s tests -v + working-directory: switcher + env: + PYTHONUTF8: '0' + - run: python -m build + working-directory: switcher + - run: python -m build fingerprint-library + - run: python fingerprint-library/tests/verify_installation.py --wheels fingerprint-library/dist/modeltrace-0.1.0-py3-none-any.whl switcher/dist/cc_switch_simple-0.2.1-py3-none-any.whl diff --git a/.gitignore b/.gitignore index 2febee5..42a7965 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,8 @@ -my_conf.yaml \ No newline at end of file +my_conf.yaml +.venv/ +__pycache__/ +*.egg-info/ +build/ +dist/ +confs/ +output/ diff --git a/README.md b/README.md index fbc6b38..81f18c3 100644 --- a/README.md +++ b/README.md @@ -1,58 +1,47 @@ -### 最简cc-switch -目前cc-switch-cli只能基于tui配置,操作比较麻烦。 -实现了一个简单版,一键导入配置,增加antropic和openai中转。 +# cc-switch-simple + +读取本地 YAML,切换 Codex 或 Claude Code 配置,用 `check` 调用 ModelTrace 检测模型指纹。 ## 安装 -```bash -chmod +x cc-simple.py -# 可选:创建软链接到 PATH -ln -s $(pwd)/cc-simple.py /usr/local/bin/cc-simple +需要 Python 3.10+。在项目目录执行: + +```sh +python3 -m venv .venv +. .venv/bin/activate +python -m pip install . ``` -## 使用 +`check` 还需要安装 ModelTrace,将下面路径替换为其源码目录: -### 1. 添加配置 -```bash -$ python3 cc-simple.py add conf.yaml - -# OpenAI 兼容 API 配置示例 -$ cat conf.yaml -name=xingshu -base_url=https://submodelhub.ilaien.com/v1 -type=openai -api_key=sk-xxxx -model_name=gpt-4 - -# Anthropic API 配置示例 -$ cat anthropic.yaml -name=anthropic-official -base_url=https://api.anthropic.com -type=anthropic -api_key=sk-ant-api03-xxxx -model_name=claude-opus-5-5 +```sh +python -m pip install /path/to/ModelTrace ``` -配置字段说明: -- `name` - 配置名称(必填) -- `base_url` - API 端点 URL(必填) -- `type` - 提供商类型:`openai` 或 `anthropic`(必填) -- `api_key` - API 密钥(必填) -- `model_name` - 模型名称(可选,默认:openai=gpt-4, anthropic=claude-opus-5-5) +## 配置 +保存为 `provider.yaml`: -### 2. 查看所有配置 -```bash -$ python3 cc-simple.py list -1 sbbbbbb https://fast.sbbbbbbbbb.xyz/v1 -2 anthropic-proxy https://api.anthropic.com +```yaml +name: my-provider +base_url: https://api.example.com/v1 +type: openai +api_key: YOUR_API_KEY +model_name: YOUR_MODEL_ID ``` -### 3. 切换配置 -```bash -$ python3 cc-simple.py use 1 -Backed up existing config to /root/.codex/config.toml.bak -Switched to configuration: sbbbbbb (ID: 1) -Provider type: openai -Base URL: https://fast.sbbbbbbbbb.xyz/v1 +`type` 为 `openai`(Codex)或 `anthropic`(Claude Code)。Codex 接口需支持 Responses。两种配置模板在 [conf_tmpl/](conf_tmpl/)。 + +## 使用 + +```sh +cc-simple add provider.yaml +cc-simple list +cc-simple use 1 --dry-run +cc-simple use 1 +cc-simple check 1 --json ``` + +`use` 和 `check` 也可直接传 YAML 路径。切换前自动备份,保留已有无关配置。`check` 不修改客户端配置,会调用模型接口并消耗接口额度。 + +Windows 安装、环境变量和高级选项见 [使用说明](docs/USAGE.md)。 diff --git a/cc-simple.py b/cc-simple.py index 7329682..7840a1b 100755 --- a/cc-simple.py +++ b/cc-simple.py @@ -1,238 +1,6 @@ #!/usr/bin/env python3 -""" -cc-simple.py - Simple Codex configuration switcher -Manages provider configurations for Codex from YAML files -""" +"""Compatibility entry point; install this project for the cc-simple command.""" +from cc_simple import main -import os -import sys -import argparse -from pathlib import Path -import shutil - - -class ConfigManager: - def __init__(self): - self.home = Path.home() - self.codex_config_dir = self.home / ".codex" - self.codex_config_file = self.codex_config_dir / "config.toml" - self.project_dir = Path(__file__).parent - self.confs_dir = self.project_dir / "confs" - - # Ensure directories exist - self.confs_dir.mkdir(exist_ok=True) - - def parse_yaml_config(self, yaml_file): - """Parse simple key=value YAML format""" - config = {} - with open(yaml_file, 'r') as f: - for line in f: - line = line.strip() - if line and '=' in line: - key, value = line.split('=', 1) - config[key.strip()] = value.strip() - return config - - def add_config(self, yaml_file): - """Add a configuration from YAML file""" - yaml_path = Path(yaml_file) - if not yaml_path.exists(): - print(f"Error: {yaml_file} not found") - return False - - config = self.parse_yaml_config(yaml_path) - - # Validate required fields - required = ['name', 'base_url', 'type'] - for field in required: - if field not in config: - print(f"Error: missing required field '{field}'") - return False - - # Copy to confs directory with index - existing_confs = list(self.confs_dir.glob("*.yaml")) - index = len(existing_confs) + 1 - dest_file = self.confs_dir / f"{index}_{config['name']}.yaml" - - shutil.copy(yaml_path, dest_file) - print(f"Added configuration: {config['name']} (ID: {index})") - return True - - def list_configs(self): - """List all stored configurations""" - confs = sorted(self.confs_dir.glob("*.yaml")) - - if not confs: - print("No configurations found") - return - - for conf_file in confs: - # Extract index from filename (format: 1_name.yaml) - parts = conf_file.stem.split('_', 1) - if len(parts) == 2 and parts[0].isdigit(): - index, name = parts - config = self.parse_yaml_config(conf_file) - base_url = config.get('base_url', 'N/A') - print(f"{index} {name} {base_url}") - # Skip files without proper ID_name format - - def generate_toml_content(self, config): - """Generate TOML content for the provider""" - provider_type = config.get('type', 'openai') - - if provider_type == 'openai': - return self._generate_openai_toml(config) - elif provider_type == 'anthropic': - return self._generate_anthropic_toml(config) - else: - print(f"Warning: unknown type '{provider_type}', using OpenAI format") - return self._generate_openai_toml(config) - - def _generate_openai_toml(self, config): - """Generate OpenAI provider TOML""" - provider_name = config.get('name', 'unnamed') - api_key = config.get('api_key', config.get('model', 'sk-xxx')) - base_url = config.get('base_url', '') - model = config.get('model_name', 'gpt-4') - - toml = f"""# Generated by cc-simple.py -# Provider: {provider_name} - -model_provider = "{provider_name}" -model = "{model}" - -[model_providers.{provider_name}] -name = "{provider_name}" -base_url = "{base_url}" -wire_api = "chat" - -[model_providers.{provider_name}.auth] -header = "Authorization" -value = "Bearer {api_key}" - -[tui] -screen_reader_detection_done = true -""" - return toml - - def _generate_anthropic_toml(self, config): - """Generate Anthropic provider TOML""" - provider_name = config.get('name', 'unnamed') - api_key = config.get('api_key', config.get('model', 'sk-ant-xxx')) - base_url = config.get('base_url', 'https://api.anthropic.com') - model = config.get('model_name', 'claude-opus-5-5') - - toml = f"""# Generated by cc-simple.py -# Provider: {provider_name} - -model_provider = "{provider_name}" -model = "{model}" - -[model_providers.{provider_name}] -name = "{provider_name}" -base_url = "{base_url}" -wire_api = "responses" - -[model_providers.{provider_name}.auth] -header = "x-api-key" -value = "{api_key}" - -[tui] -screen_reader_detection_done = true -""" - return toml - - def use_config(self, config_id): - """Switch to a specific configuration""" - # Find config file by ID - target_file = None - for conf_file in self.confs_dir.glob("*.yaml"): - parts = conf_file.stem.split('_', 1) - if len(parts) == 2 and parts[0] == str(config_id): - target_file = conf_file - break - - if not target_file: - print(f"Error: configuration ID {config_id} not found") - return False - - # Parse config - config = self.parse_yaml_config(target_file) - - # Generate TOML content - toml_content = self.generate_toml_content(config) - - # Backup existing config if it exists - if self.codex_config_file.exists(): - backup_file = self.codex_config_file.with_suffix('.toml.bak') - shutil.copy(self.codex_config_file, backup_file) - print(f"Backed up existing config to {backup_file}") - - # Ensure codex config directory exists - self.codex_config_dir.mkdir(parents=True, exist_ok=True) - - # Write new config - with open(self.codex_config_file, 'w') as f: - f.write(toml_content) - - print(f"Switched to configuration: {config.get('name', 'unnamed')} (ID: {config_id})") - print(f"Provider type: {config.get('type', 'openai')}") - print(f"Base URL: {config.get('base_url', 'N/A')}") - return True - - -def main(): - parser = argparse.ArgumentParser( - description='Simple Codex configuration switcher', - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Examples: - # Add a configuration - cc-simple.py add conf.yaml - - # List all configurations - cc-simple.py list - - # Switch to configuration ID 1 - cc-simple.py use 1 - """ - ) - - subparsers = parser.add_subparsers(dest='command', help='Command to execute') - - # Add command - add_parser = subparsers.add_parser('add', help='Add a configuration from YAML file') - add_parser.add_argument('yaml_file', help='Path to YAML configuration file') - - # List command - subparsers.add_parser('list', help='List all configurations') - - # Use command - use_parser = subparsers.add_parser('use', help='Switch to a configuration') - use_parser.add_argument('config_id', type=int, help='Configuration ID to use') - - args = parser.parse_args() - - if not args.command: - parser.print_help() - return 1 - - manager = ConfigManager() - - if args.command == 'add': - success = manager.add_config(args.yaml_file) - return 0 if success else 1 - - elif args.command == 'list': - manager.list_configs() - return 0 - - elif args.command == 'use': - success = manager.use_config(args.config_id) - return 0 if success else 1 - - return 0 - - -if __name__ == '__main__': - sys.exit(main()) +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cc_simple.py b/cc_simple.py new file mode 100644 index 0000000..febfe3c --- /dev/null +++ b/cc_simple.py @@ -0,0 +1,356 @@ +"""A small YAML configuration switcher for Codex and Claude Code.""" +import argparse +import json +import math +import os +import re +import sys +import tempfile +import uuid +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import urlsplit, urlunsplit + +import tomlkit +import yaml + + +class ConfigError(ValueError): + pass + + +class UniqueLoader(yaml.SafeLoader): + """Allow merge defaults while rejecting duplicate explicit keys.""" + def flatten_mapping(self, node): + if not getattr(node, "cc_checked", False): + keys = set() + for key_node, _ in node.value: + key = "<<" if key_node.tag == "tag:yaml.org,2002:merge" else self.construct_object(key_node) + if not isinstance(key, str) or key in keys: + raise ConfigError("YAML keys must be unique strings") + keys.add(key) + node.cc_checked = True + super().flatten_mapping(node) + + +def read_config(path): + text = Path(path).read_text(encoding="utf-8-sig") + lines = [line.strip() for line in text.splitlines() + if line.strip() and not line.lstrip().startswith("#")] + if lines and all(re.match(r"^[A-Za-z_]+\s*=", line) for line in lines): + data = {} + for line in lines: + key, value = line.split("=", 1) + if key.strip() in data: + raise ConfigError("duplicate configuration field") + data[key.strip()] = value.strip() + else: + data = yaml.load(text, Loader=UniqueLoader) + if not isinstance(data, dict): + raise ConfigError("configuration must be a YAML mapping or legacy key=value file") + allowed = {"name", "base_url", "type", "api_key", "env_key", "model_name", + "model", "target", "api_format", "temperature"} + if set(data) - allowed: + raise ConfigError("unknown configuration fields: " + ", ".join(sorted(set(data) - allowed))) + if "model" in data: + if "api_key" in data or "env_key" in data: + raise ConfigError("legacy model is a credential alias; use model_name for the API model") + data["api_key"] = data.pop("model") + for key in ("name", "base_url", "type"): + if not isinstance(data.get(key), str) or not data[key].strip(): + raise ConfigError(f"missing or invalid {key}") + data[key] = data[key].strip() + if data["type"] not in ("openai", "anthropic"): + raise ConfigError("type must be openai or anthropic") + try: + url = urlsplit(data["base_url"]) + url.port + except ValueError: + raise ConfigError("invalid base_url") from None + if (any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in data["base_url"]) or + url.scheme not in ("http", "https") or not url.hostname or + url.username or url.password or url.query or url.fragment): + raise ConfigError("base_url must be an HTTP(S) URL without credentials, query or fragment") + data["base_url"] = data["base_url"].rstrip("/") + data.setdefault("model_name", "gpt-4" if data["type"] == "openai" else "claude-opus-5-5") + if not isinstance(data["model_name"], str) or not data["model_name"].strip(): + raise ConfigError("model_name must be a non-empty string") + data["model_name"] = data["model_name"].strip() + if bool(data.get("api_key")) == bool(data.get("env_key")): + raise ConfigError("provide exactly one of api_key or env_key") + for key in ("api_key", "env_key"): + if key in data and (not isinstance(data[key], str) or not data[key].strip()): + raise ConfigError(f"{key} must be a non-empty string") + if "env_key" in data and not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", data["env_key"]): + raise ConfigError("env_key must be an environment variable name") + if "api_key" in data and any(ord(char) < 32 or ord(char) == 127 for char in data["api_key"]): + raise ConfigError("api_key contains a forbidden control character") + if data.get("target") not in (None, "codex", "claude"): + raise ConfigError("target must be codex or claude") + if data.get("api_format") not in (None, "auto", "openai", "anthropic", "responses"): + raise ConfigError("invalid api_format") + if "temperature" in data: + value = data["temperature"] + if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 <= value <= 2 or not math.isfinite(value): + raise ConfigError("temperature must be a finite number between 0 and 2") + return data + + +def resolve_target(config, target=None): + target = target or config.get("target") or ("claude" if config["type"] == "anthropic" else "codex") + if target == "claude" and config["type"] != "anthropic": + raise ConfigError("Claude Code requires an Anthropic Messages endpoint") + if target == "codex" and config["type"] == "anthropic" and config.get("api_format") != "responses": + raise ConfigError("Anthropic-to-Codex requires an actual Responses proxy and api_format: responses") + return target + + +def client_base_url(base_url, target): + """Normalize API inputs for the endpoint suffix added by each real client.""" + url = urlsplit(base_url) + path = url.path.rstrip("/") + if target == "codex": + for suffix in ("/responses", "/chat/completions"): + if path.endswith(suffix): + path = path[:-len(suffix)] + if not path: + raise ConfigError("Codex endpoints need an API prefix such as /v1/responses") + break + if path.endswith("/messages"): + raise ConfigError("Codex requires a Responses API base or endpoint") + path = path or "/v1" + else: + if path.endswith("/messages") and not path.endswith("/v1/messages"): + raise ConfigError("Claude Messages endpoints must end with /v1/messages") + for suffix in ("/v1/messages", "/v1"): + if path.endswith(suffix): + path = path[:-len(suffix)] + break + if path.endswith(("/responses", "/chat/completions")): + raise ConfigError("Claude requires a Messages API base or endpoint") + return urlunsplit((url.scheme, url.netloc, path, "", "")) + + +def credential(config): + value = config.get("api_key") or os.environ.get(config.get("env_key", "")) + if not value: + raise ConfigError("configured credential environment variable is not set") + if any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ConfigError("credential contains a forbidden control character") + return value + + +def atomic_write(path, content, exclusive=False): + path = Path(path) + if path.is_symlink(): + raise ConfigError("refusing to replace a symbolic-link configuration file") + if not path.parent.exists(): + path.parent.mkdir(parents=True, mode=0o700, exist_ok=True) + fd, temporary = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent) + try: + with os.fdopen(fd, "wb") as output: + output.write(content.encode("utf-8") if isinstance(content, str) else content) + output.flush() + os.fsync(output.fileno()) + os.chmod(temporary, 0o600) + if exclusive: + os.link(temporary, path) + else: + os.replace(temporary, path) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +def configurations(directory): + entries = {} + for path in directory.glob("*.y*ml"): + match = re.fullmatch(r"([1-9][0-9]*)_.+\.ya?ml", path.name) + if match: + identity = int(match[1]) + if identity in entries: + raise ConfigError(f"duplicate stored configuration ID {identity}") + entries[identity] = path + return entries + + +def source_config(source, directory): + if re.fullmatch(r"[1-9][0-9]*", source): + path = configurations(directory).get(int(source)) + if path is None: + raise ConfigError("configuration ID not found") + else: + path = Path(source).expanduser() + return read_config(path) + + +def summary(config): + return {key: config[key] for key in ("name", "base_url", "type", "model_name")} + + +def switch(config, target_dir=None, target=None, dry_run=False): + target = resolve_target(config, target) + config = dict(config, base_url=client_base_url(config["base_url"], target)) + variable, default = ("CODEX_HOME", ".codex") if target == "codex" else ("CLAUDE_CONFIG_DIR", ".claude") + location = target_dir or os.environ.get(variable) + directory = Path(location).expanduser() if location else Path.home() / default + path = directory / ("config.toml" if target == "codex" else "settings.json") + if path.is_symlink(): + raise ConfigError("refusing to replace a symbolic-link configuration file") + original = path.read_bytes() if path.exists() else None + if target == "codex": + document = tomlkit.parse(original.decode("utf-8-sig")) if original else tomlkit.document() + provider_id = "cc_simple" + providers = document.setdefault("model_providers", tomlkit.table()) + if not isinstance(providers, dict): + raise ConfigError("model_providers must be a TOML table") + provider = providers.get(provider_id, tomlkit.table()) + if not isinstance(provider, dict): + raise ConfigError("cc_simple provider must be a TOML table") + for key in ("env_key", "experimental_bearer_token", "auth", "http_headers", "env_http_headers", "query_params", + "gateway_oauth", "aws", "model_catalog_url", "env_key_instructions"): + provider.pop(key, None) + provider.update({"name": config["name"], "base_url": config["base_url"], "wire_api": "responses"}) + provider["requires_openai_auth"] = False + if "env_key" in config: + provider["env_key"] = config["env_key"] + else: + provider["experimental_bearer_token"] = config["api_key"] + providers[provider_id] = provider + document["model_provider"] = provider_id + document["model"] = config["model_name"] + content = tomlkit.dumps(document) + else: + document = json.loads(original.decode("utf-8-sig")) if original else {} + if not isinstance(document, dict) or not isinstance(document.get("env", {}), dict): + raise ConfigError("Claude settings and env must be JSON objects") + environment = document.setdefault("env", {}) + for key in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_MODEL"): + environment.pop(key, None) + environment["ANTHROPIC_BASE_URL"] = config["base_url"] + environment["ANTHROPIC_API_KEY"] = "" if dry_run else credential(config) + document["model"] = config["model_name"] + content = json.dumps(document, ensure_ascii=False, indent=2) + "\n" + result = dict(summary(config), target=target, path=str(path), dry_run=dry_run) + if not dry_run: + if original is not None: + suffix = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S") + "-" + uuid.uuid4().hex[:8] + backup = path.with_name(path.name + ".bak." + suffix) + atomic_write(backup, original) + result["backup"] = str(backup) + atomic_write(path, content) + return result + + +def check(config, args): + target = resolve_target(config, args.target) + config = dict(config, base_url=client_base_url(config["base_url"], target)) + api_base = config["base_url"] + ("/v1" if target == "claude" else "") + api_format = args.api_format or config.get("api_format") or ("responses" if target == "codex" else "anthropic") + protocol = ("responses" if target == "codex" else "anthropic") if api_format == "auto" else api_format + endpoint = api_base + {"responses": "/responses", "openai": "/chat/completions", "anthropic": "/messages"}[protocol] + try: + from modeltrace import check_api + except ImportError: + raise ConfigError("check requires the modeltrace library; see README installation") from None + result = check_api(base_url=endpoint, api_key=credential(config), + api_model=config["model_name"], api_format=api_format, + temperature=args.temperature if args.temperature is not None else config.get("temperature"), + timeout=args.timeout, attempts=args.attempts) + return dict(summary(config), target=target, api_base_url=api_base, api_endpoint=endpoint, requested_format=api_format, analysis=result) + + +def positive_number(value): + number = float(value) + if not math.isfinite(number) or not 0 < number <= 240: + raise argparse.ArgumentTypeError("must be a finite number between 0 and 240") + return number + + +def attempt_count(value): + number = int(value) + if not 3 <= number <= 20: + raise argparse.ArgumentTypeError("must be between 3 and 20") + return number + + +def make_parser(): + parser = argparse.ArgumentParser(description=__doc__) + common = argparse.ArgumentParser(add_help=False) + common.add_argument("--config-dir", default=argparse.SUPPRESS) + common.add_argument("--json", action="store_true", default=argparse.SUPPRESS) + parser.add_argument("--config-dir", default="~/.cc-simple/confs") + parser.add_argument("--json", action="store_true") + commands = parser.add_subparsers(dest="command", required=True) + add = commands.add_parser("add", parents=[common], help="import a YAML configuration") + add.add_argument("source") + commands.add_parser("list", parents=[common], help="list imported configurations") + use = commands.add_parser("use", parents=[common], help="switch Codex or Claude Code configuration") + use.add_argument("source", help="ID or YAML path") + use.add_argument("--target", choices=("codex", "claude")) + use.add_argument("--target-dir") + use.add_argument("--dry-run", action="store_true") + test = commands.add_parser("check", parents=[common], help="run ModelTrace fingerprint analysis") + test.add_argument("source", help="ID or YAML path") + test.add_argument("--target", choices=("codex", "claude")) + test.add_argument("--format", dest="api_format", choices=("auto", "openai", "anthropic", "responses")) + test.add_argument("--timeout", type=positive_number, default=60) + test.add_argument("--attempts", type=attempt_count, default=6) + test.add_argument("--temperature", type=float) + return parser + + +def main(argv=None): + args = make_parser().parse_args(argv) + config = None + try: + directory = Path(args.config_dir).expanduser() + if args.command == "list": + result = [dict(summary(read_config(path)), id=identity) + for identity, path in sorted(configurations(directory).items())] + else: + config = read_config(args.source) if args.command == "add" else source_config(args.source, directory) + if args.command == "add": + resolve_target(config) + identity = max(configurations(directory), default=0) + 1 + while True: + destination = directory / f"{identity}_config.yaml" + try: + atomic_write(destination, yaml.safe_dump(config, allow_unicode=True, sort_keys=False), exclusive=True) + break + except FileExistsError: + identity += 1 + result = dict(summary(config), id=identity, path=str(destination)) + elif args.command == "use": + result = switch(config, args.target_dir, args.target, args.dry_run) + else: + if args.temperature is not None and (not math.isfinite(args.temperature) or not 0 <= args.temperature <= 2): + raise ConfigError("temperature must be a finite number between 0 and 2") + result = check(config, args) + if args.json: + print(json.dumps(result, ensure_ascii=False)) + elif isinstance(result, list): + for item in result: + print(f"{item['id']} {item['name']} {item['type']} {item['base_url']} {item['model_name']}") + if not result: + print("No configurations found") + else: + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + except ConfigError as error: + code, message = 2, str(error) + except (yaml.YAMLError, tomlkit.exceptions.ParseError, UnicodeError, json.JSONDecodeError): + code, message = 2, "invalid YAML, TOML, JSON or text encoding; existing files were preserved" + except ValueError: + code, message = 2, "invalid input value; existing files were preserved" + except (OSError, RuntimeError) as error: + code, message = 1, str(error) + if config: + secret = config.get("api_key") or os.environ.get(config.get("env_key", "")) + if secret: + message = message.replace(secret, "") + if args.json: + print(json.dumps({"error": message, "exit_code": code}, ensure_ascii=False)) + else: + print("Error: " + message, file=sys.stderr) + return code diff --git a/conf.yaml b/conf.yaml index 2827fa2..7703047 100644 --- a/conf.yaml +++ b/conf.yaml @@ -1,4 +1,5 @@ -name=sbbbbbb -base_url=https://fast.sbbbbbbbbb.xyz/v1 -type=openai -model=sk-xxxx \ No newline at end of file +name: responses-example +base_url: https://your-responses-proxy.example/v1 +type: openai +env_key: MODEL_API_TOKEN +model_name: your-real-model-id diff --git a/conf_tmpl/anthropic_tmpl.yaml b/conf_tmpl/anthropic_tmpl.yaml index edad53b..54b9242 100644 --- a/conf_tmpl/anthropic_tmpl.yaml +++ b/conf_tmpl/anthropic_tmpl.yaml @@ -1,5 +1,5 @@ -name=anthropic-official -base_url=https://api.anthropic.com -type=anthropic -api_key=sk-ant-api03-your-key-here -model_name=claude-opus-5-5 +name: anthropic-proxy +base_url: https://your-anthropic-proxy.example +type: anthropic +env_key: MODEL_API_TOKEN +model_name: your-real-claude-model-id diff --git a/conf_tmpl/codex_tmpl.yaml b/conf_tmpl/codex_tmpl.yaml index 6ac2e5b..6207d34 100644 --- a/conf_tmpl/codex_tmpl.yaml +++ b/conf_tmpl/codex_tmpl.yaml @@ -1,5 +1,5 @@ -name=xingshu -base_url=https://submodelhub.ilaien.com/v1 -type=openai -api_key=sk-your-api-key-here -model_name=gpt-4 +name: responses-proxy +base_url: https://your-responses-proxy.example/v1 +type: openai +env_key: MODEL_API_TOKEN +model_name: your-real-model-id diff --git a/docs/USAGE.md b/docs/USAGE.md new file mode 100644 index 0000000..3ea3457 --- /dev/null +++ b/docs/USAGE.md @@ -0,0 +1,164 @@ +# 使用说明 + +## 安装 + +Python 3.10+。Linux 安装命令见 [README](../README.md);已有虚拟环境时可直接使用。`add`、`list`、`use` 只依赖 PyYAML 和 tomlkit,`check` 另需 ModelTrace。ModelTrace 安装在同一环境后,无需启动其网页服务。 + +Windows 在项目目录执行,将 `C:\path\to\ModelTrace` 替换为 ModelTrace 源码目录: + +```powershell +py -3 -m venv .venv +.\.venv\Scripts\python.exe -m pip install . +.\.venv\Scripts\python.exe -m pip install C:\path\to\ModelTrace +.\.venv\Scripts\cc-simple.exe --help +``` + +如果已拿到两个 wheel,也可在同一虚拟环境安装: + +```sh +python -m pip install modeltrace-0.1.0-py3-none-any.whl cc_switch_simple-0.2.1-py3-none-any.whl +``` + +安装后使用该环境的 `cc-simple` 命令。原入口 `python cc-simple.py` 仍可用。 + +## 配置字段 + +| 字段 | 含义 | +| --- | --- | +| `name` | 配置名称,必填 | +| `base_url` | HTTP(S) 接口地址,必填;不能包含凭据、查询参数或片段 | +| `type` | `openai` 或 `anthropic`,必填 | +| `api_key` | 接口密钥,与 `env_key` 二选一 | +| `env_key` | 保存密钥的环境变量名称,与 `api_key` 二选一 | +| `model_name` | 向接口请求的模型 ID,建议明确填写 | +| `target` | 可选,`codex` 或 `claude`,默认按 `type` 选择 | +| `api_format` | 可选,`auto`、`openai`、`anthropic` 或 `responses` | +| `temperature` | 可选,0 至 2;省略时不发送温度参数 | + +`model_name` 省略时沿用旧版默认值:`openai` 为 `gpt-4`,`anthropic` 为 `claude-opus-5-5`。接口未必提供这些 ID,应按服务商实际模型填写。 + +使用环境变量时,将 YAML 的 `api_key` 换为 `env_key`: + +```yaml +name: my-provider +base_url: https://api.example.com/v1 +type: openai +env_key: MODEL_API_TOKEN +model_name: YOUR_MODEL_ID +``` + +在当前终端设置密钥: + +```sh +export MODEL_API_TOKEN='YOUR_API_KEY' +``` + +```powershell +$env:MODEL_API_TOKEN = 'YOUR_API_KEY' +``` + +文件使用 UTF-8,支持 Windows UTF-8 BOM、YAML 锚点和 `<<` 合并。显式字段可覆盖合并默认值;重复显式字段、重复合并键、未知字段和无效配置会报错。 + +旧版逐行 `key=value` 文件仍可读取。旧字段 `model=密钥` 是 `api_key` 的别名,模型名称必须用 `model_name`。 + +## 导入与切换 + +`add` 将配置保存到 `~/.cc-simple/confs` 并分配数字 ID,`list` 按 ID 排序。配置名称不会用作文件路径。用 `--config-dir` 指定其他目录;旧版 `confs/` 可通过 `--config-dir ./confs` 使用。重复 ID 会报错。 + +`use` 和 `check` 接受数字 ID 或 YAML 路径: + +```sh +cc-simple use provider.yaml --dry-run +cc-simple use provider.yaml +cc-simple check provider.yaml --json +``` + +`--dry-run` 校验输入与已有配置,返回目标摘要,不写文件。默认切换位置为: + +| 类型 | 配置文件 | +| --- | --- | +| `openai` | `~/.codex/config.toml` | +| `anthropic` | `~/.claude/settings.json` | + +若设置 `CODEX_HOME` 或 `CLAUDE_CONFIG_DIR`,则分别使用该目录中的 `config.toml` 或 `settings.json`;空环境变量使用上述默认目录。显式 `--target-dir` 优先于环境变量。`--dry-run` 使用相同目录解析规则。 + +切换保留 TOML 注释、其他 provider、项目配置、Claude permissions 等无关设置。Codex 的 `cc_simple` provider 会清除旧认证、headers 和 query 参数,保留 timeout、retry 等无关字段。Claude 会清除旧 `ANTHROPIC_AUTH_TOKEN` 和 `ANTHROPIC_MODEL`,写入新接口、密钥及模型。 + +已有文件在切换前保存为唯一备份,例如 `config.toml.bak.20261009T120000-a1b2c3d4`;新文件写完并刷新后原子替换。现有 TOML/JSON 无效时停止写入。恢复时先退出客户端,再用所需备份替换目标文件。 + +`api_key` 会保存在配置与备份中。Codex 使用 `env_key` 时保留变量名;Claude 使用 `env_key` 时解析当前变量值并写入 settings。Linux 新配置目录权限为 700,配置和备份为 600;Windows 沿用父目录 ACL。列表、普通结果及 dry-run 不展示密钥。 + +测试时可隔离配置目录和目标目录: + +```sh +cc-simple --config-dir ./demo-store add provider.yaml +cc-simple --config-dir ./demo-store use 1 --target-dir ./demo-codex --dry-run +cc-simple --config-dir ./demo-store use 1 --target-dir ./demo-codex +cc-simple --config-dir ./demo-store list --json +``` + +## 协议与地址 + +Codex 使用 Responses,认证字段为 `env_key` 或 `experimental_bearer_token`。只有 Chat Completions 的接口可用 `check --format openai` 检测,但不能直接供 Codex 运行。 + +Claude 使用 Anthropic Messages 的标准 `x-api-key` 认证,检测与切换使用相同认证方式。只接受 Bearer token 的中转站需要提供标准 Messages 兼容认证。 + +若 Anthropic 模型通过 Responses 中转,可设置 `api_format: responses`,配合 `target: codex` 或 `--target codex`。服务商必须实际支持 Responses。 + +`use` 与 `check` 共用地址规范化,保留自定义前缀: + +| 目标 | YAML 地址路径 | 写入客户端的路径 | 默认检测路径 | +| --- | --- | --- | --- | +| Codex | 裸域名、`/v1`、`/v1/responses` | `/v1` | `/v1/responses` | +| Codex | `/gateway/v1`、`/gateway/v1/responses` | `/gateway/v1` | `/gateway/v1/responses` | +| Codex | `/gateway`、`/gateway/responses` | `/gateway` | `/gateway/responses` | +| Claude | 裸域名、`/v1`、`/v1/messages` | 裸域名 | `/v1/messages` | +| Claude | `/gateway`、`/gateway/v1`、`/gateway/v1/messages` | `/gateway` | `/gateway/v1/messages` | + +Codex 裸域名按 `/v1` 处理,无前缀的完整 `/responses`、`/chat/completions` 地址会被拒绝。Claude 完整端点需以 `/v1/messages` 结尾,只有 `/messages` 的地址会被拒绝。末尾斜线会去除。结果的 `base_url` 为客户端地址,`check` 的 `api_endpoint` 为实际请求端点。 + +## 指纹检测 + +`check` 调用 `modeltrace.check_api`,不修改客户端配置。默认最多尝试 6 个数字挑战,得到 3 份有效回答后分析;网络失败、鉴权失败或有效回答不足会报错。请求会消耗接口额度。 + +默认协议随目标选择:Codex 为 `responses`,Claude 为 `anthropic`;`--format auto` 也遵循该规则。可通过 `--format` 覆盖协议,其中 `openai` 指 Chat Completions,`responses` 指 Responses。覆盖协议后检测成功,只能证明该协议可用。 + +```sh +cc-simple check 1 --format openai --timeout 60 --attempts 6 --json +cc-simple check provider.yaml --format anthropic --temperature 0 +``` + +`--timeout` 为每个挑战的请求时间预算,大于 0 且不超过 240 秒。`--attempts` 为 3 至 20。命令行的 `--temperature` 覆盖 YAML 设置。 + +JSON 结果字段: + +| 字段 | 含义 | +| --- | --- | +| `analysis.results` | 候选模型排序 | +| `analysis.families` | 模型家族排序 | +| `analysis.used_outputs` | 有效回答数量 | +| `analysis.api_test` | 实际协议、尝试次数等检测信息 | + +指纹结果基于候选样本的统计相似性,可辅助核验模型;排名与分数不能保证真实身份。 + +## Python 调用与退出码 + +配置读取和切换可调用 `cc_simple.read_config(path)`、`cc_simple.switch(config, target_dir=...)`。`switch` 返回配置摘要与备份路径,配置错误抛出 `cc_simple.ConfigError`。 + +指纹检测调用 `modeltrace.check_api`,成功返回分析字典;接口失败或有效回答不足抛出 `modeltrace.APIError`,输入错误抛出 `ValueError`。 + +| 退出码 | 含义 | +| --- | --- | +| `0` | 成功 | +| `2` | 输入、配置或格式错误 | +| `1` | API、网络或文件操作失败 | + +`--json` 可放在子命令前或后;业务错误返回 JSON 的 `error`、`exit_code`。命令行参数语法错误由 argparse 在 stderr 提示。程序不回显请求密钥。 + +## 测试 + +在项目目录执行,测试使用临时目录和模拟接口: + +```sh +python -m unittest discover -s tests -v +``` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c0be881 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,17 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "cc-switch-simple" +version = "0.2.1" +description = "Minimal YAML configuration switcher with ModelTrace fingerprint checks" +readme = "README.md" +requires-python = ">=3.10" +dependencies = ["PyYAML>=6,<7", "tomlkit>=0.13,<1"] + +[project.scripts] +cc-simple = "cc_simple:main" + +[tool.setuptools] +py-modules = ["cc_simple"] diff --git a/tests/demo-provider.yaml b/tests/demo-provider.yaml new file mode 100644 index 0000000..0c657fc --- /dev/null +++ b/tests/demo-provider.yaml @@ -0,0 +1,7 @@ +# Only works while the local acceptance demo API is running. +name: local-acceptance-demo +base_url: http://127.0.0.1:17861/v1 +type: openai +api_key: local-test-only +model_name: local-mock +api_format: openai diff --git a/tests/test_cc_simple.py b/tests/test_cc_simple.py new file mode 100644 index 0000000..a4b640c --- /dev/null +++ b/tests/test_cc_simple.py @@ -0,0 +1,275 @@ +import contextlib +import io +import json +import os +import subprocess +import sys +import tempfile +import types +import unittest +from pathlib import Path +from unittest.mock import patch + +import tomlkit +import yaml + +import cc_simple as cli + + +class SwitcherTests(unittest.TestCase): + def test_cli_respects_client_configuration_directories(self): + script = Path(cli.__file__).with_name("cc-simple.py") + for target, variable, filename in (("codex", "CODEX_HOME", "config.toml"), + ("claude", "CLAUDE_CONFIG_DIR", "settings.json")): + configured = self.root / (target + "-configured") + explicit = self.root / (target + "-explicit") + source = self.write_source(dict(self.config, type="openai" if target == "codex" else "anthropic")) + environment = {**os.environ, "PYTHONUTF8": "1", "PYTHONIOENCODING": "utf-8", variable: str(configured)} + command = [sys.executable, str(script), "--json", "use", str(source), "--target", target] + with self.subTest(target=target): + preview = subprocess.run([*command, "--dry-run"], env=environment, capture_output=True, encoding="utf-8", check=True) + self.assertEqual(Path(json.loads(preview.stdout)["path"]), configured / filename) + self.assertFalse(configured.exists()) + result = subprocess.run(command, env=environment, capture_output=True, encoding="utf-8", check=True) + self.assertEqual(Path(json.loads(result.stdout)["path"]), configured / filename) + self.assertTrue((configured / filename).is_file()) + override = subprocess.run([*command, "--target-dir", str(explicit)], env=environment, capture_output=True, encoding="utf-8", check=True) + self.assertEqual(Path(json.loads(override.stdout)["path"]), explicit / filename) + environment[variable] = "" + fallback = subprocess.run([*command, "--dry-run"], env=environment, capture_output=True, encoding="utf-8", check=True) + self.assertEqual(Path(json.loads(fallback.stdout)["path"]), Path.home() / (".codex" if target == "codex" else ".claude") / filename) + + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.store = self.root / "store" + self.target = self.root / "target" + self.config = {"name": "../中文 配置", "base_url": "https://example.test/v1", + "type": "openai", "api_key": "test-only-SECRET", "model_name": "test-model"} + self.source = self.write_source(self.config) + + def write_source(self, config, name="provider.yaml"): + path = self.root / name + path.write_text(yaml.safe_dump(config, allow_unicode=True), encoding="utf-8") + return path + + def run_cli(self, *arguments): + output = io.StringIO() + errors = io.StringIO() + with contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): + code = cli.main(["--config-dir", str(self.store), "--json", *map(str, arguments)]) + return code, json.loads(output.getvalue()), errors.getvalue() + + def test_yaml_roundtrip_safe_ids_and_gaps(self): + code, added, _ = self.run_cli("add", self.source) + self.assertEqual(code, 0) + self.assertEqual(Path(added["path"]).parent, self.store) + self.assertEqual(Path(added["path"]).name, "1_config.yaml") + (self.store / "8_legacy.yaml").write_text(self.source.read_text(encoding="utf-8"), encoding="utf-8") + code, added, _ = self.run_cli("add", self.source) + self.assertEqual(added["id"], 9) + code, items, _ = self.run_cli("list") + self.assertEqual([item["id"] for item in items], [1, 8, 9]) + self.assertNotIn(self.config["api_key"], json.dumps(items)) + code, used, _ = self.run_cli("use", "9", "--target-dir", self.target) + self.assertEqual(code, 0) + self.assertEqual(used["target"], "codex") + + def test_legacy_key_equals_and_model_credential(self): + self.source.write_text("name=legacy\nbase_url=https://example.test/v1\ntype=openai\nmodel=old-secret\n", encoding="utf-8-sig") + loaded = cli.read_config(self.source) + self.assertEqual(loaded["api_key"], "old-secret") + self.assertEqual(loaded["model_name"], "gpt-4") + + def test_duplicate_keys_unknown_fields_and_missing_token_fail_without_write(self): + invalid = [dict(self.config, tyep="openai"), {k: v for k, v in self.config.items() if k != "api_key"}, + dict(self.config, env_key="MY_TOKEN"), dict(self.config, base_url="file:///tmp/test"), + dict(self.config, base_url="https://api.exam ple.test/v1")] + for config in invalid: + with self.subTest(config=config): + source = self.write_source(config) + code, error, _ = self.run_cli("add", source) + self.assertEqual(code, 2) + self.assertIn("error", error) + self.assertFalse(self.store.exists()) + self.source.write_text("name: first\nname: second\n", encoding="utf-8") + self.assertEqual(self.run_cli("add", self.source)[0], 2) + + def test_numeric_list_order_and_duplicate_id_fail(self): + self.store.mkdir() + for identity in (2, 10, 1): + (self.store / f"{identity}_config.yaml").write_bytes(self.source.read_bytes()) + self.assertEqual([item["id"] for item in self.run_cli("list")[1]], [1, 2, 10]) + (self.store / "2_other.yaml").write_bytes(self.source.read_bytes()) + self.assertEqual(self.run_cli("list")[0], 2) + + def test_codex_preserves_settings_comments_and_unique_backups(self): + self.target.mkdir() + existing = '# retained comment\nmodel="old"\napproval_policy="never"\n[projects."/work"]\ntrust_level="trusted"\n[model_providers.other]\nname="other"\n[model_providers.cc_simple]\nrequest_max_retries=7\n' + path = self.target / "config.toml" + path.write_text(existing, encoding="utf-8") + code, first, _ = self.run_cli("use", self.source, "--target-dir", self.target) + self.assertEqual(code, 0) + content = path.read_text(encoding="utf-8") + loaded = tomlkit.parse(content) + self.assertIn("# retained comment", content) + self.assertEqual(loaded["approval_policy"], "never") + self.assertEqual(loaded["projects"]["/work"]["trust_level"], "trusted") + self.assertEqual(loaded["model_providers"]["other"]["name"], "other") + provider = loaded["model_providers"]["cc_simple"] + self.assertEqual(provider["request_max_retries"], 7) + self.assertEqual(provider["wire_api"], "responses") + self.assertEqual(provider["experimental_bearer_token"], self.config["api_key"]) + self.assertEqual(Path(first["backup"]).read_text(encoding="utf-8"), existing) + code, second, _ = self.run_cli("use", self.source, "--target-dir", self.target) + self.assertNotEqual(first["backup"], second["backup"]) + + def test_environment_auth_is_reference_for_codex(self): + config = dict(self.config, env_key="TEST_MODEL_TOKEN") + del config["api_key"] + source = self.write_source(config) + with patch.dict(os.environ, {}, clear=True): + code, _, _ = self.run_cli("use", source, "--target-dir", self.target) + self.assertEqual(code, 0) + provider = tomlkit.parse((self.target / "config.toml").read_text(encoding="utf-8"))["model_providers"]["cc_simple"] + self.assertEqual(provider["env_key"], "TEST_MODEL_TOKEN") + self.assertNotIn("experimental_bearer_token", provider) + + def test_new_host_does_not_inherit_old_endpoint_credentials(self): + self.target.mkdir() + path = self.target / "config.toml" + path.write_text('[model_providers.cc_simple]\nbase_url="https://old.test/v1"\nrequest_max_retries=7\nmodel_catalog_url="https://old.test/models"\nenv_key_instructions="old auth"\n[model_providers.cc_simple.aws]\nregion="us-east-1"\n[model_providers.cc_simple.gateway_oauth]\naudience="old"\n[model_providers.cc_simple.http_headers]\nx-api-key="OLD_SECRET"\n[model_providers.cc_simple.env_http_headers]\nx-key="OLD_SECRET_ENV"\n[model_providers.cc_simple.query_params]\napi-key="OLD_SECRET"\n', encoding="utf-8") + self.assertEqual(self.run_cli("use", self.source, "--target-dir", self.target)[0], 0) + provider = tomlkit.parse(path.read_text(encoding="utf-8"))["model_providers"]["cc_simple"] + for key in ("http_headers", "env_http_headers", "query_params", "aws", "gateway_oauth", "model_catalog_url", "env_key_instructions"): + self.assertNotIn(key, provider) + self.assertEqual(provider["request_max_retries"], 7) + + def test_claude_preserves_settings_and_removes_conflicting_auth(self): + self.target.mkdir() + path = self.target / "settings.json" + path.write_text(json.dumps({"permissions": {"allow": ["Read"]}, "env": {"OTHER": "retained", "ANTHROPIC_API_KEY": "old-key", "ANTHROPIC_AUTH_TOKEN": "old-bearer", "ANTHROPIC_MODEL": "old-model"}}), encoding="utf-8") + source = self.write_source(dict(self.config, type="anthropic", base_url="https://example.test")) + code, result, _ = self.run_cli("use", source, "--target-dir", self.target) + self.assertEqual(code, 0) + self.assertEqual(result["target"], "claude") + loaded = json.loads(path.read_text(encoding="utf-8")) + self.assertEqual(loaded["permissions"], {"allow": ["Read"]}) + self.assertEqual(loaded["env"]["OTHER"], "retained") + self.assertNotIn("ANTHROPIC_AUTH_TOKEN", loaded["env"]) + self.assertNotIn("ANTHROPIC_MODEL", loaded["env"]) + self.assertEqual(loaded["env"]["ANTHROPIC_API_KEY"], self.config["api_key"]) + self.assertEqual(loaded["model"], "test-model") + + def test_claude_environment_is_resolved_and_missing_value_fails(self): + config = dict(self.config, type="anthropic", env_key="TEST_MODEL_TOKEN") + del config["api_key"] + source = self.write_source(config) + with patch.dict(os.environ, {}, clear=True): + self.assertEqual(self.run_cli("use", source, "--target-dir", self.target)[0], 2) + self.assertFalse(self.target.exists()) + with patch.dict(os.environ, {"TEST_MODEL_TOKEN": "environment-secret"}): + self.assertEqual(self.run_cli("use", source, "--target-dir", self.target)[0], 0) + loaded = json.loads((self.target / "settings.json").read_text(encoding="utf-8")) + self.assertEqual(loaded["env"]["ANTHROPIC_API_KEY"], "environment-secret") + + def test_anthropic_codex_requires_explicit_responses_proxy(self): + source = self.write_source(dict(self.config, type="anthropic")) + self.assertEqual(self.run_cli("use", source, "--target", "codex", "--target-dir", self.target)[0], 2) + source = self.write_source(dict(self.config, type="anthropic", api_format="responses")) + self.assertEqual(self.run_cli("use", source, "--target", "codex", "--target-dir", self.target)[0], 0) + + def test_dry_run_does_not_write_or_leak_credentials(self): + code, output, _ = self.run_cli("use", self.source, "--target-dir", self.target, "--dry-run") + self.assertEqual(code, 0) + self.assertTrue(output["dry_run"]) + self.assertFalse(self.target.exists()) + self.assertNotIn(self.config["api_key"], json.dumps(output)) + + def test_invalid_existing_file_is_preserved_and_error_hides_source(self): + self.target.mkdir() + path = self.target / "config.toml" + original = 'token = "test-only-SECRET\n' + path.write_text(original, encoding="utf-8") + code, output, _ = self.run_cli("use", self.source, "--target-dir", self.target) + self.assertEqual(code, 2) + self.assertEqual(path.read_text(encoding="utf-8"), original) + self.assertNotIn(self.config["api_key"], json.dumps(output)) + self.assertEqual(len(list(self.target.glob("*.bak.*"))), 0) + + def test_failed_atomic_replace_keeps_original_and_cleans_temp(self): + self.target.mkdir() + path = self.target / "config.toml" + path.write_text("original", encoding="utf-8") + with patch("cc_simple.os.replace", side_effect=OSError("write failed")): + with self.assertRaises(OSError): + cli.atomic_write(path, "replacement") + self.assertEqual(path.read_text(encoding="utf-8"), "original") + self.assertEqual(list(self.target.iterdir()), [path]) + + def test_concurrent_add_never_overwrites_a_stored_configuration(self): + script = Path(cli.__file__).with_name("cc-simple.py") + environment = {**os.environ, "PYTHONUTF8": "1", "PYTHONIOENCODING": "utf-8"} + processes = [subprocess.Popen([sys.executable, str(script), "--json", "--config-dir", str(self.store), "add", str(self.source)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8", env=environment) for _ in range(4)] + results = [process.communicate(timeout=15) for process in processes] + self.assertTrue(all(process.returncode == 0 for process in processes), results) + identities = [json.loads(output)["id"] for output, _ in results] + self.assertEqual(sorted(identities), [1, 2, 3, 4]) + self.assertEqual(len(list(self.store.glob("*.yaml"))), 4) + + def test_check_calls_library_protocol_and_does_not_write(self): + calls = [] + fake = types.ModuleType("modeltrace") + fake.check_api = lambda **arguments: calls.append(arguments) or {"results": [], "families": [], "api_test": {"ok": True}, "used_outputs": 3} + with patch.dict(sys.modules, {"modeltrace": fake}): + code, output, _ = self.run_cli("check", self.source) + self.assertEqual(code, 0) + self.assertEqual(calls[-1]["api_format"], "responses") + self.assertEqual(calls[-1]["api_key"], self.config["api_key"]) + code, output, _ = self.run_cli("check", self.source, "--format", "openai", "--timeout", "7", "--temperature", "0") + self.assertEqual(calls[-1]["api_format"], "openai") + self.assertEqual(calls[-1]["timeout"], 7) + self.assertEqual(calls[-1]["temperature"], 0) + self.assertFalse(self.store.exists()) + self.assertFalse(self.target.exists()) + self.assertNotIn(self.config["api_key"], json.dumps(output)) + + def test_runtime_api_errors_are_nonzero_and_redacted(self): + fake = types.ModuleType("modeltrace") + def fail(**arguments): + raise RuntimeError("upstream rejected " + arguments["api_key"]) + fake.check_api = fail + with patch.dict(sys.modules, {"modeltrace": fake}): + code, output, _ = self.run_cli("check", self.source) + self.assertEqual(code, 1) + self.assertIn("", output["error"]) + self.assertNotIn(self.config["api_key"], output["error"]) + + def test_bad_temperature_and_library_value_error_are_clean(self): + self.assertEqual(self.run_cli("check", self.source, "--temperature", "-1")[0], 2) + fake = types.ModuleType("modeltrace") + def fail(**arguments): + raise ValueError("a secret value") + fake.check_api = fail + with patch.dict(sys.modules, {"modeltrace": fake}): + code, output, _ = self.run_cli("check", self.source) + self.assertEqual(code, 2) + self.assertNotIn("secret", output["error"]) + + def test_yaml_syntax_error_does_not_leak_credential_snippet(self): + self.source.write_text('api_key: [test-only-SECRET\n', encoding="utf-8") + code, output, _ = self.run_cli("add", self.source) + self.assertEqual(code, 2) + self.assertNotIn("test-only-SECRET", output["error"]) + + def test_real_entrypoint_returns_failure_exit_code(self): + script = Path(cli.__file__).with_name("cc-simple.py") + environment = {**os.environ, "PYTHONUTF8": "1", "PYTHONIOENCODING": "utf-8"} + process = subprocess.run([sys.executable, str(script), "--json", "--config-dir", str(self.store), "use", "999"], capture_output=True, encoding="utf-8", env=environment, timeout=15) + self.assertEqual(process.returncode, 2) + self.assertEqual(json.loads(process.stdout)["exit_code"], 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_library_integration.py b/tests/test_library_integration.py new file mode 100644 index 0000000..46c57de --- /dev/null +++ b/tests/test_library_integration.py @@ -0,0 +1,114 @@ +"""Actual CLI subprocess + installed ModelTrace + local mock API integration.""" +import json +import os +import subprocess +import sys +import sysconfig +import tempfile +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import yaml + +try: + import modeltrace +except ImportError: + modeltrace = None + + +class MockAPI(BaseHTTPRequestHandler): + def do_POST(self): + request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + if self.server.behavior == "auth": + self.send_response(401) + payload = {"error": {"message": "test rejection"}} + else: + self.send_response(200) + count = 800 + text = "1 2 3" if self.server.behavior == "short" else ",".join(str((i * 17 + 13) % 1000) for i in range(count)) + if self.path.endswith("/responses"): + payload = {"output": [{"type": "message", "content": [{"type": "output_text", "text": text}]}]} + elif self.path.endswith("/messages"): + payload = {"content": [{"type": "text", "text": text}]} + else: + payload = {"choices": [{"message": {"content": text}}]} + self.server.requests.append((self.path, request)) + encoded = json.dumps(payload).encode("utf-8") + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def log_message(self, *arguments): + pass + + +@unittest.skipIf(modeltrace is None, "installed ModelTrace library is required") +class LibraryIntegrationTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.server = ThreadingHTTPServer(("127.0.0.1", 0), MockAPI) + self.server.behavior = "valid" + self.server.requests = [] + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + self.addCleanup(self.stop_server) + self.source = self.root / "provider.yaml" + self.config = {"name": "integration", "base_url": f"http://127.0.0.1:{self.server.server_port}/v1", + "type": "openai", "api_key": "local-integration-only", "model_name": "local-mock"} + self.source.write_text(yaml.safe_dump(self.config), encoding="utf-8") + + def stop_server(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(2) + + def command(self, *arguments): + # Run the installed console script away from the repository. + binary = Path(sysconfig.get_path("scripts")) / ("cc-simple.exe" if sys.platform == "win32" else "cc-simple") + environment = {**os.environ, "PYTHONUTF8": "1", "PYTHONIOENCODING": "utf-8"} + result = subprocess.run([str(binary), "--json", "--config-dir", str(self.root / "store"), "check", str(self.source), "--timeout", "5", "--attempts", "3", *arguments], cwd=self.root, env=environment, capture_output=True, encoding="utf-8", timeout=30) + self.assertNotIn("Traceback", result.stderr) + self.assertNotIn(self.config["api_key"], result.stdout + result.stderr) + return result.returncode, json.loads(result.stdout) + + def test_responses_default_full_fingerprint(self): + code, result = self.command() + self.assertEqual(code, 0, result) + self.assertEqual(result["analysis"]["used_outputs"], 3) + self.assertEqual(result["analysis"]["api_test"]["api_format"], "responses") + self.assertEqual({item["model"] for item in result["analysis"]["results"]}, + {item["id"] for item in modeltrace.load_bank()["models"]}) + self.assertEqual([path for path, _ in self.server.requests], ["/v1/responses"] * 3) + self.assertFalse((self.root / "store").exists()) + + def test_chat_protocol_override(self): + code, result = self.command("--format", "openai") + self.assertEqual(code, 0, result) + self.assertEqual(result["analysis"]["api_test"]["api_format"], "openai") + self.assertEqual([path for path, _ in self.server.requests], ["/v1/chat/completions"] * 3) + + def test_anthropic_default_protocol(self): + self.config["type"] = "anthropic" + self.config["base_url"] = self.config["base_url"].removesuffix("/v1") + self.source.write_text(yaml.safe_dump(self.config), encoding="utf-8") + code, result = self.command() + self.assertEqual(code, 0, result) + self.assertEqual(result["analysis"]["api_test"]["api_format"], "anthropic") + self.assertEqual([path for path, _ in self.server.requests], ["/v1/messages"] * 3) + + def test_auth_failure_and_short_outputs_return_nonzero(self): + for behavior in ("auth", "short"): + with self.subTest(behavior=behavior): + self.server.behavior = behavior + code, result = self.command() + self.assertEqual(code, 1) + self.assertIn("error", result) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_review_regressions.py b/tests/test_review_regressions.py new file mode 100644 index 0000000..fb72293 --- /dev/null +++ b/tests/test_review_regressions.py @@ -0,0 +1,165 @@ +"""Regressions for the independent review's URL, integer and YAML findings.""" + +import contextlib +import io +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +import tomlkit +import yaml + +import cc_simple as cli +import test_library_integration as integration + + +class ConfigRegressionTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.config = {"name": "regression", "base_url": "https://example.test/v1", + "type": "openai", "api_key": "fake-only", "model_name": "mock"} + + def test_codex_base_shapes_write_the_same_api_prefix(self): + cases = {"": "/v1", "/": "/v1", "/v1": "/v1", "/v1/responses/": "/v1", + "/gateway/v1": "/gateway/v1", "/gateway/v1/responses": "/gateway/v1", + "/gateway": "/gateway", "/gateway/responses": "/gateway", + "/v1/chat/completions": "/v1"} + for suffix, expected in cases.items(): + with self.subTest(suffix=suffix): + config = dict(self.config, base_url="https://example.test" + suffix) + target = self.root / str(len(list(self.root.iterdir()))) + result = cli.switch(config, target_dir=target) + provider = tomlkit.parse((target / "config.toml").read_text(encoding="utf-8"))["model_providers"]["cc_simple"] + self.assertEqual(provider["base_url"], "https://example.test" + expected) + self.assertEqual(result["base_url"], provider["base_url"]) + self.assertEqual(config["base_url"], "https://example.test" + suffix) + + def test_claude_base_shapes_preserve_gateway_prefix(self): + cases = {"": "", "/": "", "/v1": "", "/v1/messages/": "", + "/gateway": "/gateway", "/gateway/v1": "/gateway", + "/gateway/v1/messages": "/gateway", "/tenant/v1/edge": "/tenant/v1/edge"} + for suffix, expected in cases.items(): + with self.subTest(suffix=suffix): + config = dict(self.config, type="anthropic", base_url="https://example.test" + suffix) + target = self.root / str(len(list(self.root.iterdir()))) + result = cli.switch(config, target_dir=target) + settings = json.loads((target / "settings.json").read_text(encoding="utf-8")) + self.assertEqual(settings["env"]["ANTHROPIC_BASE_URL"], "https://example.test" + expected) + self.assertEqual(result["base_url"], settings["env"]["ANTHROPIC_BASE_URL"]) + + def test_ambiguous_message_endpoints_fail_before_writing(self): + for suffix in ("/messages", "/gateway/messages", "/v1/responses"): + with self.subTest(suffix=suffix): + config = dict(self.config, type="anthropic", base_url="https://example.test" + suffix) + with self.assertRaises(cli.ConfigError): + cli.switch(config, target_dir=self.root / "target") + self.assertFalse((self.root / "target").exists()) + + def test_codex_unversioned_full_endpoints_are_explicitly_rejected(self): + for suffix in ("/responses", "/chat/completions"): + with self.subTest(suffix=suffix): + with self.assertRaises(cli.ConfigError): + cli.switch(dict(self.config, base_url="https://example.test" + suffix), + target_dir=self.root / "target") + self.assertFalse((self.root / "target").exists()) + + def test_huge_temperature_returns_json_error_without_touching_files(self): + source = self.root / "huge.yaml" + source.write_text(yaml.safe_dump(dict(self.config, temperature=10 ** 400)), encoding="utf-8") + target = self.root / "target" + target.mkdir() + active = target / "config.toml" + original = b'# kept\r\nmodel="existing"\r\n' + active.write_bytes(original) + script = Path(cli.__file__).with_name("cc-simple.py") + for command, extra in (("add", []), ("check", []), ("use", ["--target-dir", str(target)])): + with self.subTest(command=command): + process = subprocess.run([sys.executable, str(script), "--json", "--config-dir", + str(self.root / "store"), command, str(source), *extra], cwd=self.root, + env={**os.environ, "PYTHONUTF8": "1", "PYTHONIOENCODING": "utf-8"}, + capture_output=True, encoding="utf-8", timeout=15) + self.assertEqual(process.returncode, 2, process.stderr) + self.assertEqual(json.loads(process.stdout)["exit_code"], 2) + self.assertNotIn("Traceback", process.stderr) + self.assertEqual(active.read_bytes(), original) + self.assertEqual(list(target.iterdir()), [active]) + self.assertFalse((self.root / "store").exists()) + + def test_yaml_merge_defaults_and_explicit_overrides(self): + documents = [ + '<<: &defaults {name: merged, base_url: "https://example.test/v1", type: openai, api_key: fake-only}\nmodel_name: mock\n', + '<<: [{name: first, base_url: "https://example.test/v1", type: openai, api_key: fake-only}, {name: second, model_name: mock}]\nname: override\n', + '<<: {<<: {name: inner, type: openai, api_key: fake-only}, name: middle, base_url: "https://example.test/v1"}\nname: outer\n', + ] + for document in documents: + with self.subTest(document=document): + source = self.root / "merge.yaml" + source.write_text(document, encoding="utf-8") + expected = yaml.safe_load(document) + loaded = cli.read_config(source) + self.assertEqual(loaded["name"], expected["name"]) + self.assertEqual(loaded["api_key"], "fake-only") + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli.main(["--json", "--config-dir", str(self.root / "store"), "add", str(source)]) + self.assertEqual(code, 0, output.getvalue()) + stored = yaml.safe_load(Path(json.loads(output.getvalue())["path"]).read_text(encoding="utf-8")) + self.assertEqual(stored, loaded) + + def test_merge_alias_reuse_and_duplicate_explicit_keys(self): + document = 'first: &base {name: old, <<: {name: default}}\nsecond: {<<: *base, name: new}\nthird: {<<: *base}\n' + self.assertEqual(yaml.load(document, Loader=cli.UniqueLoader), yaml.safe_load(document)) + for document in ('name: a\nname: b\n', '<<: {name: a, name: b}\n', + '<<: {name: a}\nname: b\nname: c\n', + '<<: {name: a}\n<<: {name: b}\n', '<<: {1: bad}\n'): + with self.subTest(document=document): + with self.assertRaises(cli.ConfigError): + yaml.load(document, Loader=cli.UniqueLoader) + + +@unittest.skipIf(integration.modeltrace is None, "installed ModelTrace library is required") +class AddressIntegrationTests(unittest.TestCase): + """Installed CLI checks use the same prefixes as the generated client config.""" + + setUp = integration.LibraryIntegrationTests.setUp + stop_server = integration.LibraryIntegrationTests.stop_server + command = integration.LibraryIntegrationTests.command + + def test_codex_check_endpoint_matrix(self): + host = f"http://127.0.0.1:{self.server.server_port}" + cases = {"": "/v1/responses", "/v1": "/v1/responses", "/v1/responses": "/v1/responses", + "/gateway": "/gateway/responses", "/gateway/v1": "/gateway/v1/responses", + "/gateway/v1/responses": "/gateway/v1/responses"} + for suffix, expected in cases.items(): + with self.subTest(suffix=suffix): + self.server.requests.clear() + self.config["base_url"] = host + suffix + self.source.write_text(yaml.safe_dump(self.config), encoding="utf-8") + code, result = self.command() + self.assertEqual(code, 0, result) + self.assertEqual([path for path, _ in self.server.requests], [expected] * 3) + + def test_claude_check_endpoint_matrix(self): + host = f"http://127.0.0.1:{self.server.server_port}" + cases = {"": "/v1/messages", "/v1": "/v1/messages", "/v1/messages": "/v1/messages", + "/gateway": "/gateway/v1/messages", "/gateway/v1": "/gateway/v1/messages", + "/gateway/v1/messages": "/gateway/v1/messages"} + self.config["type"] = "anthropic" + for suffix, expected in cases.items(): + with self.subTest(suffix=suffix): + self.server.requests.clear() + self.config["base_url"] = host + suffix + self.source.write_text(yaml.safe_dump(self.config), encoding="utf-8") + code, result = self.command() + self.assertEqual(code, 0, result) + self.assertEqual([path for path, _ in self.server.requests], [expected] * 3) + + +if __name__ == "__main__": + unittest.main()