87 lines
4.4 KiB
Python
87 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Run against a built CLI: python3 Tests/cli_regression.py /absolute/path/container-compose.
|
|
Uses a fake container executable; never touches the real runtime.
|
|
"""
|
|
import json, os, pathlib, signal, subprocess, sys, tempfile, time
|
|
|
|
binary = str(pathlib.Path(sys.argv[1]).resolve())
|
|
with tempfile.TemporaryDirectory(prefix='container-compose-regression-', dir=os.environ.get('CONTAINER_COMPOSE_TEST_TMPDIR', '/Volumes/ChatGPT' if pathlib.Path('/Volumes/ChatGPT').is_dir() else tempfile.gettempdir())) as tmp:
|
|
root = pathlib.Path(tmp)
|
|
fake = root / 'container'
|
|
fake.write_text('#!' + sys.executable + '\n' + r'''
|
|
import json, os, signal, sys, time
|
|
args = sys.argv[1:]
|
|
with open(os.environ['CALLS'], 'a') as f: f.write(json.dumps(args) + '\n')
|
|
if args[0] == 'inspect': sys.exit(1)
|
|
if args[0] == 'run':
|
|
if os.environ.get('SLOW_START'): time.sleep(0.7)
|
|
print('created', flush=True)
|
|
if args[0] == 'logs':
|
|
print('LIVE ' + args[-1], flush=True)
|
|
if os.environ.get('FAIL_LOGS'): sys.exit(7)
|
|
while True: time.sleep(0.05)
|
|
''')
|
|
fake.chmod(0o755)
|
|
compose = root / 'compose.yaml'
|
|
compose.write_text('name: regression\nservices:\n db:\n image: example/db\n web:\n image: example/web\n depends_on: [db]\n')
|
|
calls = root / 'calls.jsonl'
|
|
env = dict(os.environ, PATH=str(root)+':'+os.environ.get('PATH',''), CALLS=str(calls))
|
|
def recorded(): return [json.loads(x) for x in calls.read_text().splitlines()]
|
|
def reset(): calls.write_text('')
|
|
def command(args, extra=None):
|
|
return subprocess.Popen([binary, '-f', str(compose)] + args, env=env | (extra or {}), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=True)
|
|
count = 0
|
|
# Both detach aliases, verbosity positions and all requested verbosity levels.
|
|
for detach in ['-d', '--detach']:
|
|
for level in ['', '-v', '-vv', '-vvv']:
|
|
for before in [False, True]:
|
|
reset()
|
|
args = ([level] if level and before else []) + ['up', detach] + ([level] if level and not before else [])
|
|
p = command(args); output = p.communicate(timeout=5)[0].decode()
|
|
assert p.returncode == 0, output
|
|
records = recorded()
|
|
assert len([x for x in records if x[0]=='run']) == 2, records
|
|
assert not any(x[0] in ['logs','stop'] for x in records), records
|
|
if level: assert 'mode=detached' in output and f'verbosity={len(level)-1}' in output, output
|
|
count += 1
|
|
# Attached log output arrives while CLI is still alive. Interrupt parent or process group.
|
|
for level in ['', '-v', '-vv', '-vvv']:
|
|
for before in [False, True]:
|
|
reset()
|
|
args = ([level] if level and before else []) + ['up'] + ([level] if level and not before else [])
|
|
p = command(args)
|
|
output = b''
|
|
import select
|
|
deadline = time.monotonic()+5
|
|
while b'LIVE regression-db-1' not in output or b'LIVE regression-web-1' not in output:
|
|
assert time.monotonic()<deadline, output
|
|
if select.select([p.stdout],[],[],0.1)[0]: output += os.read(p.stdout.fileno(),4096)
|
|
assert p.poll() is None, (p.returncode, output, p.stdout.read(), recorded())
|
|
if before: os.killpg(p.pid, signal.SIGINT)
|
|
else: p.send_signal(signal.SIGINT)
|
|
output += p.communicate(timeout=5)[0]
|
|
assert p.returncode == 0, output
|
|
assert [x for x in recorded() if x[0]=='stop'] == [['stop','regression-web-1'],['stop','regression-db-1']], recorded()
|
|
assert not any(x[0]=='delete' for x in recorded())
|
|
count += 1
|
|
reset()
|
|
p = command(['up'], {'SLOW_START':'1'})
|
|
deadline = time.monotonic()+5
|
|
while not any(x[0]=='run' for x in recorded()):
|
|
assert time.monotonic()<deadline
|
|
time.sleep(.02)
|
|
p.send_signal(signal.SIGINT)
|
|
output = p.communicate(timeout=5)[0]
|
|
assert p.returncode==0, output
|
|
assert [x for x in recorded() if x[0]=='stop']==[['stop','regression-db-1']], recorded()
|
|
count += 1
|
|
reset()
|
|
p=command(['up'],{'FAIL_LOGS':'1'}); output=p.communicate(timeout=5)[0]
|
|
assert p.returncode!=0 and b'Log follower failed' in output, output
|
|
count += 1
|
|
reset()
|
|
p=command(['up','--bogus']); output=p.communicate(timeout=5)[0]
|
|
assert p.returncode!=0 and recorded()==[], output
|
|
count += 1
|
|
print(f'PASS: {count} CLI regression cases')
|