diff --git a/.github/workflows/binary-release.yml b/.github/workflows/binary-release.yml new file mode 100644 index 0000000000..a6af93a26f --- /dev/null +++ b/.github/workflows/binary-release.yml @@ -0,0 +1,150 @@ +name: Build and Release with PyInstaller + +on: + push: + tags: + - "*" + +jobs: + build: + strategy: + matrix: + os: [ ubuntu-latest, macos-latest, windows-latest ] + + runs-on: ${{ matrix.os }} + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + # 二进制包仅依赖精简版 SDK（tencentcloud-sdk-python-common），不安装完整 SDK。 + # 使用 --no-deps 安装 tccli 本体，避免 setup.py 声明的 tencentcloud-sdk-python（完整 SDK）被拉入， + # 从而显著减小二进制体积。存量 pip 安装不走此流程，行为不受影响。 + - name: Install dependencies (lightweight common SDK for binary) + run: | + python -m pip install --upgrade pip + pip install pyinstaller + pip install tencentcloud-sdk-python-common jmespath six cos-python-sdk-v5 + pip install -e . --no-deps + + # 裁剪各服务的 *_client.py。删除后，运行时 command.py 中 + # Services.action_caller() 会抛 ImportError，从而自动降级到基于 CommonClient + # 的 GenericActionCaller。api.json / examples.json 等元数据保留，供 GenericActionCaller 使用。 + # 使用 Python 实现，保证在 Linux/macOS/Windows 三平台通用。 + - name: Strip per-service *_client.py (binary uses CommonClient) + run: python -c "import glob, os; files = glob.glob('tccli/services/**/*_client.py', recursive=True); [os.remove(f) for f in files]; print('removed %d client files' % len(files))" + + # 使用目录模式(-D)打包，避免单文件模式每次运行都解压的启动延迟。 + - name: Run PyInstaller + run: pyinstaller tccli/main.py -y -D --collect-all tccli --exclude-module tccli.examples --name tccli + + # --collect-all 可能从已安装包中收集到 *_client.py，打包后二次裁剪确保干净 + - name: Strip *_client.py from dist (post-build cleanup) + run: python -c "import glob, os; files = glob.glob('dist/tccli/_internal/tccli/services/**/*_client.py', recursive=True); [os.remove(f) for f in files]; print('post-build removed %d client files' % len(files))" + + # macOS 会校验主程序与其加载的 Python.framework / dylib 签名是否兼容。 + # 按“内部 Mach-O -> Python.framework -> 主程序”的顺序统一做 ad-hoc 签名； + # 主程序最后签名，避免后续修改内部文件导致签名失效。 + - name: Re-sign macOS onedir bundle + if: runner.os == 'macOS' + shell: bash + run: | + set -euo pipefail + DIST_DIR="dist/tccli" + + # 先签 _internal 中的动态库、扩展模块和 Framework 可执行文件。 + while IFS= read -r -d '' file_path; do + if file -b "$file_path" | grep -q 'Mach-O'; then + echo "Signing Mach-O: $file_path" + codesign --remove-signature "$file_path" 2>/dev/null || true + codesign --force --sign - --timestamp=none "$file_path" + fi + done < <(find "$DIST_DIR/_internal" -type f -print0) + + # 再签 Framework 容器，使其嵌套签名与内部 Python 可执行文件一致。 + if [[ -d "$DIST_DIR/_internal/Python.framework" ]]; then + codesign --force --deep --sign - --timestamp=none \ + "$DIST_DIR/_internal/Python.framework" + fi + + # 下载后的 ad-hoc 程序没有 Apple Team ID；启用 hardened runtime 时，dyld 默认会 + # 拒绝加载无相同 Team ID 的 Python.framework。显式关闭主程序的 Library Validation， + # 同时保留其它 hardened-runtime 保护。 + ENTITLEMENTS="$RUNNER_TEMP/tccli-entitlements.plist" + cat > "$ENTITLEMENTS" <<'PLIST' + + + + + com.apple.security.cs.disable-library-validation + + + + PLIST + + # 最后签主程序，并嵌入允许加载随包 Python.framework 的 entitlement。 + codesign --remove-signature "$DIST_DIR/tccli" 2>/dev/null || true + codesign --force --sign - --timestamp=none --options runtime \ + --entitlements "$ENTITLEMENTS" "$DIST_DIR/tccli" + + # 在上传 Release 前验证签名、entitlement，并实际启动一次；失败则终止工作流。 + codesign --verify --strict --verbose=4 "$DIST_DIR/tccli" + if [[ -f "$DIST_DIR/_internal/Python.framework/Versions/3.12/Python" ]]; then + codesign --verify --strict --verbose=4 \ + "$DIST_DIR/_internal/Python.framework/Versions/3.12/Python" + fi + codesign -d --entitlements :- "$DIST_DIR/tccli" 2>&1 | \ + grep -q 'com.apple.security.cs.disable-library-validation' + "$DIST_DIR/tccli" --version + + - name: Package binary (Linux) + if: runner.os == 'Linux' + working-directory: dist + run: tar -czf tccli-linux.tar.gz tccli/ + + - name: Package binary (macOS) + if: runner.os == 'macOS' + working-directory: dist + run: tar -czf tccli-macos.tar.gz tccli/ + + - name: Package binary (Windows) + if: runner.os == 'Windows' + run: Compress-Archive -Path "dist\tccli" -DestinationPath "dist\tccli-windows.zip" + shell: powershell + + - name: Upload artifact + uses: actions/upload-artifact@v4 + with: + name: "tccli-${{ runner.os }}" + path: | + dist/*.tar.gz + dist/*.zip + + release: + needs: build + runs-on: ubuntu-latest + + steps: + - name: Download all artifacts + uses: actions/download-artifact@v4 + with: + path: artifacts + + - name: Create GitHub Release + id: create_release + uses: softprops/action-gh-release@v2 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + with: + tag_name: ${{ github.ref_name }} + name: Release ${{ github.ref_name }} + draft: false + prerelease: false + files: | + artifacts/**/*.tar.gz + artifacts/**/*.zip diff --git a/tccli/action_caller.py b/tccli/action_caller.py new file mode 100644 index 0000000000..ef7aa8f60d --- /dev/null +++ b/tccli/action_caller.py @@ -0,0 +1,253 @@ +# -*- coding:utf-8 -*- +import json +import os +import os.path as path +import time + +import six +from jmespath import search +from tencentcloud.common import credential +from tencentcloud.common.common_client import CommonClient +from tencentcloud.common.profile.client_profile import ClientProfile +from tencentcloud.common.profile.http_profile import HttpProfile + +import tccli.format_output as format_output +import tccli.options_define as options_define +from tccli import __version__ +from tccli.exceptions import ConfigurationError, ClientError +from tccli.loaders import Loader, BASE_TYPE +from tccli.utils import Utils + + +class GenericActionCaller(object): + def __init__(self, module, action): + self._module = module + self._action = action + self._avail_vers = [] + + def __call__(self, args, parsed_globals): + self._call_json(args, parsed_globals) + + def available_versions(self): + if not self._avail_vers: + svc_path = os.path.join(path.dirname(path.abspath(__file__)), 'services', self._module) + try: + dirs = os.listdir(svc_path) + except OSError: + raise ConfigurationError("service '%s' not found" % self._module) + self._avail_vers = [d for d in dirs if d[0] == "v" and os.path.isdir(path.join(svc_path, d))] + + return self._avail_vers + + @staticmethod + def convert_version_str(ver): + return ver[1:5] + "-" + ver[5:7] + "-" + ver[7:9] + + def _call_json(self, args, parsed_globals): + g_param = self.parse_global_arg(parsed_globals) + + if g_param[options_define.UseCVMRole.replace('-', '_')]: + cred = credential.CVMRoleCredential() + elif g_param[options_define.RoleArn.replace('-', '_')] and g_param[ + options_define.RoleSessionName.replace('-', '_')]: + cred = credential.STSAssumeRoleCredential( + g_param[options_define.SecretId], g_param[options_define.SecretKey], + g_param[options_define.RoleArn.replace('-', '_')], + g_param[options_define.RoleSessionName.replace('-', '_')], endpoint=g_param["sts_cred_endpoint"] + ) + elif os.getenv(options_define.ENV_TKE_REGION) and os.getenv(options_define.ENV_TKE_PROVIDER_ID) and os.getenv( + options_define.ENV_TKE_WEB_IDENTITY_TOKEN_FILE) and os.getenv(options_define.ENV_TKE_ROLE_ARN): + cred = credential.DefaultTkeOIDCRoleArnProvider().get_credentials() + else: + cred = credential.Credential( + g_param[options_define.SecretId], g_param[options_define.SecretKey], g_param[options_define.Token] + ) + http_profile = HttpProfile( + reqTimeout=60 if g_param[options_define.Timeout] is None else int(g_param[options_define.Timeout]), + reqMethod="POST", + endpoint=g_param[options_define.Endpoint], + proxy=g_param[options_define.HttpsProxy.replace('-', '_')] + ) + cpf = ClientProfile(httpProfile=http_profile, signMethod="HmacSHA256") + if g_param[options_define.Language]: + cpf.language = g_param[options_define.Language] + version = self.convert_version_str(g_param[options_define.Version]) + region = g_param[options_define.Region] + client = CommonClient(self._module, version, cred, region, cpf) + client._sdkVersion += ("_CLI_" + __version__) + + start_time = time.time() + while True: + raw = client.call_json(self._action, args) + json_obj = raw.get("Response", raw) if isinstance(raw, dict) else raw + json_obj = self._filter_response(json_obj, g_param[options_define.Version]) + + if not g_param[options_define.Waiter] or search(g_param['OptionsDefine.WaiterInfo']['expr'], json_obj) == \ + g_param['OptionsDefine.WaiterInfo']['to']: + break + + cur_time = time.time() + if cur_time - start_time >= g_param['OptionsDefine.WaiterInfo']['timeout']: + raise ClientError('Request timeout, wait `%s` to `%s` timeout, last request is %s' % + (g_param['OptionsDefine.WaiterInfo']['expr'], + g_param['OptionsDefine.WaiterInfo']['to'], + search(g_param['OptionsDefine.WaiterInfo']['expr'], json_obj))) + else: + print('Inquiry result is %s.' % search(g_param['OptionsDefine.WaiterInfo']['expr'], json_obj)) + time.sleep(g_param['OptionsDefine.WaiterInfo']['interval']) + + format_output.output("action", json_obj, g_param[options_define.Output], g_param[options_define.Filter]) + + def parse_global_arg(self, parsed_globals): + g_param = parsed_globals + cvm_role_flag = True + for param in parsed_globals.keys(): + if param in [options_define.SecretKey, options_define.SecretId, options_define.RoleArn, + options_define.RoleSessionName]: + if parsed_globals[param] is not None: + cvm_role_flag = False + break + is_exist_profile = True + if not parsed_globals["profile"]: + is_exist_profile = False + g_param["profile"] = os.environ.get("TCCLI_PROFILE", "default") + + configure_path = os.path.join(os.path.expanduser("~"), ".tccli") + is_conf_exist, conf_path = Utils.file_existed(configure_path, g_param["profile"] + ".configure") + is_cred_exist, cred_path = Utils.file_existed(configure_path, g_param["profile"] + ".credential") + + conf = {} + cred = {} + + if is_conf_exist: + conf = Utils.load_json_msg(conf_path) + if is_cred_exist: + cred = Utils.load_json_msg(cred_path) + + if not (isinstance(conf, dict) and isinstance(cred, dict)): + raise ConfigurationError( + "file: %s or %s is not json format" + % (g_param["profile"] + ".configure", g_param["profile"] + ".credential")) + + if options_define.Token not in cred: + cred[options_define.Token] = None + + if not is_exist_profile: + if os.environ.get(options_define.ENV_SECRET_ID) and os.environ.get(options_define.ENV_SECRET_KEY): + cred[options_define.SecretId] = os.environ.get(options_define.ENV_SECRET_ID) + cred[options_define.SecretKey] = os.environ.get(options_define.ENV_SECRET_KEY) + cred[options_define.Token] = os.environ.get(options_define.ENV_TOKEN) + cvm_role_flag = False + + if os.environ.get(options_define.ENV_REGION): + conf[options_define.SysParam][options_define.Region] = os.environ.get(options_define.ENV_REGION) + + if os.environ.get(options_define.ENV_ROLE_ARN) and os.environ.get(options_define.ENV_ROLE_SESSION_NAME): + cred[options_define.RoleArn] = os.environ.get(options_define.ENV_ROLE_ARN) + cred[options_define.RoleSessionName] = os.environ.get(options_define.ENV_ROLE_SESSION_NAME) + cvm_role_flag = False + + if cvm_role_flag: + if "type" in cred and cred["type"] == "cvm-role": + g_param[options_define.UseCVMRole.replace('-', '_')] = True + + for param in g_param.keys(): + if g_param[param] is None: + if param in [options_define.SecretKey, options_define.SecretId, options_define.Token]: + if param in cred: + g_param[param] = cred[param] + elif not (g_param[options_define.UseCVMRole.replace('-', '_')] + or os.getenv(options_define.ENV_TKE_ROLE_ARN)): + raise ConfigurationError("%s is invalid" % param) + elif param in [options_define.Region, options_define.Output, options_define.Language]: + if param in conf[options_define.SysParam]: + g_param[param] = conf[options_define.SysParam][param] + elif param != options_define.Language: + raise ConfigurationError("%s is invalid" % param) + elif param.replace('_', '-') in [options_define.RoleArn, options_define.RoleSessionName]: + if param.replace('_', '-') in cred: + g_param[param] = cred[param.replace('_', '-')] + + try: + if g_param[options_define.ServiceVersion]: + g_param[options_define.Version] = "v" + g_param[options_define.ServiceVersion].replace('-', '') + else: + version = conf[self._module][options_define.Version] + g_param[options_define.Version] = "v" + version.replace('-', '') + + if g_param[options_define.Endpoint] is None: + g_param[options_define.Endpoint] = conf[self._module][options_define.Endpoint] + g_param["sts_cred_endpoint"] = conf.get("sts", {}).get("endpoint") + except Exception as err: + raise ConfigurationError("config file:%s error, %s" % (conf_path, str(err))) + + if g_param[options_define.Version] not in self.available_versions(): + raise Exception("available versions: %s" % " ".join(self.available_versions())) + + if g_param[options_define.Waiter]: + try: + param = json.loads(g_param[options_define.Waiter]) + except ValueError as e: + raise Exception('`--waiter` must be a valid JSON string: %s' % str(e)) + if 'expr' not in param: + raise Exception('`expr` in `--waiter` must be defined') + if 'to' not in param: + raise Exception('`to` in `--waiter` must be defined') + if 'timeout' not in param: + if 'waiter' in conf and 'timeout' in conf['waiter']: + param['timeout'] = conf['waiter']['timeout'] + else: + param['timeout'] = 180 + if 'interval' not in param: + if 'waiter' in conf and 'interval' in conf['waiter']: + param['interval'] = conf['waiter']['interval'] + else: + param['interval'] = 5 + param['interval'] = min(param['interval'], param['timeout']) + g_param['OptionsDefine.WaiterInfo'] = param + + if six.PY2: + for key, value in g_param.items(): + if isinstance(value, six.text_type): + g_param[key] = value.encode('utf-8') + return g_param + + def _filter_response(self, data, version): + """加载 api.json 中 ActionResponse 的 schema，对响应 dict 做白名单字段投影""" + try: + loader = Loader() + service_model = loader.get_service_model(self._module, self.convert_version_str(version)) + objects = service_model["objects"] + resp_name = self._action + "Response" + if resp_name not in objects: + return data + return self._filter_by_schema(data, objects[resp_name]["members"], objects) + except Exception: + return data + + def _filter_by_schema(self, data, members, objects): + """按 schema members 递归投影 dict，丢弃未定义字段，复杂类型递归处理""" + if not isinstance(data, dict): + return data + result = {} + for para in members: + name = para["name"] + if name not in data: + continue + value = data[name] + if para["type"] == "list": + if para["member"] not in BASE_TYPE: + if isinstance(value, list): + sub_members = objects[para["member"]]["members"] + result[name] = [self._filter_by_schema(item, sub_members, objects) for item in value] + else: + result[name] = value + else: + result[name] = value + else: + if para["member"] not in BASE_TYPE: + sub_members = objects[para["member"]]["members"] + result[name] = self._filter_by_schema(value, sub_members, objects) + else: + result[name] = value + return result diff --git a/tccli/command.py b/tccli/command.py index fe94f6d113..da3c640e0f 100644 --- a/tccli/command.py +++ b/tccli/command.py @@ -8,6 +8,7 @@ from collections import OrderedDict from tccli import credentials +from tccli.action_caller import GenericActionCaller from tccli.utils import Utils from tccli.argument import CLIArgument, CustomArgument, ListArgument, BooleanArgument from tccli.exceptions import UnknownArgumentError @@ -176,11 +177,21 @@ def _get_command_map(self): def _build_command_map(self): command_map = OrderedDict() service_model = self._get_service_model() + # 优先使用各服务的 *_client.py（存量 pip 安装行为，依赖完整 SDK）； + # 若 client.py 不可用（例如二进制包已裁剪掉 *_client.py），则降级到 + # 基于 CommonClient 的 GenericActionCaller。存量 pip 安装 client.py 始终存在，行为保持不变。 + try: + service_action_caller = Services.action_caller(self._service_name)() + except ImportError: + service_action_caller = None for action in service_model["actions"]: action_model = service_model["actions"][action] action_caller = action_model.get("action_caller", None) if not action_caller: - action_caller = Services.action_caller(self._service_name)()[action] + if service_action_caller is not None: + action_caller = service_action_caller[action] + else: + action_caller = GenericActionCaller(self._service_name, action) command_map[action] = ActionCommand( service_name=self._service_name, version=self._version, diff --git a/tccli/plugins/test/add.py b/tccli/plugins/test/add.py index d3475a6345..8516df5cb9 100644 --- a/tccli/plugins/test/add.py +++ b/tccli/plugins/test/add.py @@ -4,7 +4,7 @@ from tencentcloud.common import credential from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException -from tencentcloud.cvm.v20170312 import cvm_client, models +from tencentcloud.common.common_client import CommonClient def add_command(args, parsed_globals): @@ -21,11 +21,10 @@ def add_command(args, parsed_globals): # do api call with secret key cred = credential.Credential(secret_id, secret_key, token) - cli = cvm_client.CvmClient(cred, region) + cli = CommonClient("cvm", "2017-03-12", cred, region) - req = models.DescribeInstancesRequest() try: - resp = cli.DescribeInstances(req) - print(resp.to_json_string(indent=2)) + resp = cli.call("DescribeInstances", {"Limit": 10}) + print(json.dumps(resp)) except TencentCloudSDKException as e: logging.exception(e) diff --git a/tests/test_action_caller.py b/tests/test_action_caller.py new file mode 100644 index 0000000000..30bfb98815 --- /dev/null +++ b/tests/test_action_caller.py @@ -0,0 +1,19 @@ +# -*- coding: utf-8 -*- +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from tccli.action_caller import GenericActionCaller + + +# ── convert_version_str ─────────────────────────────────────────────────────── + +@pytest.mark.parametrize("ver_input,expected", [ + ("v20170312", "2017-03-12"), + ("v20230101", "2023-01-01"), +]) +def test_convert_version_str(ver_input, expected): + assert GenericActionCaller.convert_version_str(ver_input) == expected diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000000..e180b63120 --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,10 @@ +# -*- coding:utf8 -*- + +from utils import shell, recover_profile + + +@recover_profile() +def test_auth(): + assert "在浏览器中转到以下链接, 并根据提示完成登录" in shell("tccli auth login --browser no") + + assert "登出成功, 密钥凭证已被删除:" in shell("tccli auth logout") diff --git a/tests/test_configure.py b/tests/test_configure.py new file mode 100644 index 0000000000..5080c444a7 --- /dev/null +++ b/tests/test_configure.py @@ -0,0 +1,147 @@ +# -*- coding: utf-8 -*- +import io +import json +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from tccli.configure import ConfigureGetCommand, ConfigureSetCommand, ConfigureListCommand +from tccli.exceptions import ConfigurationError, ParamError +from utils import shell, recover_profile + + +# ── ConfigureSetCommand：output / language 非法值回退 ───────────────────────── + +def _make_set_cmd(): + return ConfigureSetCommand() + + +@pytest.mark.parametrize("bad_output", ["xml", "yaml", ""]) +def test_configure_set_invalid_output_falls_back_to_json(tmp_path, monkeypatch, bad_output, capsys): + monkeypatch.setattr("tccli.configure.BasicConfigure.cli_path", str(tmp_path), raising=False) + cmd = _make_set_cmd() + monkeypatch.setattr(cmd, "cli_path", str(tmp_path)) + + class FakeArgs: + varname = ["output", bad_output] + + class FakeGlobals: + profile = "unit_test" + + cmd._run_main(FakeArgs(), FakeGlobals()) + _, conf_path = cmd._profile_existed("unit_test.configure") + data = json.loads(open(conf_path).read()) + assert data["_sys_param"]["output"] == "json" + + +@pytest.mark.parametrize("bad_lang", ["fr-FR", "ja-JP", ""]) +def test_configure_set_invalid_language_falls_back_to_zh_cn(tmp_path, monkeypatch, bad_lang): + cmd = _make_set_cmd() + monkeypatch.setattr(cmd, "cli_path", str(tmp_path)) + + class FakeArgs: + varname = ["language", bad_lang] + + class FakeGlobals: + profile = "unit_test" + + cmd._run_main(FakeArgs(), FakeGlobals()) + _, conf_path = cmd._profile_existed("unit_test.configure") + data = json.loads(open(conf_path).read()) + assert data["_sys_param"]["language"] == "zh-CN" + + +def test_configure_set_odd_varname_raises(tmp_path, monkeypatch): + cmd = _make_set_cmd() + monkeypatch.setattr(cmd, "cli_path", str(tmp_path)) + + class FakeArgs: + varname = ["region"] # 奇数个参数 + + class FakeGlobals: + profile = "unit_test" + + with pytest.raises(ParamError): + cmd._run_main(FakeArgs(), FakeGlobals()) + + +# ── ConfigureGetCommand：字段不存在时抛 ConfigurationError ──────────────────── + +def test_configure_get_nonexistent_key_raises(tmp_path, monkeypatch): + cmd = ConfigureGetCommand() + monkeypatch.setattr(cmd, "cli_path", str(tmp_path)) + + conf_path = tmp_path / "unit_test.configure" + conf_path.write_text(json.dumps({"_sys_param": {"region": "ap-guangzhou"}})) + cred_path = tmp_path / "unit_test.credential" + cred_path.write_text(json.dumps({})) + + class FakeArgs: + varname = ["nonexistent_key"] + + class FakeGlobals: + profile = "unit_test" + + with pytest.raises(ConfigurationError): + cmd._run_main(FakeArgs(), FakeGlobals()) + + +def test_configure_get_existing_key_outputs_value(tmp_path, monkeypatch): + stream = io.StringIO() + cmd = ConfigureGetCommand(stream=stream) + monkeypatch.setattr(cmd, "cli_path", str(tmp_path)) + + conf_path = tmp_path / "unit_test.configure" + conf_path.write_text(json.dumps({"_sys_param": {"region": "ap-beijing"}})) + cred_path = tmp_path / "unit_test.credential" + cred_path.write_text(json.dumps({})) + + class FakeArgs: + varname = ["region"] + + class FakeGlobals: + profile = "unit_test" + + cmd._run_main(FakeArgs(), FakeGlobals()) + assert "region = ap-beijing" in stream.getvalue() + + +# ── 集成测试（依赖已安装的 tccli，不发真实 API 请求）──────────────────────────── + +def test_configure_list_output_structure(): + output = shell("tccli configure list") + assert "credential:" in output + assert "configure:" in output + assert "cvm.version =" in output + assert "cvm.endpoint =" in output + + +@recover_profile() +def test_configure_get_set_region(): + shell("tccli configure set region ap-guangzhou") + assert "region = ap-guangzhou" in shell("tccli configure get region") + + +@recover_profile() +def test_configure_set_output_valid_values(): + for fmt in ["json", "text", "table"]: + shell("tccli configure set output %s" % fmt) + assert ("output = %s" % fmt) in shell("tccli configure get output") + + +@recover_profile() +def test_configure_set_secretid_and_get(): + sec_id = "AKIDabcdefghijklmn1234" + shell("tccli configure set secretId %s" % sec_id) + output = shell("tccli configure get secretId") + assert sec_id in output + + +@recover_profile("user2") +def test_configure_profile_isolation(): + shell("tccli configure --profile user2 set region ap-shanghai") + assert "region = ap-shanghai\n" == shell("tccli configure --profile user2 get region") + assert "region = ap-shanghai\n" == shell("TCCLI_PROFILE=user2 tccli configure get region") diff --git a/tests/test_cvm.py b/tests/test_cvm.py index afde580189..306f47499d 100644 --- a/tests/test_cvm.py +++ b/tests/test_cvm.py @@ -1,76 +1,36 @@ -from multiprocessing import Process, Queue -from tccli.command import CLICommand -from utils import TestCli +from utils import shell + def test_describe_regions(): - cmd = 'tccli cvm DescribeRegions' - expect = "\"Region\": \"ap-guangzhou\"" - test_cli = TestCli() - test_cli.equal(cmd, expect) + assert "\"Region\": \"ap-guangzhou\"" in shell("tccli cvm DescribeRegions") def test_describe_instances(): - cmd = 'tccli cvm DescribeInstances' - expect = "\"TotalCount\":" - test_cli = TestCli() - test_cli.equal(cmd, expect) + assert "\"TotalCount\":" in shell("tccli cvm DescribeInstances") def test_describe_disaster_disaster_recover_group_quota(): - cmd = 'tccli cvm DescribeDisasterRecoverGroupQuota' - expect = "\"CvmInHostGroupQuota\":" - test_cli = TestCli() - test_cli.equal(cmd, expect) + assert "\"CvmInHostGroupQuota\":" in shell("tccli cvm DescribeDisasterRecoverGroupQuota") def test_describe_disaster_recover_groups(): - cmd = 'tccli cvm DescribeDisasterRecoverGroups' - expect = "DisasterRecoverGroupSet" - test_cli = TestCli() - test_cli.equal(cmd, expect) + assert "DisasterRecoverGroupSet" in shell("tccli cvm DescribeDisasterRecoverGroups") def test_describe_hosts(): - cmd = 'tccli cvm DescribeHosts --cli-unfold-argument' + cmd = 'tccli cvm DescribeHosts --cli-unfold-argument --version 2017-03-12' cmd += ' --Filters.0.Name zone' cmd += ' --Filters.0.Values ap-guangzhou-2' cmd += ' --Offset 0 --Limit 20' - expect = "\"HostSet\": []" - test_cli = TestCli() - test_cli.equal(cmd, expect) + + assert "\"HostSet\": []" in shell(cmd) def test_create_image_dry_run(): cmd = 'tccli cvm CreateImage --ImageName test-image --DryRun true' - expect = "\"RequestId\": " - test_cli = TestCli() - test_cli.equal(cmd, expect) + assert "\"RequestId\": " in shell(cmd) def test_create_image_dry_run_with_unfold_argument(): cmd = 'tccli cvm CreateImage --cli-unfold-argument --ImageName test-image --DryRun true' - expect = "\"RequestId\": " - test_cli = TestCli() - test_cli.equal(cmd, expect) - - -def test_multi_process(): - queue = Queue() - def run_cvm_describe_regions(idx, q): - args = ["cvm", "DescribeRegions"] - try: - CLICommand()(args) - q.put(0) - except Exception: - q.put(255) - - processes = [] - for i in range(10): - p = Process(target=run_cvm_describe_regions, args=(i, queue)) - p.start() - processes.append(p) - for p in processes: - p.join() - while not queue.empty(): - ret = queue.get() - assert ret == 0 + assert "\"RequestId\": " in shell(cmd) diff --git a/tests/test_helper.py b/tests/test_helper.py index eda2567f46..5f6727c170 100644 --- a/tests/test_helper.py +++ b/tests/test_helper.py @@ -1,30 +1,30 @@ # -*- coding:utf8 -*- -from utils import TestCli +from utils import shell def test_help(): cmd = 'tccli help' expect = 'tccli [options] [options] [options] [options and parameters]' - test_cli = TestCli() - test_cli.equal(cmd, expect) + + assert expect in shell(cmd) def test_help_detail(): cmd = 'tccli help --detail' expect = 'cvm\n 介绍如何使用API对云服务器进行操作，包括使用并管理实例、镜像、密钥等资源。' - test_cli = TestCli() - test_cli.equal(cmd, expect) + + assert expect in shell(cmd) def test_cvm_help(): - cmd = 'tccli cvm help' + cmd = 'tccli cvm help --version 2017-03-12' expect = '' - test_cli = TestCli() - test_cli.equal(cmd, expect) + + assert expect in shell(cmd) def test_cvm_help_detail(): - cmd = 'tccli cvm help --detail' + cmd = 'tccli cvm help --detail --version 2017-03-12' expect = 'AVAILABLE ACTIONS\n AllocateHosts' - test_cli = TestCli() - test_cli.equal(cmd, expect) + + assert expect in shell(cmd) diff --git a/tests/test_sso.py b/tests/test_sso.py new file mode 100644 index 0000000000..4b459dba5c --- /dev/null +++ b/tests/test_sso.py @@ -0,0 +1,12 @@ +# -*- coding:utf8 -*- + +from utils import shell, recover_profile + + +@recover_profile() +def test_sso_configure_and_logout(): + url = "https://tencentcloudsso.com/test/login" + assert ("url 已配置为 '%s', 接下来可以使用 `tccli sso login` 进行登陆\n" % url == + shell("tccli sso configure --url %s" % url)) + + assert "登出成功, 密钥凭证已被删除:" in shell("tccli sso logout") diff --git a/tests/utils.py b/tests/utils.py index 383882ead4..cfbc77a24f 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -1,7 +1,111 @@ +import contextlib +import functools import os +import subprocess +import sys + + +def shell(cmd): + if isinstance(cmd, str): + args = cmd + use_shell = True + else: + args = cmd + use_shell = False + p = subprocess.Popen(args, shell=use_shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + stdout, _ = p.communicate() + + if sys.version_info.major >= 3: + return stdout.decode("utf-8") + + return stdout + + +_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +_TCCLI_RUNNER = ( + "{py} -c \"import sys; sys.path.insert(0, '{repo}'); " + "from tccli.main import main; sys.exit(main())\"" +).format(py=sys.executable, repo=_REPO_ROOT) + + +def shell_with_stderr(cmd, clean_cred_env=False): + """执行命令，返回 (stdout, stderr, returncode)。 + 将命令中的 tccli 替换为当前源码入口，确保测试当前分支代码。 + clean_cred_env=True 时从 env 中移除 AK/SK，用于测试无凭证场景。 + """ + cmd = cmd.replace("tccli ", _TCCLI_RUNNER + " ", 1) + env = os.environ.copy() + if clean_cred_env: + env.pop("TENCENTCLOUD_SECRET_ID", None) + env.pop("TENCENTCLOUD_SECRET_KEY", None) + env.pop("TENCENTCLOUD_SECRET_TOKEN", None) + p = subprocess.Popen( + cmd, shell=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + env=env, + ) + stdout, stderr = p.communicate() + if sys.version_info.major >= 3: + return stdout.decode("utf-8"), stderr.decode("utf-8"), p.returncode + return stdout, stderr, p.returncode + + +def recover_profile(prof="default"): + def decorator(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + with recover_profile_ctx(prof): + return func(*args, **kwargs) + + return wrapper + + return decorator + + +@contextlib.contextmanager +def recover_profile_ctx(prof="default"): + cred_path = os.path.expanduser("~/.tccli/%s.credential" % prof) + conf_path = os.path.expanduser("~/.tccli/%s.configure" % prof) + + cred_backup = None + conf_backup = None + + if os.path.exists(cred_path): + with open(cred_path, "r") as cred_f: + cred_backup = cred_f.readlines() + + if os.path.exists(conf_path): + with open(conf_path, "r") as conf_f: + conf_backup = conf_f.readlines() + + yield + + if sys.version_info.major == 2: + FileNotFound = OSError + else: + FileNotFound = FileNotFoundError + + if cred_backup: + with open(cred_path, "w") as cred_f: + cred_f.writelines(cred_backup) + else: + try: + os.remove(cred_path) + except FileNotFound: + pass + + if conf_backup: + with open(conf_path, "w") as conf_f: + conf_f.writelines(conf_backup) + else: + try: + os.remove(conf_path) + except FileNotFound: + pass class TestCli(object): + """保留国内站原有的 TestCli 风格入口，供既有测试沿用。""" def run_cmd(self, cmd): result = os.popen(cmd) @@ -10,5 +114,4 @@ def run_cmd(self, cmd): def equal(self, cmd, expect): run = self.run_cmd(cmd) result = run.read() - # self.assertEqual(result, expect, "Unexpected Result") assert result.find(expect) >= 0