test(cli): add compatibility job with semantic assertions

This commit is contained in:
huanghuoguoguo
2026-10-30 09:28:40 +08:00
parent 7119cc94ba
commit c05a86054b
8 changed files with 240 additions and 3 deletions
+20
View File
@@ -618,6 +618,26 @@ or latency thresholds. `gate/suite/cases.json` retains per-request evidence and
default daily or PR smoke cases. It does not test cross-Workspace isolation or
high concurrency.
## 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
+6 -1
View File
@@ -30,6 +30,7 @@ SUITE_BUDGETS = {
'state': 300,
'isolation': 300,
'plugin-lifecycle': 600,
'cli': 180,
}
sys.path.insert(0, str(ROOT))
from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories # noqa: E402
@@ -62,6 +63,10 @@ def suite_profile(name, run_id, groups=(), api_seed=1):
profile.update(resources=[], cases=[], coverage_suite='plugin-lifecycle')
add_cases(profile)
if name == 'cli':
from regression_cli_cases import configure
configure(profile)
for resource in profile['resources']:
if resource['kind'] == 'pipeline':
resource['body'].setdefault(
@@ -205,7 +210,7 @@ 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 in ('api', 'isolation', 'plugin-lifecycle'):
if args.suite in ('api', 'isolation', 'plugin-lifecycle', 'cli'):
groups = [args.suite]
if 'plugins' in groups and not {'misc/EssentialCommands', 'misc/FAQManager'} <= set(args.extra_plugins):
groups.remove('plugins')
+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
@@ -474,6 +474,9 @@
},
{"id": "isolation.same_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent sessions in one Pipeline retain their own request identity and prior history", "bindings": [{"profile": "isolation", "case": "isolation.same_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]},
{"id": "isolation.cross_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent Pipelines retain their own model routing, Bot identity, and conversation history", "bindings": [{"profile": "isolation", "case": "isolation.cross_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]},
{"id": "isolation.cancel_neighbor", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Cancelling an active tool run leaves a concurrent neighbor completed and no delayed tool side effect", "bindings": [{"profile": "isolation", "case": "isolation.cancel_neighbor", "checks": ["cancel_isolation", "neighbor_active_at_cancel", "cancellation_overlapped_neighbor", "neighbor_identity_once", "no_late_side_effect", "box_idle"]}]}
{"id": "isolation.cancel_neighbor", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Cancelling an active tool run leaves a concurrent neighbor completed and no delayed tool side effect", "bindings": [{"profile": "isolation", "case": "isolation.cancel_neighbor", "checks": ["cancel_isolation", "neighbor_active_at_cancel", "cancellation_overlapped_neighbor", "neighbor_identity_once", "no_late_side_effect", "box_idle"]}]},
{"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
@@ -17,6 +17,7 @@ GROUPS = {
'api': 'regression_api_cases',
'isolation': 'regression_isolation_cases',
'plugin-lifecycle': 'regression_plugin_lifecycle',
'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()
+1 -1
View File
@@ -289,7 +289,7 @@ class Routes:
name = binding['profile']
if name not in profiles:
path = Path(__file__).with_name('regression_profiles') / f'{name}.json'
if name in ('api', 'state', 'isolation', 'plugin-lifecycle'):
if name in ('api', 'state', 'isolation', 'plugin-lifecycle', 'cli'):
from regression_ci import suite_profile
profile, _ = suite_profile(name, 'catalog-test')