feat: add independent VM manager
This commit is contained in:
Codex
@@ -0,0 +1,54 @@
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from hydra.profiles import Profile, load_profile, save_profile
|
||||
|
||||
|
||||
class ProfilesTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix='hydra-test-')
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
self.iso = self.root / 'installer with spaces,comma.iso'
|
||||
self.iso.touch()
|
||||
|
||||
def test_roundtrip_paths_and_defaults(self):
|
||||
profile = Profile('guest', self.iso)
|
||||
save_profile(self.root / 'guest.toml', profile)
|
||||
loaded = load_profile(self.root / 'guest.toml')
|
||||
self.assertEqual(loaded.iso, self.iso)
|
||||
self.assertEqual((loaded.memory, loaded.cpus, loaded.disk_size), (4096, 2, '40G'))
|
||||
self.assertEqual((loaded.graphics, loaded.network, loaded.bridge), ('auto', 'user', 'hydrabr0'))
|
||||
|
||||
def test_invalid_names_resources_and_modes(self):
|
||||
for name in ('', '../guest', 'Guest', '-guest', 'a' * 33, 'a\n'):
|
||||
with self.subTest(name=name), self.assertRaises(ValueError):
|
||||
Profile(name, self.iso)
|
||||
for field, values in {'memory': (0, -1, True, '4096'), 'cpus': (0, -1, True, 257),
|
||||
'disk_size': ('0G', '-1G', '40G,evil', '40'),
|
||||
'graphics': ('gl',), 'network': ('public',),
|
||||
'bridge': ('../bad', 'a,evil')}.items():
|
||||
for value in values:
|
||||
with self.subTest(field=field, value=value), self.assertRaises(ValueError):
|
||||
Profile('guest', self.iso, **{field: value})
|
||||
|
||||
def test_missing_iso_and_duplicate_profile_preserve_existing(self):
|
||||
with self.assertRaises(ValueError):
|
||||
Profile('guest', self.root / 'missing.iso')
|
||||
dest = self.root / 'guest.toml'
|
||||
save_profile(dest, Profile('guest', self.iso))
|
||||
original = dest.read_bytes()
|
||||
with self.assertRaises(FileExistsError):
|
||||
save_profile(dest, Profile('guest', self.iso, memory=8192))
|
||||
self.assertEqual(dest.read_bytes(), original)
|
||||
|
||||
def test_profile_name_must_match_filename(self):
|
||||
path = self.root / 'guest.toml'
|
||||
path.write_text(f'name = "other"\niso = "{self.iso}"\n')
|
||||
with self.assertRaises(ValueError):
|
||||
load_profile(path)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,141 @@
|
||||
import json
|
||||
import socket
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from hydra.qmp import QMPClient, QMPError
|
||||
|
||||
|
||||
class QMPServer:
|
||||
def __init__(self, path, handler):
|
||||
self.socket = socket.socket(socket.AF_UNIX)
|
||||
self.socket.bind(str(path))
|
||||
self.socket.listen(4)
|
||||
self.socket.settimeout(0.1)
|
||||
self.handler = handler
|
||||
self.stop = threading.Event()
|
||||
self.errors = []
|
||||
self.thread = threading.Thread(target=self.run, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def run(self):
|
||||
while not self.stop.is_set():
|
||||
try:
|
||||
conn, _ = self.socket.accept()
|
||||
except socket.timeout:
|
||||
continue
|
||||
except OSError:
|
||||
break
|
||||
try:
|
||||
with conn:
|
||||
conn.settimeout(1)
|
||||
self.handler(conn)
|
||||
except (BrokenPipeError, ConnectionResetError, socket.timeout):
|
||||
pass
|
||||
except Exception as exc:
|
||||
self.errors.append(exc)
|
||||
|
||||
def close(self):
|
||||
self.stop.set()
|
||||
self.socket.close()
|
||||
self.thread.join(2)
|
||||
|
||||
|
||||
def send(conn, message):
|
||||
conn.sendall(json.dumps(message).encode() + b'\r\n')
|
||||
|
||||
|
||||
def greeting(conn):
|
||||
send(conn, {'QMP': {'version': {'qemu': {'major': 10, 'minor': 0, 'micro': 0},
|
||||
'package': ''}, 'capabilities': []}})
|
||||
|
||||
|
||||
class QMPTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix='hq-')
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.path = Path(self.tmp.name) / 'qmp'
|
||||
|
||||
def server(self, handler):
|
||||
srv = QMPServer(self.path, handler)
|
||||
self.addCleanup(srv.close)
|
||||
return srv
|
||||
|
||||
def test_handshake_ids_events_and_return(self):
|
||||
commands = []
|
||||
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
stream = conn.makefile('rb')
|
||||
for index in range(2):
|
||||
request = json.loads(stream.readline())
|
||||
commands.append(request)
|
||||
send(conn, {'event': 'STOP', 'data': {}})
|
||||
send(conn, {'return': {'ignored': True}, 'id': 'unrelated'})
|
||||
send(conn, {'return': {} if index == 0 else {'status': 'running'}, 'id': request['id']})
|
||||
self.server(handler)
|
||||
with QMPClient(self.path, timeout=0.5) as client:
|
||||
self.assertEqual(client.command('query-status'), {'status': 'running'})
|
||||
self.assertEqual([cmd['execute'] for cmd in commands], ['qmp_capabilities', 'query-status'])
|
||||
self.assertNotEqual(commands[0]['id'], commands[1]['id'])
|
||||
|
||||
def test_command_error(self):
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
stream = conn.makefile('rb')
|
||||
for index in range(2):
|
||||
request = json.loads(stream.readline())
|
||||
if index == 0:
|
||||
send(conn, {'return': {}, 'id': request['id']})
|
||||
else:
|
||||
send(conn, {'error': {'class': 'CommandNotFound', 'desc': 'unsupported'}, 'id': request['id']})
|
||||
self.server(handler)
|
||||
with QMPClient(self.path, timeout=0.5) as client, self.assertRaises(QMPError):
|
||||
client.command('missing')
|
||||
|
||||
def test_greeting_timeout_is_bounded(self):
|
||||
self.server(lambda conn: time.sleep(0.3))
|
||||
started = time.monotonic()
|
||||
with self.assertRaises(QMPError):
|
||||
QMPClient(self.path, timeout=0.08)
|
||||
self.assertLess(time.monotonic() - started, 0.25)
|
||||
|
||||
def test_event_stream_cannot_extend_command_deadline(self):
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
request = json.loads(conn.makefile('rb').readline())
|
||||
while True:
|
||||
send(conn, {'event': 'STOP', 'data': {}})
|
||||
time.sleep(0.01)
|
||||
self.server(handler)
|
||||
started = time.monotonic()
|
||||
with self.assertRaises(QMPError):
|
||||
QMPClient(self.path, timeout=0.08)
|
||||
self.assertLess(time.monotonic() - started, 0.25)
|
||||
|
||||
def test_malformed_greeting_rejected(self):
|
||||
self.server(lambda conn: send(conn, {'return': {}}))
|
||||
with self.assertRaises(QMPError):
|
||||
QMPClient(self.path)
|
||||
|
||||
def test_command_timeout_after_successful_handshake(self):
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
with conn.makefile('rb') as stream:
|
||||
request = json.loads(stream.readline())
|
||||
send(conn, {'return': {}, 'id': request['id']})
|
||||
stream.readline()
|
||||
time.sleep(0.3)
|
||||
self.server(handler)
|
||||
with QMPClient(self.path, timeout=0.08) as client:
|
||||
started = time.monotonic()
|
||||
with self.assertRaises(QMPError):
|
||||
client.command('query-status')
|
||||
self.assertLess(time.monotonic() - started, 0.25)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,297 @@
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from hydra.profiles import Profile, save_profile
|
||||
from hydra.runtime import Runtime, RuntimeError as VMError, build_qemu_argv
|
||||
from test_qmp import QMPServer, greeting, send
|
||||
|
||||
|
||||
# External protocol fixtures isolate lifecycle tests from user VMs and real disks.
|
||||
FAKE_QEMU = r'''#!/usr/bin/env python3
|
||||
import json, os, socket, sys
|
||||
args = sys.argv[1:]
|
||||
def val(flag): return args[args.index(flag)+1]
|
||||
if os.environ.get('HYDRA_TEST_FAIL'):
|
||||
print('intentional boot failure', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
path = val('-chardev').split('path=', 1)[1].split(',server=', 1)[0].replace(',,', ',')
|
||||
pid = os.fork()
|
||||
if pid:
|
||||
sys.exit(0)
|
||||
os.setsid()
|
||||
with open(val('-pidfile'), 'w') as f: f.write(str(os.getpid()))
|
||||
s = socket.socket(socket.AF_UNIX)
|
||||
s.bind(path)
|
||||
s.listen(8)
|
||||
alive = True
|
||||
while alive:
|
||||
c, _ = s.accept()
|
||||
try:
|
||||
with c:
|
||||
c.sendall(b'{"QMP":{"version":{"qemu":{"major":10,"minor":0,"micro":0},"package":""},"capabilities":[]}}\n')
|
||||
with c.makefile('rb') as stream:
|
||||
for line in stream:
|
||||
cmd = json.loads(line)
|
||||
name = cmd['execute']
|
||||
result = {'query-name': {'name': val('-name')}, 'query-uuid': {'UUID': val('-uuid')},
|
||||
'query-status': {'status': 'running', 'running': True, 'singlestep': False}}.get(name, {})
|
||||
c.sendall(json.dumps({'return': result, 'id': cmd['id']}).encode()+b'\n')
|
||||
if name == 'quit':
|
||||
alive = False
|
||||
break
|
||||
except (BrokenPipeError, ConnectionResetError):
|
||||
pass
|
||||
s.close()
|
||||
os.unlink(path)
|
||||
'''
|
||||
|
||||
FAKE_IMG = r'''#!/usr/bin/env python3
|
||||
import sys
|
||||
with open(sys.argv[-2], 'xb') as f: f.write(b'new qcow2 fixture')
|
||||
'''
|
||||
|
||||
|
||||
class RuntimeTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix='hr-')
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
self.runtime = Runtime(self.root / 'config space,comma', self.root / 'state space,comma',
|
||||
self.root / 'run,comma', boot_timeout=1)
|
||||
self.iso = self.root / 'installer space,comma.iso'
|
||||
self.iso.touch()
|
||||
self.profile = Profile('guest', self.iso, graphics='software', disk_size='8M')
|
||||
self.runtime.add(self.profile)
|
||||
self.qemu = self.root / 'fake qemu'
|
||||
self.qemu.write_text(FAKE_QEMU)
|
||||
self.qemu.chmod(0o700)
|
||||
self.img = self.root / 'fake img'
|
||||
self.img.write_text(FAKE_IMG)
|
||||
self.img.chmod(0o700)
|
||||
env = patch.dict(os.environ, {'HYDRA_QEMU': str(self.qemu), 'HYDRA_QEMU_IMG': str(self.img)})
|
||||
env.start()
|
||||
self.addCleanup(env.stop)
|
||||
self.addCleanup(self.cleanup_guest)
|
||||
|
||||
def cleanup_guest(self):
|
||||
try:
|
||||
if self.runtime.status('guest')['status'] == 'running':
|
||||
self.runtime.stop('guest', force=True)
|
||||
except VMError:
|
||||
pass
|
||||
|
||||
def test_argv_is_private_and_comma_safe(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
argv = build_qemu_argv(self.profile, paths)
|
||||
self.assertEqual(argv[0], str(self.qemu))
|
||||
blockdevs = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
|
||||
self.assertIn(str(paths.disk), [b.get('filename') for b in blockdevs])
|
||||
self.assertIn(str(self.iso), [b.get('filename') for b in blockdevs])
|
||||
spice = argv[argv.index('-spice')+1]
|
||||
self.assertIn('unix=on', spice)
|
||||
self.assertIn('disable-copy-paste=on', spice)
|
||||
self.assertIn('disable-agent-file-xfer=on', spice)
|
||||
self.assertIn('run,,comma', spice)
|
||||
self.assertIn('virtio-blk-pci,drive=disk', argv)
|
||||
self.assertIn('virtio-net-pci,netdev=net', argv)
|
||||
self.assertNotIn('-vnc', argv)
|
||||
self.assertNotIn('-virtfs', argv)
|
||||
|
||||
def test_argv_provides_audio_and_manual_usb_channels(self):
|
||||
argv = build_qemu_argv(self.profile, self.runtime.paths('guest'))
|
||||
self.assertIn('spice,id=audio', argv)
|
||||
self.assertIn('hda-duplex,audiodev=audio', argv)
|
||||
for channel in range(3):
|
||||
self.assertIn(f'spicevmc,id=usbredir{channel},name=usbredir', argv)
|
||||
self.assertIn(f'usb-redir,chardev=usbredir{channel}', argv)
|
||||
self.assertFalse(any('usb-host' in argument for argument in argv))
|
||||
|
||||
def test_metadata_symlink_does_not_overwrite_user_data(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
valuable = self.root / 'valuable data'
|
||||
valuable.write_text('preserve me')
|
||||
paths.metadata.symlink_to(valuable)
|
||||
self.runtime.start('guest')
|
||||
self.assertEqual(valuable.read_text(), 'preserve me')
|
||||
self.assertFalse(paths.metadata.is_symlink())
|
||||
|
||||
def test_disk_symlink_never_changes_its_target(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
valuable = self.root / 'valuable disk'
|
||||
valuable.write_text('preserve me')
|
||||
paths.disk.symlink_to(valuable)
|
||||
with self.assertRaises(VMError):
|
||||
self.runtime.start('guest')
|
||||
self.assertEqual(valuable.read_text(), 'preserve me')
|
||||
|
||||
def test_malformed_identity_record_is_safe_unknown(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
paths.metadata.write_text('[]')
|
||||
server = QMPServer(paths.qmp, greeting)
|
||||
self.addCleanup(server.close)
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
|
||||
|
||||
def test_foreign_qmp_name_blocks_control_even_when_uuid_matches(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
||||
commands = []
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
for line in conn.makefile('rb'):
|
||||
req = json.loads(line)
|
||||
commands.append(req['execute'])
|
||||
result = {'query-name': {'name': 'other'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
|
||||
send(conn, {'return': result, 'id': req['id']})
|
||||
server = QMPServer(paths.qmp, handler)
|
||||
self.addCleanup(server.close)
|
||||
with self.assertRaises(VMError):
|
||||
self.runtime.stop('guest', force=True)
|
||||
self.assertNotIn('quit', commands)
|
||||
|
||||
def test_reused_pid_without_qmp_never_controls_a_process(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
paths.pid.write_text(str(os.getpid()))
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
||||
with patch('os.kill', side_effect=AssertionError('must not signal any PID')):
|
||||
with self.assertRaises(VMError):
|
||||
self.runtime.stop('guest', force=True)
|
||||
|
||||
def test_malformed_query_result_is_safe_unknown(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
for line in conn.makefile('rb'):
|
||||
req = json.loads(line)
|
||||
send(conn, {'return': [], 'id': req['id']})
|
||||
server = QMPServer(paths.qmp, handler)
|
||||
self.addCleanup(server.close)
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
|
||||
|
||||
def test_cli_qmp_failure_reports_error_without_traceback(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
for line in conn.makefile('rb'):
|
||||
req = json.loads(line)
|
||||
result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
|
||||
if req['execute'] == 'system_powerdown':
|
||||
send(conn, {'error': {'class': 'GenericError', 'desc': 'cannot power down'}, 'id': req['id']})
|
||||
else:
|
||||
send(conn, {'return': result, 'id': req['id']})
|
||||
server = QMPServer(paths.qmp, handler)
|
||||
self.addCleanup(server.close)
|
||||
run = subprocess.run([sys.executable, 'bin/hydra', '--config', str(self.runtime.config),
|
||||
'--state', str(self.runtime.state), '--runtime', str(self.runtime.runtime),
|
||||
'shutdown', 'guest'], capture_output=True, text=True, timeout=4)
|
||||
self.assertEqual(run.returncode, 1)
|
||||
self.assertIn('cannot power down', run.stderr)
|
||||
self.assertNotIn('Traceback', run.stderr)
|
||||
|
||||
def test_disk_and_profile_survive_shutdown_and_force(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
paths.disk.parent.mkdir(parents=True, exist_ok=True)
|
||||
paths.disk.write_bytes(b'existing valuable disk')
|
||||
self.runtime.start('guest')
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'running')
|
||||
self.runtime.shutdown('guest')
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'running')
|
||||
with self.assertRaises(VMError):
|
||||
self.runtime.stop('guest', force=False)
|
||||
self.runtime.stop('guest', force=True)
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
||||
self.assertEqual(paths.disk.read_bytes(), b'existing valuable disk')
|
||||
|
||||
def test_simultaneous_start_only_launches_one_vm(self):
|
||||
barrier = threading.Barrier(2)
|
||||
results = []
|
||||
def start():
|
||||
barrier.wait()
|
||||
try:
|
||||
results.append(self.runtime.start('guest')['status'])
|
||||
except VMError as exc:
|
||||
results.append(str(exc))
|
||||
threads = [threading.Thread(target=start) for _ in range(2)]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
for thread in threads:
|
||||
thread.join(4)
|
||||
self.assertFalse(thread.is_alive())
|
||||
self.assertEqual(results.count('running'), 1)
|
||||
self.assertEqual(len(results), 2)
|
||||
self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
|
||||
|
||||
def test_stale_socket_is_replaced_only_when_no_listener(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
stale = socket.socket(socket.AF_UNIX)
|
||||
stale.bind(str(paths.qmp))
|
||||
stale.close()
|
||||
self.runtime.start('guest')
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'running')
|
||||
|
||||
def test_failed_boot_is_reported_and_disk_preserved(self):
|
||||
with patch.dict(os.environ, {'HYDRA_TEST_FAIL': '1'}), self.assertRaises(VMError):
|
||||
self.runtime.start('guest')
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
||||
self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
|
||||
self.assertIn('intentional boot failure', self.runtime.paths('guest').log.read_text())
|
||||
|
||||
def test_foreign_qmp_uuid_blocks_all_controls_and_restart(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.prepare(paths)
|
||||
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
||||
paths.pid.write_text(str(os.getpid()))
|
||||
commands = []
|
||||
def handler(conn):
|
||||
greeting(conn)
|
||||
for line in conn.makefile('rb'):
|
||||
req = json.loads(line)
|
||||
commands.append(req['execute'])
|
||||
result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': 'foreign-uuid'}}.get(req['execute'], {})
|
||||
send(conn, {'return': result, 'id': req['id']})
|
||||
server = QMPServer(paths.qmp, handler)
|
||||
self.addCleanup(server.close)
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
|
||||
for control in (lambda: self.runtime.shutdown('guest'), lambda: self.runtime.stop('guest', force=True),
|
||||
lambda: self.runtime.start('guest')):
|
||||
with self.assertRaises(VMError):
|
||||
control()
|
||||
self.assertNotIn('quit', commands)
|
||||
self.assertNotIn('system_powerdown', commands)
|
||||
self.assertFalse(paths.disk.exists())
|
||||
|
||||
def test_private_runtime_and_socket_length(self):
|
||||
self.runtime.prepare(self.runtime.paths('guest'))
|
||||
self.assertEqual(self.runtime.runtime.stat().st_mode & 0o777, 0o700)
|
||||
with self.assertRaises(VMError):
|
||||
Runtime(self.root / 'c', self.root / 's', self.root / ('r' * 100)).paths('guest')
|
||||
|
||||
def test_cli_lifecycle_and_global_paths(self):
|
||||
base = [sys.executable, 'bin/hydra', '--config', str(self.runtime.config), '--state', str(self.runtime.state),
|
||||
'--runtime', str(self.runtime.runtime)]
|
||||
for args, expected in [(['list'], 'guest'), (['start', 'guest'], 'running'),
|
||||
(['status', 'guest'], 'running'), (['stop', 'guest', '--force'], 'stopped')]:
|
||||
run = subprocess.run(base + args, text=True, capture_output=True, timeout=4)
|
||||
self.assertEqual(run.returncode, 0, run.stderr)
|
||||
self.assertIn(expected, run.stdout)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user