import json import subprocess import sys import tempfile import unittest from pathlib import Path from hydra.profiles import load_profile class CLITests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix='hc-') self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.defaults = self.root / 'defaults.toml' self.base = [sys.executable, 'bin/hydra', '--config', str(self.root / 'config'), '--state', str(self.root / 'state'), '--runtime', str(self.root / 'run'), '--defaults', str(self.defaults)] def run_cli(self, *args): return subprocess.run(self.base + list(args), capture_output=True, text=True, timeout=5) def test_defaults_affect_new_profiles_only_and_explicit_options_win(self): self.defaults.write_text('network="bridge"\nbridge="hydrabr0"\nwidth=1920\nheight=1080\nrefresh_millihz=144000\n') added = self.run_cli('add', 'first', '--width', '1600', '--network', 'user', '--refresh-millihz', '0') self.assertEqual(added.returncode, 0, added.stderr) first = load_profile(self.root / 'config/first.toml') self.assertEqual((first.network, first.width, first.height, first.refresh_millihz), ('user', 1600, 1080, 0)) added = self.run_cli('add', 'second') self.assertEqual(added.returncode, 0, added.stderr) second = load_profile(self.root / 'config/second.toml') self.assertEqual((second.network, second.width, second.height, second.refresh_millihz), ('bridge', 1920, 1080, 144000)) self.defaults.write_text('width=1024\nnetwork="user"\n') self.assertEqual(load_profile(self.root / 'config/second.toml'), second) self.defaults.write_text('width="bad"\n') listed = self.run_cli('list') self.assertEqual(listed.returncode, 0, listed.stderr) self.assertEqual(len(json.loads(listed.stdout)), 2) self.assertEqual(self.run_cli('add', 'third').returncode, 1) self.assertFalse((self.root / 'config/third.toml').exists()) def test_missing_defaults_use_developer_preferences_and_media_cli(self): added = self.run_cli('add', 'guest') self.assertEqual(added.returncode, 0, added.stderr) profile = load_profile(self.root / 'config/guest.toml') self.assertEqual((profile.network, profile.width, profile.height, profile.refresh_millihz), ('user', 1280, 800, 0)) iso = self.root / 'installer.iso' iso.touch() attached = self.run_cli('media', 'guest', '--iso', str(iso)) self.assertEqual(attached.returncode, 0, attached.stderr) iso.unlink() listed = self.run_cli('list') self.assertEqual(listed.returncode, 0, listed.stderr) self.assertEqual(self.run_cli('status', 'guest').returncode, 0) detached = self.run_cli('media', 'guest', '--detach') self.assertEqual(detached.returncode, 0, detached.stderr) self.assertIsNone(load_profile(self.root / 'config/guest.toml').iso) def test_unknown_or_invalid_distribution_defaults_are_rejected(self): for content in ('iso="/private.iso"\n', 'network="public"\n', 'refresh_millihz=true\n', '[display]\nwidth=1920\n'): self.defaults.write_text(content) result = self.run_cli('add', 'guest') self.assertEqual(result.returncode, 1, result.stderr) self.assertNotIn('Traceback', result.stderr) self.assertFalse((self.root / 'config/guest.toml').exists()) def test_plain_noninteractive_invocation_prints_help_without_creating_state(self): result = self.run_cli() self.assertEqual(result.returncode, 0, result.stderr) self.assertIn('Hydra OS', result.stdout) self.assertIn('run', result.stdout) self.assertFalse((self.root / 'state').exists()) def test_run_and_open_are_headless_and_existing_name_is_never_replaced(self): import os from unittest.mock import patch from test_media import iso_bytes from test_runtime import FAKE_IMG, FAKE_QEMU from hydra.runtime import Runtime iso = self.root / 'installer.iso' iso.write_bytes(iso_bytes()) qemu = self.root / 'qemu' qemu.write_text(FAKE_QEMU) qemu.chmod(0o700) img = self.root / 'img' img.write_text(FAKE_IMG) img.chmod(0o700) runtime = Runtime(self.root / 'config', self.root / 'state', self.root / 'run') def cleanup(): if runtime.status('guest')['status'] == 'running': runtime.stop('guest', force=True) self.addCleanup(cleanup) with patch.dict(os.environ, {'HYDRA_QEMU': str(qemu), 'HYDRA_QEMU_IMG': str(img), 'DISPLAY': '', 'WAYLAND_DISPLAY': '', 'HYDRA_SESSION_MANIFEST': '/not/a/manifest'}): started = self.run_cli('run', str(iso), '--name', 'guest', '--memory', '256', '--cpus', '1', '--disk-size', '8M', '--no-view') self.assertEqual(started.returncode, 0, started.stderr) original = (self.root / 'config/guest.toml').read_bytes() pid = runtime.paths('guest').pid.read_bytes() duplicate = self.run_cli('run', str(iso), '--name', 'guest', '--no-view') self.assertEqual(duplicate.returncode, 1, duplicate.stderr) self.assertIn('hydra open guest', duplicate.stderr) iso.unlink() opened = self.run_cli('open', 'guest', '--no-view') self.assertEqual(opened.returncode, 0, opened.stderr) self.assertEqual((self.root / 'config/guest.toml').read_bytes(), original) self.assertEqual(runtime.paths('guest').pid.read_bytes(), pid) def test_download_rejects_credential_urls_without_echoing_secrets(self): result = self.run_cli('download', 'https://user:supersecret@example.org/installer.iso?token=private') self.assertEqual(result.returncode, 1, result.stderr) self.assertNotIn('supersecret', result.stderr) self.assertNotIn('private', result.stderr) self.assertNotIn('Traceback', result.stderr) def guided_defaults_profile(self, command): import contextlib import io import os from unittest.mock import patch from hydra.cli import main from hydra.runtime import Runtime from test_media import iso_bytes from test_runtime import FAKE_IMG, FAKE_QEMU self.defaults.write_text('network="bridge"\nbridge="testbr0"\nwidth=1600\nheight=900\nrefresh_millihz=75000\n') iso = self.root / 'guided.iso' iso.write_bytes(iso_bytes()) qemu = self.root / 'fake-qemu' qemu.write_text(FAKE_QEMU); qemu.chmod(0o700) image = self.root / 'fake-img' image.write_text(FAKE_IMG); image.chmod(0o700) viewer = self.root / 'fake-viewer' viewer.write_text('#!/bin/sh\necho --hide-toolbar\n'); viewer.chmod(0o700) runtime = Runtime(self.root / 'config', self.root / 'state', self.root / 'run') def cleanup(): if runtime.status('guided')['status'] == 'running': runtime.stop('guided', force=True) self.addCleanup(cleanup) inputs = [':run ' + str(iso), 'guided', '512', '1', '8M', '', ':quit'] with patch.dict(os.environ, {'HYDRA_QEMU': str(qemu), 'HYDRA_QEMU_IMG': str(image), 'HYDRA_VIEWER': str(viewer), 'WAYLAND_DISPLAY': 'unused', 'HYDRA_SESSION_MANIFEST': ''}), \ patch('sys.stdin.isatty', return_value=True), patch('builtins.input', side_effect=inputs), \ patch('hydra.launch.available_memory_mib', return_value=8192), \ patch('hydra.session.view', return_value=0), contextlib.redirect_stdout(io.StringIO()): self.assertEqual(main(self.base[2:] + command), 0) self.assertEqual(runtime.status('guided')['status'], 'running') return runtime.profile('guided') def test_menu_guided_run_honors_cli_custom_network_and_display_defaults(self): profile = self.guided_defaults_profile(['menu']) self.assertEqual((profile.network, profile.bridge, profile.width, profile.height, profile.refresh_millihz), ('bridge', 'testbr0', 1600, 900, 75000)) def test_plain_interactive_guided_run_honors_cli_custom_network_and_display_defaults(self): profile = self.guided_defaults_profile([]) self.assertEqual((profile.network, profile.bridge, profile.width, profile.height, profile.refresh_millihz), ('bridge', 'testbr0', 1600, 900, 75000))