test(cli): check compatibility without freezing presentation

This commit is contained in:
huanghuoguoguo
2026-10-29 11:58:52 +08:00
parent f5924164d2
commit a5cf5f4f4d
8 changed files with 242 additions and 6 deletions
+20
View File
@@ -538,6 +538,26 @@ An HTTP route hit includes rejected requests and is not proof of all behavior
behind that route. Invalid MCP modes must return HTTP 400 and leave the original
configuration unchanged. Any failed contract or cleanup makes the job fail.
## CLI compatibility check
Run on the same disposable Linux host:
```sh
uv run --no-sync python tools/regression_ci.py --suite cli \
--lbctl /path/to/lbctl --plugins plugins --output .qa/cli-001
```
The finite job checks
Workspace identity, capability discovery, Pipeline create/read/update/delete,
and invalid credentials. It uses the supplied CLI binary and records its version;
select the latest stable release when validating the evolving CLI.
Assertions allow additional output fields and changes to human-readable wording.
Workspace identity, created resource identity, actual backend state, cleanup and
nonzero exit on authentication failure remain required. HTTP readback provides
an independent oracle, so an exit code of zero alone cannot pass a mutation.
The suite is opt-in and adds no daily schedule or PR gate.
## Sustained regression and baseline comparison
[`regression_profiles/endurance.json`](regression_profiles/endurance.json) runs
+7 -3
View File
@@ -21,7 +21,7 @@ from regression_suite import Suite
from regression_traffic import owned_box_containers, read_endpoint_snapshot
ROOT = Path(__file__).resolve().parents[1]
SUITE_BUDGETS = {'smoke': 60, 'scenarios': 300, 'daily': 1860, 'endurance': 900, 'api': 600}
SUITE_BUDGETS = {'smoke': 60, 'scenarios': 300, 'daily': 1860, 'endurance': 900, 'api': 600, 'cli': 180}
sys.path.insert(0, str(ROOT))
from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories # noqa: E402
@@ -34,6 +34,10 @@ def suite_profile(name, run_id, groups=(), api_seed=1):
profile = expanded_profile(groups)
else:
profile = json.loads(profile_path.read_text())
if name == 'cli':
from regression_cli_cases import configure
configure(profile)
if name == 'api':
from regression_api_cases import add_cases
@@ -179,8 +183,8 @@ def run(args):
groups = args.scenario_groups
if groups is None:
groups = list(DEFAULT_GROUPS) if args.suite in ('scenarios', 'daily') else []
if args.suite == 'api':
groups = ['api']
if args.suite in ('api', 'cli'):
groups = [args.suite]
if 'plugins' in groups and not {'misc/EssentialCommands', 'misc/FAQManager'} <= set(args.extra_plugins):
groups.remove('plugins')
profile, profile_path = suite_profile(args.suite, secrets.token_hex(8), groups, args.api_seed)
+115
View File
@@ -0,0 +1,115 @@
"""Published CLI compatibility checks with strict backend state and flexible presentation."""
import subprocess
from regression_api_checks import check, request, tracked_resource, unique_name
from regression_coverage import checked
from regression_local_cases import require
from regression_resource_cases import journal
CASE_NAMES = ('cli.identity', 'cli.pipeline_roundtrip', 'cli.invalid_credential')
def configure(profile):
profile.update(resources=[], cases=[], coverage_suite='cli', scenario_groups=['cli'])
profile['cases'] = [
{'name': name, 'scenario': name, 'stage': 'cli', 'expect': 'CLI_COMPAT_OK'} for name in CASE_NAMES
]
def identity(suite, row):
row['cli_version'] = suite.call(['version'])
result = suite.call(['whoami'])
check(
row,
'workspace_identity',
result.get('workspace_uuid') == suite.variables['workspace_uuid'],
'CLI resolved a different Workspace',
)
capabilities = suite.call(['capabilities'])
check(row, 'capability_discovery', bool(capabilities), 'CLI capability discovery returned no data')
def pipeline_roundtrip(suite, row):
name, collection = unique_name(), '/api/v1/pipelines'
with tracked_resource(suite, row, collection, name, item_key='pipeline', list_key='pipelines') as owned:
created = suite.call(['pipeline', 'apply'], {'name': name, 'description': 'CLI fixture 中文'})
matches = [p for p in request(suite, row, 'GET', collection)['pipelines'] if p['name'] == name]
require(len(matches) == 1, 'CLI create did not persist exactly one owned Pipeline')
uid = matches[0]['uuid']
owned.update(uuid=uid, state='created')
journal(suite, row)
path = collection + '/' + uid
check(
row,
'create_identity',
created.get('pipeline', created).get('uuid') == uid,
'CLI create omitted the created resource identity',
)
current = suite.call(['pipeline', 'get', uid])
check(
row,
'cli_get',
current.get('pipeline', current).get('uuid') == uid,
'CLI get omitted the owned entity identity',
)
listed = suite.call(['pipeline', 'list'])
items = listed.get('pipelines', []) if isinstance(listed, dict) else listed
check(
row,
'cli_list',
isinstance(items, list) and any(item.get('uuid') == uid for item in items),
'CLI list omitted the owned entity identity',
)
before = request(suite, row, 'GET', path)['pipeline']
check(row, 'create_state', before['description'] == 'CLI fixture 中文', 'CLI create changed submitted content')
suite.call(
['pipeline', 'apply'],
{
'uuid': uid,
'name': name + '-updated',
'description': 'CLI updated 🌍',
'config': before['config'],
},
)
after = request(suite, row, 'GET', path)['pipeline']
check(
row,
'update_state',
after['name'] == name + '-updated'
and after['description'] == 'CLI updated 🌍'
and after['config'] == before['config'],
'CLI update lost submitted or existing configuration',
)
suite.call(['pipeline', 'delete', uid, '--yes'])
request(suite, row, 'GET', path, expected=404)
def invalid_credential(suite, row):
source = suite.connection.get('credential_source', '')
require(source.startswith('env:'), 'CLI credential probe requires the disposable environment credential')
result = subprocess.run(
[*suite.cli, 'pipeline', 'list'],
capture_output=True,
text=True,
timeout=suite.timeout(30),
env={**suite.env, source.removeprefix('env:'): 'qa-invalid-credential'},
)
row['rejected_exit_code'] = result.returncode
# Exit code is stable; error wording and JSON error details may evolve.
check(row, 'credential_rejected', result.returncode != 0, 'CLI reported success with an invalid credential')
check(row, 'error_signal', bool(result.stdout.strip() or result.stderr.strip()), 'CLI failure had no diagnostic')
suite.call(['pipeline', 'list'])
def run_case(suite, case, row):
checked(
row,
'cli_chain',
lambda: {
'cli.identity': identity,
'cli.pipeline_roundtrip': pipeline_roundtrip,
'cli.invalid_credential': invalid_credential,
}[case['scenario']](suite, row),
)
return {'result': case['expect']}
+4 -1
View File
@@ -434,6 +434,9 @@
"priority": "P1",
"description": "Invalid MCP transport is rejected without changing stored configuration",
"bindings": [{"profile": "api", "case": "fixed_api_mcp", "checks": ["api_chain", "mcp_invalid_transport_preserved"]}, {"profile": "api", "case": "shuffled_api_mcp", "checks": ["api_chain", "mcp_invalid_transport_preserved"]}]
}
},
{"id": "cli.identity", "area": "cli", "kind": "blackbox", "priority": "P2", "description": "CLI resolves the intended Workspace and discovers capabilities without freezing presentation", "bindings": [{"profile": "cli", "case": "cli.identity", "checks": ["cli_chain", "workspace_identity", "capability_discovery"]}]},
{"id": "cli.pipeline_roundtrip", "area": "cli", "kind": "blackbox", "priority": "P2", "description": "CLI mutations preserve submitted and existing Pipeline state, with independent HTTP readback", "bindings": [{"profile": "cli", "case": "cli.pipeline_roundtrip", "checks": ["cli_chain", "create_identity", "cli_get", "cli_list", "create_state", "update_state"]}]},
{"id": "cli.invalid_credential", "area": "cli", "kind": "blackbox", "priority": "P2", "description": "CLI rejects invalid credentials with a nonzero exit and a diagnostic signal", "bindings": [{"profile": "cli", "case": "cli.invalid_credential", "checks": ["cli_chain", "credential_rejected", "error_signal"]}]}
]
}
+1
View File
@@ -15,6 +15,7 @@ GROUPS = {
'lifecycle': 'regression_lifecycle_cases',
'plugins': 'regression_plugin_cases',
'api': 'regression_api_cases',
'cli': 'regression_cli_cases',
}
+5
View File
@@ -568,6 +568,11 @@ class Suite:
error = error.replace(secret, '[REDACTED]')
raise RuntimeError(error[:3000])
response = json.loads(result.stdout)
if not isinstance(response, dict) or response.get('ok') is False or response.get('error'):
error = result.stdout
for secret in self.secrets:
error = error.replace(secret, '[REDACTED]')
raise RuntimeError('CLI returned a failure response: ' + error[:3000])
if args == ['whoami']:
self.connection = response.get('meta') or {}
return response['data']
+88
View File
@@ -0,0 +1,88 @@
"""Keep CLI presentation flexible while rejecting wrong identity and false success."""
from pathlib import Path
from types import SimpleNamespace
import tempfile
import json
import unittest
from unittest.mock import Mock, patch
from regression_cli_cases import identity, invalid_credential, pipeline_roundtrip
from regression_ci import suite_profile
from regression_suite import Suite, validate
class CLICompatibilityTests(unittest.TestCase):
def test_wrong_workspace_fails_even_when_cli_exits_successfully(self):
suite = SimpleNamespace(
call=Mock(side_effect=[{'version': 'future', 'new_field': True}, {'workspace_uuid': 'wrong'}]),
variables={'workspace_uuid': 'expected'},
)
row = {}
with self.assertRaisesRegex(ValueError, 'different Workspace'):
identity(suite, row)
self.assertEqual(row['assertions']['workspace_identity'], 'failed')
def test_invalid_credential_must_have_nonzero_exit(self):
suite = SimpleNamespace(
connection={'credential_source': 'env:QA_KEY'},
cli=['lbctl', '-o', 'json'],
env={'QA_KEY': 'secret'},
timeout=lambda value: value,
)
with patch(
'regression_cli_cases.subprocess.run', return_value=SimpleNamespace(returncode=0, stdout='{}', stderr='')
):
with self.assertRaisesRegex(ValueError, 'success with an invalid credential'):
invalid_credential(suite, {})
def test_successful_cli_create_without_backend_state_fails(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory)
suite = Suite(
{'cli': {'config': 'unused'}, 'cases': [], 'vars': {'workspace_uuid': 'test'}}, path / 'p.json', path
)
suite.call = Mock(return_value={'uuid': 'reported-but-not-created'})
row = {}
with (
patch('regression_cli_cases.request', return_value={'pipelines': []}),
patch('regression_api_checks.api', return_value=(200, {'code': 0, 'data': {'pipelines': []}})),
):
with self.assertRaisesRegex(ValueError, 'exactly one owned Pipeline'):
pipeline_roundtrip(suite, row)
self.assertEqual(row['owned_resources'][0]['state'], 'absent')
def test_zero_exit_does_not_hide_an_explicit_failure_envelope(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory)
suite = Suite({'cli': {'config': 'unused'}, 'cases': []}, path / 'p.json', path, env={})
for response in (
{'ok': False, 'data': {'requested_uuid': 'owned'}},
{'error': {'type': 'failed'}, 'data': {'uuid': 'owned'}},
):
process = SimpleNamespace(returncode=0, stdout=json.dumps(response), stderr='')
with self.subTest(response=response), patch('regression_suite.subprocess.run', return_value=process):
with self.assertRaisesRegex(RuntimeError, 'failure response'):
suite.call(['pipeline', 'get', 'owned'])
process = SimpleNamespace(
returncode=0,
stdout=json.dumps(
{'ok': True, 'new_field': 'allowed', 'data': {'pipeline': {'uuid': 'owned', 'extra': 1}}}
),
stderr='',
)
with patch('regression_suite.subprocess.run', return_value=process):
self.assertEqual(suite.call(['pipeline', 'get', 'owned'])['pipeline']['uuid'], 'owned')
def test_profile_is_finite_and_uses_no_shared_resources(self):
profile, _ = suite_profile('cli', 'unit-test')
validate(profile)
self.assertEqual(profile['resources'], [])
self.assertEqual(len(profile['cases']), 3)
self.assertNotIn('execution', profile)
smoke, _ = suite_profile('smoke', 'unit-test')
self.assertFalse(any(case.get('scenario', '').startswith('cli.') for case in smoke['cases']))
if __name__ == '__main__':
unittest.main()
+2 -2
View File
@@ -289,10 +289,10 @@ class Routes:
name = binding['profile']
if name not in profiles:
path = Path(__file__).with_name('regression_profiles') / f'{name}.json'
if name == 'api':
if name in ('api', 'cli'):
from regression_ci import suite_profile
profile, _ = suite_profile('api', 'catalog-test')
profile, _ = suite_profile(name, 'catalog-test')
elif name == 'scenarios':
from regression_scenarios import expanded_profile
from regression_local_cases import DEFAULT_GROUPS