75 lines
4.8 KiB
Python
75 lines
4.8 KiB
Python
"""Explicit service cancellation is scoped before mutation; never launches native."""
|
|
import argparse
|
|
import ast
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
RUNNER = ROOT/'tools/headless_workbench.py'
|
|
SOURCE = RUNNER.read_text()
|
|
TREE = ast.parse(SOURCE)
|
|
|
|
class ServiceCancelRunner(unittest.TestCase):
|
|
def refused(self, smoke, project='chat', rounds='1', timeout='420', extra=(), message='--route-service-cancel requires --smoke aircraft-route'):
|
|
with tempfile.TemporaryDirectory(prefix='erri-service-cancel-parse-') as directory:
|
|
temporary = Path(directory)
|
|
result = subprocess.run([sys.executable, str(RUNNER), '--compat', str(temporary/'absent-prefix'), '--profile', str(temporary/'absent-profile'), '--smoke', smoke, '--project', project, '--rounds', rounds, '--timeout', timeout, '--route-service-cancel', *extra], capture_output=True, text=True, timeout=10)
|
|
self.assertEqual(result.returncode, 2, result.stderr)
|
|
self.assertIn(message, result.stderr)
|
|
self.assertEqual(list(temporary.iterdir()), [], 'Invalid opt-in cannot create a prefix lock or profile')
|
|
|
|
def test_unrelated_outcomes_refused_before_prefix_or_audio(self):
|
|
for smoke in ('victory','aircraft-control','aircraft-cancel','aircraft-translate','powered-flight'):
|
|
with self.subTest(smoke=smoke): self.refused(smoke)
|
|
guard = SOURCE.index("if a.route_service_cancel and a.smoke != 'aircraft-route'")
|
|
self.assertLess(guard, SOURCE.index('prefix_lock ='))
|
|
self.assertLess(guard, SOURCE.index('profile.mkdir('))
|
|
self.assertLess(guard, SOURCE.index("silent_sink ="))
|
|
|
|
def test_original_route_requirements_unchanged(self):
|
|
for project, rounds, timeout in [('core','1','420'),('chat','2','420'),('chat','1','419')]:
|
|
with self.subTest(project=project,rounds=rounds,timeout=timeout):
|
|
self.refused('aircraft-route', project, rounds, timeout, message='requires Chat, one round, and at least 420 seconds')
|
|
|
|
def test_valid_opt_in_still_requires_disposable_prefix(self):
|
|
self.refused('aircraft-route', message='must be a disposable cloned prefix')
|
|
|
|
def test_no_second_flight_input_mode(self):
|
|
for extra, message in [(['--flight-action-inputs'],'flight-action-inputs requires the Core powered-flight fixture'), (['--flight-autohover'],'flight-autohover requires Core powered-flight with flight-action-inputs')]:
|
|
with self.subTest(extra=extra): self.refused('aircraft-route', extra=extra, message=message)
|
|
|
|
def test_actual_argparse_default_and_exact_command(self):
|
|
# Execute only actual argparse declarations and command assembly AST nodes.
|
|
# No prefix, process, environment, profile or audio code is evaluated.
|
|
parser_nodes = []
|
|
for node in TREE.body:
|
|
if isinstance(node, ast.Assign) and any(isinstance(t,ast.Name) and t.id=='p' for t in node.targets): parser_nodes.append(node)
|
|
if isinstance(node, ast.Expr) and isinstance(node.value,ast.Call) and isinstance(node.value.func,ast.Attribute) and isinstance(node.value.func.value,ast.Name) and node.value.func.value.id=='p' and node.value.func.attr=='add_argument': parser_nodes.append(node)
|
|
scope={'argparse':argparse,'Path':Path,'__doc__':'test'}
|
|
exec(compile(ast.Module(body=parser_nodes,type_ignores=[]),str(RUNNER),'exec'),scope)
|
|
for requested in (False,True):
|
|
args=scope['p'].parse_args(['--compat','/tmp/absent','--profile','/tmp/absent-profile','--smoke','aircraft-route',*(['--route-service-cancel'] if requested else [])])
|
|
self.assertIs(args.route_service_cancel,requested)
|
|
nodes=[]
|
|
for node in TREE.body:
|
|
if isinstance(node,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='command' for t in node.targets):nodes.append(node)
|
|
if isinstance(node,ast.If) and isinstance(node.test,ast.Attribute) and isinstance(node.test.value,ast.Name) and node.test.value.id=='a' and node.test.attr in {'route_service_cancel','station_teardown_diagnostics','flight_input_diagnostics','flight_action_inputs','flight_autohover'}:nodes.append(node)
|
|
context={'a':args,'ROOT':ROOT,'proton':Path('/tmp/nonexistent-proton'),'wb':Path('/tmp/nonexistent-workbench'),'profile_arg':'Z:/tmp/absent-profile'}
|
|
exec(compile(ast.Module(body=nodes,type_ignores=[]),str(RUNNER),'exec'),context)
|
|
command=context['command']
|
|
self.assertEqual(command.count('-errSmokeRouteServiceCancel'),int(requested))
|
|
if requested:
|
|
self.assertEqual(command[-2:],['-errSmokeRouteServiceCancel','1'])
|
|
self.assertEqual(command[:-2],normal)
|
|
else:normal=command
|
|
self.assertEqual(command[command.index('-errSmokeOutcome')+1],'aircraft-route')
|
|
self.assertIn('--backend',command)
|
|
self.assertEqual(command[command.index('--backend')+1],'headless')
|
|
self.assertNotIn('-errFlightActionInputs',command)
|
|
self.assertNotIn('-errFlightAutohover',command)
|
|
|
|
if __name__=='__main__':unittest.main()
|