本文整理汇总了Python中virttest.utils_misc.find_free_port函数的典型用法代码示例。如果您正苦于以下问题：Python find_free_port函数的具体用法？Python find_free_port怎么用？Python find_free_port使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了find_free_port函数的13个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setup_unattended_http_server
def setup_unattended_http_server(self):
'''
Setup a builtin http server for serving the kickstart file
Does nothing if unattended file is not a kickstart file
'''
if self.unattended_file.endswith('.ks'):
# Red Hat kickstart install
dest_fname = 'ks.cfg'
answer_path = os.path.join(self.tmpdir, dest_fname)
self.answer_kickstart(answer_path)
if self.unattended_server_port is None:
self.unattended_server_port = utils_misc.find_free_port(
8000,
8099,
self.url_auto_content_ip)
start_unattended_server_thread(self.unattended_server_port,
self.tmpdir)
# Point installation to this kickstart url
ks_param = 'ks=http://%s:%s/%s' % (self.url_auto_content_ip,
self.unattended_server_port,
dest_fname)
if 'ks=' in self.kernel_params:
kernel_params = re.sub('ks\=[\w\d\:\.\/]+',
ks_param,
self.kernel_params)
else:
kernel_params = '%s %s' % (self.kernel_params, ks_param)
# reflect change on params
self.kernel_params = kernel_params
开发者ID:aginies，项目名称:virt-test，代码行数:35，代码来源:unattended_install.py
示例2: nc_transfer
def nc_transfer(src, dst):
nc_port = utils_misc.find_free_port(1025, 5334, vm_ip[dst])
listen_cmd = params.get("listen_cmd")
send_cmd = params.get("send_cmd")
#listen in dst
listen_cmd = listen_cmd % (nc_port, "receive")
session[dst].sendline(listen_cmd)
time.sleep(2)
#send file from src to dst
send_cmd = send_cmd % (vlan_ip[dst], str(nc_port), "file")
session[src].cmd(send_cmd, timeout=60)
try:
session[dst].read_up_to_prompt(timeout=60)
except aexpect.ExpectError:
raise error.TestFail ("Fail to receive file"
" from vm%s to vm%s" % (src+1, dst+1))
#check MD5 message digest of receive file in dst
output = session[dst].cmd_output("md5sum receive").strip()
digest_receive = re.findall(r'(\w+)', output)[0]
if digest_receive == digest_origin[src]:
logging.info("file succeed received in vm %s", vlan_ip[dst])
else:
logging.info("digest_origin is %s", digest_origin[src])
logging.info("digest_receive is %s", digest_receive)
raise error.TestFail("File transfered differ from origin")
session[dst].cmd("rm -f receive")
开发者ID:LeiCui，项目名称:virt-test，代码行数:27，代码来源:vlan.py
示例3: test
def test(self):
# This is virtual cpu flags which are supported by
# qemu but no with host cpu.
cpu_model, extra_flags = parse_cpu_model()
flags = HgFlags(cpu_model, extra_flags)
cpuf_model = cpu_model + ",enforce"
logging.debug("Unsupported flags %s.", str(flags.host_all_unsupported_flags))
# Add unsupported flags.
for fadd in flags.host_all_unsupported_flags:
cpuf_model += ",+" + str(fadd)
vnc_port = utils_misc.find_free_port(5900, 6100) - 5900
cmd = "%s -cpu %s -vnc :%d -enable-kvm" % (qemu_binary, cpuf_model, vnc_port)
out = None
try:
try:
out = utils.run(cmd, timeout=5, ignore_status=True).stderr
except error.CmdError:
logging.error("Host boot with unsupported flag")
finally:
uns_re = re.compile(r"^warning:.*flag '(.+)'", re.MULTILINE)
warn_flags = set(map(utils_misc.Flag, uns_re.findall(out)))
fwarn_flags = flags.host_all_unsupported_flags - warn_flags
if fwarn_flags:
raise error.TestFail("Qemu did not warn the use of " "flags %s" % str(fwarn_flags))
开发者ID:sibiaoluo，项目名称:virt-test，代码行数:28，代码来源:cpuflags.py
示例4: setup_unattended_http_server
def setup_unattended_http_server(self):
"""
Setup a builtin http server for serving the kickstart file
Does nothing if unattended file is not a kickstart file
"""
if self.unattended_file.endswith(".ks"):
# Red Hat kickstart install
dest_fname = "ks.cfg"
answer_path = os.path.join(self.tmpdir, dest_fname)
self.answer_kickstart(answer_path)
if self.unattended_server_port is None:
self.unattended_server_port = utils_misc.find_free_port(8000, 8099, self.url_auto_content_ip)
start_unattended_server_thread(self.unattended_server_port, self.tmpdir)
# Point installation to this kickstart url
ks_param = "ks=http://%s:%s/%s" % (self.url_auto_content_ip, self.unattended_server_port, dest_fname)
self.kernel_params = getattr(self, "kernel_params")
if "ks=" in self.kernel_params:
kernel_params = re.sub("ks\=[\w\d\:\.\/]+", ks_param, self.kernel_params)
else:
kernel_params = "%s %s" % (self.kernel_params, ks_param)
# reflect change on params
self.kernel_params = kernel_params
self.params["kernel_params"] = self.kernel_params
开发者ID:vliaskov，项目名称:virt-test，代码行数:29，代码来源:unattended_install.py
示例5: test
def test(self):
# This is virtual cpu flags which are supported by
# qemu but no with host cpu.
cpu_model, extra_flags = parse_cpu_model()
flags = HgFlags(cpu_model, extra_flags)
logging.debug("Unsupported flags %s.", str(flags.host_all_unsupported_flags))
cpuf_model = cpu_model + ",check"
# Add unsupported flags.
for fadd in flags.host_all_unsupported_flags:
cpuf_model += ",+" + str(fadd)
vnc_port = utils_misc.find_free_port(5900, 6100) - 5900
cmd = "%s -cpu %s -vnc :%d -enable-kvm" % (qemu_binary, cpuf_model, vnc_port)
out = None
try:
try:
out = utils.run(cmd, timeout=5, ignore_status=True).stderr
raise error.TestFail("Guest not boot with unsupported " "flags.")
except error.CmdError, e:
out = e.result_obj.stderr
finally:
uns_re = re.compile(r"^warning:.*flag '(.+)'", re.MULTILINE)
nf_re = re.compile(r"^CPU feature (.+) not found", re.MULTILINE)
warn_flags = set([utils_misc.Flag(x) for x in uns_re.findall(out)])
not_found = set([utils_misc.Flag(x) for x in nf_re.findall(out)])
fwarn_flags = flags.host_all_unsupported_flags - warn_flags
fwarn_flags -= not_found
if fwarn_flags:
raise error.TestFail("Qemu did not warn the use of " "flags %s" % str(fwarn_flags))
开发者ID:tjamrisk，项目名称:virt-test，代码行数:33，代码来源:cpuflags.py
示例6: migration_scenario
def migration_scenario(self):
sync = SyncData(self.master_id(), self.hostid, self.hosts,
self.id, self.sync_server)
srchost = self.params.get("hosts")[0]
dsthost = self.params.get("hosts")[1]
vms = [params.get("vms").split()[0]]
def worker(mig_data):
vm = mig_data.vms[0]
session = vm.wait_for_login(timeout=self.login_timeout)
cpuflags.install_cpuflags_util_on_vm(test, vm, install_path,
extra_flags="-msse3 -msse2")
cmd = ("%s/cpuflags-test --stressmem %d,%d" %
(os.path.join(install_path, "cpu_flags"),
vm_mem * 4, vm_mem / 2))
logging.debug("Sending command: %s" % (cmd))
session.sendline(cmd)
if self.master_id() == self.hostid:
server_port = utils_misc.find_free_port(5200, 6000)
server = listen_server(port=server_port)
data_len = 0
sync.sync(server_port, timeout=120)
client = server.socket.accept()[0]
endtime = time.time() + 30
while endtime > time.time():
data_len += len(client.recv(2048))
client.close()
server.close()
self.link_speed = data_len / (30 * 1024 * 1024)
logging.info("Link speed %d MB/s" % (self.link_speed))
ms = utils.convert_data_size(mig_speed, 'M')
if (ms > data_len / 30):
logging.warn("Migration speed %s MB/s is set faster than "
"real link speed %d MB/s" % (mig_speed,
self.link_speed))
else:
self.link_speed = ms / (1024 * 1024)
else:
data = ""
for _ in range(10000):
data += "i"
server_port = sync.sync(timeout=120)[self.master_id()]
sock = socket.socket(socket.AF_INET,
socket.SOCK_STREAM)
sock.connect((self.master_id(), server_port))
try:
endtime = time.time() + 10
while endtime > time.time():
sock.sendall(data)
sock.close()
except:
pass
self.migrate_wait(vms, srchost, dsthost, worker)
开发者ID:bssrikanth，项目名称:tp-qemu，代码行数:56，代码来源:migration_multi_host_with_speed_measurement.py
示例7: setup_cdrom
def setup_cdrom(self):
"""
Mount cdrom and copy vmlinuz and initrd.img.
"""
error.context("Copying vmlinuz and initrd.img from install cdrom %s" % self.cdrom_cd1)
if not os.path.isdir(self.image_path):
os.makedirs(self.image_path)
if self.params.get("unattended_delivery_method") in ["integrated", "url"]:
i = iso9660.Iso9660Mount(self.cdrom_cd1)
self.cdrom_cd1_mount = i.mnt_dir
else:
i = iso9660.iso9660(self.cdrom_cd1)
if i is None:
raise error.TestFail("Could not instantiate an iso9660 class")
i.copy(os.path.join(self.boot_path, os.path.basename(self.kernel)), self.kernel)
assert os.path.getsize(self.kernel) > 0
i.copy(os.path.join(self.boot_path, os.path.basename(self.initrd)), self.initrd)
assert os.path.getsize(self.initrd) > 0
if self.unattended_file.endswith(".preseed"):
self.preseed_initrd()
if self.params.get("vm_type") == "libvirt":
if self.vm.driver_type == "qemu":
# Virtinstall command needs files "vmlinuz" and "initrd.img"
os.chdir(self.image_path)
base_kernel = os.path.basename(self.kernel)
base_initrd = os.path.basename(self.initrd)
if base_kernel != "vmlinuz":
utils.run("mv %s vmlinuz" % base_kernel, verbose=DEBUG)
if base_initrd != "initrd.img":
utils.run("mv %s initrd.img" % base_initrd, verbose=DEBUG)
if self.params.get("unattended_delivery_method") != "integrated":
i.close()
utils_disk.cleanup(self.cdrom_cd1_mount)
elif (self.vm.driver_type == "xen") and (self.params.get("hvm_or_pv") == "pv"):
logging.debug("starting unattended content web server")
self.url_auto_content_port = utils_misc.find_free_port(8100, 8199, self.url_auto_content_ip)
start_auto_content_server_thread(self.url_auto_content_port, self.cdrom_cd1_mount)
self.medium = "url"
self.url = "http://%s:%s" % (self.url_auto_content_ip, self.url_auto_content_port)
pxe_path = os.path.join(os.path.dirname(self.image_path), "xen")
if not os.path.isdir(pxe_path):
os.makedirs(pxe_path)
pxe_kernel = os.path.join(pxe_path, os.path.basename(self.kernel))
pxe_initrd = os.path.join(pxe_path, os.path.basename(self.initrd))
utils.run("cp %s %s" % (self.kernel, pxe_kernel))
utils.run("cp %s %s" % (self.initrd, pxe_initrd))
开发者ID:vliaskov，项目名称:virt-test，代码行数:56，代码来源:unattended_install.py
示例8: migration_scenario
def migration_scenario(self):
srchost = self.params.get("hosts")[0]
dsthost = self.params.get("hosts")[1]
mig_port = None
if params.get("hostid") == self.master_id():
mig_port = utils_misc.find_free_port(5200, 6000)
sync = SyncData(self.master_id(), self.hostid,
self.params.get("hosts"),
{'src': srchost, 'dst': dsthost,
'port': "ports"}, self.sync_server)
mig_port = sync.sync(mig_port, timeout=120)
mig_port = mig_port[srchost]
logging.debug("Migration port %d" % (mig_port))
if params.get("hostid") != self.master_id():
s = self._connect_to_server(srchost, mig_port)
try:
fd = s.fileno()
logging.debug("File descrtiptor %d used for"
" migration." % (fd))
self.migrate_wait(["vm1"], srchost, dsthost, mig_mode="fd",
params_append={"migration_fd": fd})
finally:
s.close()
else:
s = self._create_server(mig_port)
try:
conn, _ = s.accept()
fd = conn.fileno()
logging.debug("File descrtiptor %d used for"
" migration." % (fd))
#Prohibits descriptor inheritance.
flags = fcntl.fcntl(fd, fcntl.F_GETFD)
flags |= fcntl.FD_CLOEXEC
fcntl.fcntl(fd, fcntl.F_SETFD, flags)
self.migrate_wait(["vm1"], srchost, dsthost, mig_mode="fd",
params_append={"migration_fd": fd})
conn.close()
finally:
s.close()
开发者ID:tjamrisk，项目名称:autotest，代码行数:45，代码来源:migration_multi_host_fd.py
示例9: migrate_wait
def migrate_wait(self, vms_name, srchost, dsthost, start_work=None,
check_work=None, mig_mode="fd", params_append=None):
vms_count = len(vms_name)
mig_ports = []
if self.params.get("hostid") == srchost:
last_port = 5199
for _ in range(vms_count):
last_port = utils_misc.find_free_port(last_port + 1, 6000)
mig_ports.append(last_port)
sync = SyncData(self.master_id(), self.hostid,
self.params.get("hosts"),
{'src': srchost, 'dst': dsthost,
'port': "ports"}, self.sync_server)
mig_ports = sync.sync(mig_ports, timeout=120)
mig_ports = mig_ports[srchost]
logging.debug("Migration port %s" % (mig_ports))
if self.params.get("hostid") != srchost:
sockets = []
for mig_port in mig_ports:
sockets.append(self._connect_to_server(srchost, mig_port))
try:
fds = {}
for s, vm_name in zip(sockets, vms_name):
fds["migration_fd_%s" % vm_name] = s.fileno()
logging.debug("File descrtiptors %s used for"
" migration." % (fds))
super_cls = super(MultihostMigrationFd, self)
super_cls.migrate_wait(vms_name, srchost, dsthost,
start_work=start_work, mig_mode="fd",
params_append=fds)
finally:
for s in sockets:
s.close()
else:
sockets = []
for mig_port in mig_ports:
sockets.append(self._create_server(mig_port))
try:
conns = []
for s in sockets:
conns.append(s.accept()[0])
fds = {}
for conn, vm_name in zip(conns, vms_name):
fds["migration_fd_%s" % vm_name] = conn.fileno()
logging.debug("File descrtiptors %s used for"
" migration." % (fds))
# Prohibits descriptor inheritance.
for fd in fds.values():
flags = fcntl.fcntl(fd, fcntl.F_GETFD)
flags |= fcntl.FD_CLOEXEC
fcntl.fcntl(fd, fcntl.F_SETFD, flags)
super_cls = super(MultihostMigrationFd, self)
super_cls.migrate_wait(vms_name, srchost, dsthost,
start_work=start_work, mig_mode="fd",
params_append=fds)
for conn in conns:
conn.close()
finally:
for s in sockets:
s.close()
开发者ID:aginies，项目名称:virt-test，代码行数:67，代码来源:qemu.py
示例10: setup_cdrom
def setup_cdrom(self):
"""
Mount cdrom and copy vmlinuz and initrd.img.
"""
error.context("Copying vmlinuz and initrd.img from install cdrom %s" %
self.cdrom_cd1)
if not os.path.isdir(self.image_path):
os.makedirs(self.image_path)
if (self.params.get('unattended_delivery_method') in
['integrated', 'url']):
i = iso9660.Iso9660Mount(self.cdrom_cd1)
self.cdrom_cd1_mount = i.mnt_dir
else:
i = iso9660.iso9660(self.cdrom_cd1)
if i is None:
raise error.TestFail("Could not instantiate an iso9660 class")
i.copy(os.path.join(self.boot_path, os.path.basename(self.kernel)),
self.kernel)
assert(os.path.getsize(self.kernel) > 0)
i.copy(os.path.join(self.boot_path, os.path.basename(self.initrd)),
self.initrd)
assert(os.path.getsize(self.initrd) > 0)
if self.unattended_file.endswith('.preseed'):
self.preseed_initrd()
if self.params.get("vm_type") == "libvirt":
if self.vm.driver_type == 'qemu':
# Virtinstall command needs files "vmlinuz" and "initrd.img"
os.chdir(self.image_path)
base_kernel = os.path.basename(self.kernel)
base_initrd = os.path.basename(self.initrd)
if base_kernel != 'vmlinuz':
utils.run("mv %s vmlinuz" % base_kernel, verbose=DEBUG)
if base_initrd != 'initrd.img':
utils.run("mv %s initrd.img" % base_initrd, verbose=DEBUG)
if (self.params.get('unattended_delivery_method') !=
'integrated'):
i.close()
utils_disk.cleanup(self.cdrom_cd1_mount)
elif ((self.vm.driver_type == 'xen') and
(self.params.get('hvm_or_pv') == 'pv')):
logging.debug("starting unattended content web server")
self.url_auto_content_port = utils_misc.find_free_port(8100,
8199,
self.url_auto_content_ip)
start_auto_content_server_thread(self.url_auto_content_port,
self.cdrom_cd1_mount)
self.medium = 'url'
self.url = ('http://%s:%s' % (self.url_auto_content_ip,
self.url_auto_content_port))
pxe_path = os.path.join(
os.path.dirname(self.image_path), 'xen')
if not os.path.isdir(pxe_path):
os.makedirs(pxe_path)
pxe_kernel = os.path.join(pxe_path,
os.path.basename(self.kernel))
pxe_initrd = os.path.join(pxe_path,
os.path.basename(self.initrd))
utils.run("cp %s %s" % (self.kernel, pxe_kernel))
utils.run("cp %s %s" % (self.initrd, pxe_initrd))
if 'repo=cdrom' in self.kernel_params:
# Red Hat
self.kernel_params = re.sub('repo\=[\:\w\d\/]*',
'repo=http://%s:%s' %
(self.url_auto_content_ip,
self.url_auto_content_port),
self.kernel_params)
开发者ID:aginies，项目名称:virt-test，代码行数:77，代码来源:unattended_install.py
示例11: setup_boot_disk
def setup_boot_disk(self):
if self.unattended_file.endswith('.sif'):
dest_fname = 'winnt.sif'
setup_file = 'winnt.bat'
boot_disk = utils_disk.FloppyDisk(self.floppy,
self.qemu_img_binary,
self.tmpdir, self.vfd_size)
answer_path = boot_disk.get_answer_file_path(dest_fname)
self.answer_windows_ini(answer_path)
setup_file_path = os.path.join(self.unattended_dir, setup_file)
boot_disk.copy_to(setup_file_path)
if self.install_virtio == "yes":
boot_disk.setup_virtio_win2003(self.virtio_floppy,
self.virtio_oemsetup_id)
boot_disk.copy_to(self.finish_program)
elif self.unattended_file.endswith('.ks'):
# Red Hat kickstart install
dest_fname = 'ks.cfg'
if self.params.get('unattended_delivery_method') == 'integrated':
ks_param = 'ks=cdrom:/dev/sr0:/isolinux/%s' % dest_fname
kernel_params = self.kernel_params
if 'ks=' in kernel_params:
kernel_params = re.sub('ks\=[\w\d\:\.\/]+',
ks_param,
kernel_params)
else:
kernel_params = '%s %s' % (kernel_params, ks_param)
# Standard setting is kickstart disk in /dev/sr0 and
# install cdrom in /dev/sr1. As we merge them together,
# we need to change repo configuration to /dev/sr0
if 'repo=cdrom' in kernel_params:
kernel_params = re.sub('repo\=cdrom[\:\w\d\/]*',
'repo=cdrom:/dev/sr0',
kernel_params)
self.kernel_params = None
boot_disk = utils_disk.CdromInstallDisk(
self.cdrom_unattended,
self.tmpdir,
self.cdrom_cd1_mount,
kernel_params)
elif self.params.get('unattended_delivery_method') == 'url':
if self.unattended_server_port is None:
self.unattended_server_port = utils_misc.find_free_port(
8000,
8099,
self.url_auto_content_ip)
path = os.path.join(os.path.dirname(self.cdrom_unattended),
'ks')
boot_disk = RemoteInstall(path, self.url_auto_content_ip,
self.unattended_server_port,
dest_fname)
ks_param = 'ks=%s' % boot_disk.get_url()
kernel_params = self.kernel_params
if 'ks=' in kernel_params:
kernel_params = re.sub('ks\=[\w\d\:\.\/]+',
ks_param,
kernel_params)
else:
kernel_params = '%s %s' % (kernel_params, ks_param)
# Standard setting is kickstart disk in /dev/sr0 and
# install cdrom in /dev/sr1. When we get ks via http,
# we need to change repo configuration to /dev/sr0
kernel_params = re.sub('repo\=cdrom[\:\w\d\/]*',
'repo=cdrom:/dev/sr0',
kernel_params)
self.kernel_params = kernel_params
elif self.params.get('unattended_delivery_method') == 'cdrom':
boot_disk = utils_disk.CdromDisk(self.cdrom_unattended,
self.tmpdir)
elif self.params.get('unattended_delivery_method') == 'floppy':
boot_disk = utils_disk.FloppyDisk(self.floppy,
self.qemu_img_binary,
self.tmpdir, self.vfd_size)
ks_param = 'ks=floppy'
kernel_params = self.kernel_params
if 'ks=' in kernel_params:
# Reading ks from floppy directly doesn't work in some OS,
# options 'ks=hd:/dev/fd0' can reading ks from mounted
# floppy, so skip repace it;
if not re.search("fd\d+", kernel_params):
kernel_params = re.sub('ks\=[\w\d\:\.\/]+',
ks_param,
kernel_params)
else:
kernel_params = '%s %s' % (kernel_params, ks_param)
kernel_params = re.sub('repo\=cdrom[\:\w\d\/]*',
'repo=cdrom:/dev/sr0',
kernel_params)
self.kernel_params = kernel_params
else:
raise ValueError("Neither cdrom_unattended nor floppy set "
"on the config file, please verify")
answer_path = boot_disk.get_answer_file_path(dest_fname)
#.........这里部分代码省略.........
开发者ID:aginies，项目名称:virt-test，代码行数:101，代码来源:unattended_install.py
示例12: setup_boot_disk
def setup_boot_disk(self):
if self.unattended_file.endswith(".sif"):
dest_fname = "winnt.sif"
setup_file = "winnt.bat"
boot_disk = utils_disk.FloppyDisk(self.floppy, self.qemu_img_binary, self.tmpdir, self.vfd_size)
answer_path = boot_disk.get_answer_file_path(dest_fname)
self.answer_windows_ini(answer_path)
setup_file_path = os.path.join(self.unattended_dir, setup_file)
boot_disk.copy_to(setup_file_path)
if self.install_virtio == "yes":
boot_disk.setup_virtio_win2003(self.virtio_floppy, self.virtio_oemsetup_id)
boot_disk.copy_to(self.finish_program)
elif self.unattended_file.endswith(".ks"):
# Red Hat kickstart install
dest_fname = "ks.cfg"
if self.params.get("unattended_delivery_method") == "integrated":
ks_param = "ks=cdrom:/dev/sr0:/isolinux/%s" % dest_fname
kernel_params = self.kernel_params
if "ks=" in kernel_params:
kernel_params = re.sub("ks\=[\w\d\:\.\/]+", ks_param, kernel_params)
else:
kernel_params = "%s %s" % (kernel_params, ks_param)
# Standard setting is kickstart disk in /dev/sr0 and
# install cdrom in /dev/sr1. As we merge them together,
# we need to change repo configuration to /dev/sr0
if "repo=cdrom" in kernel_params:
kernel_params = re.sub("repo\=cdrom[\:\w\d\/]*", "repo=cdrom:/dev/sr0", kernel_params)
self.kernel_params = None
boot_disk = utils_disk.CdromInstallDisk(
self.cdrom_unattended, self.tmpdir, self.cdrom_cd1_mount, kernel_params
)
elif self.params.get("unattended_delivery_method") == "url":
if self.unattended_server_port is None:
self.unattended_server_port = utils_misc.find_free_port(8000, 8099, self.url_auto_content_ip)
path = os.path.join(os.path.dirname(self.cdrom_unattended), "ks")
boot_disk = RemoteInstall(path, self.url_auto_content_ip, self.unattended_server_port, dest_fname)
ks_param = "ks=%s" % boot_disk.get_url()
kernel_params = self.kernel_params
if "ks=" in kernel_params:
kernel_params = re.sub("ks\=[\w\d\:\.\/]+", ks_param, kernel_params)
else:
kernel_params = "%s %s" % (kernel_params, ks_param)
# Standard setting is kickstart disk in /dev/sr0 and
# install cdrom in /dev/sr1. When we get ks via http,
# we need to change repo configuration to /dev/sr0
kernel_params = re.sub("repo\=cdrom[\:\w\d\/]*", "repo=cdrom:/dev/sr0", kernel_params)
self.kernel_params = kernel_params
elif self.params.get("unattended_delivery_method") == "cdrom":
boot_disk = utils_disk.CdromDisk(self.cdrom_unattended, self.tmpdir)
elif self.params.get("unattended_delivery_method") == "floppy":
boot_disk = utils_disk.FloppyDisk(self.floppy, self.qemu_img_binary, self.tmpdir, self.vfd_size)
ks_param = "ks=floppy"
kernel_params = self.kernel_params
if "ks=" in kernel_params:
kernel_params = re.sub("ks\=[\w\d\:\.\/]+", ks_param, kernel_params)
else:
kernel_params = "%s %s" % (kernel_params, ks_param)
kernel_params = re.sub("repo\=cdrom[\:\w\d\/]*", "repo=cdrom:/dev/sr0", kernel_params)
self.kernel_params = kernel_params
else:
raise ValueError("Neither cdrom_unattended nor floppy set " "on the config file, please verify")
answer_path = boot_disk.get_answer_file_path(dest_fname)
self.answer_kickstart(answer_path)
elif self.unattended_file.endswith(".xml"):
if "autoyast" in self.kernel_params:
# SUSE autoyast install
dest_fname = "autoinst.xml"
if self.cdrom_unattended:
boot_disk = utils_disk.CdromDisk(self.cdrom_unattended, self.tmpdir)
elif self.floppy:
autoyast_param = "autoyast=floppy"
kernel_params = self.kernel_params
if "autoyast=" in kernel_params:
kernel_params = re.sub("autoyast\=[\w\d\:\.\/]+", autoyast_param, kernel_params)
else:
kernel_params = "%s %s" % (kernel_params, autoyast_param)
self.kernel_params = kernel_params
boot_disk = utils_disk.FloppyDisk(self.floppy, self.qemu_img_binary, self.tmpdir, self.vfd_size)
else:
raise ValueError("Neither cdrom_unattended nor floppy set " "on the config file, please verify")
answer_path = boot_disk.get_answer_file_path(dest_fname)
self.answer_suse_xml(answer_path)
else:
# Windows unattended install
dest_fname = "autounattend.xml"
boot_disk = utils_disk.FloppyDisk(self.floppy, self.qemu_img_binary, self.tmpdir, self.vfd_size)
answer_path = boot_disk.get_answer_file_path(dest_fname)
self.answer_windows_xml(answer_path)
if self.install_virtio == "yes":
#.........这里部分代码省略.........
开发者ID:sibiaoluo，项目名称:virt-test，代码行数:101，代码来源:unattended_install.py
示例13: setup_boot_disk
def setup_boot_disk(self):
if self.unattended_file.endswith('.sif'):
dest_fname = 'winnt.sif'
setup_file = 'winnt.bat'
boot_disk = utils_disk.FloppyDisk(self.floppy,
self.qemu_img_binary,
self.tmpdir, self.vfd_size)
answer_path = boot_disk.get_answer_file_path(dest_fname)
self.answer_windows_ini(answer_path)
setup_file_path = os.path.join(self.unattended_dir, setup_file)
boot_disk.copy_to(setup_file_path)
if self.install_virtio == "yes":
boot_disk.setup_virtio_win2003(self.virtio_floppy,
self.virtio_oemsetup_id)
boot_disk.copy_to(self.finish_program)
elif self.unattended_file.endswith('.ks'):
# Red Hat kickstart install
dest_fname = 'ks.cfg'
if self.params.get('unattended_delivery_method') == 'integrated':
ks_param = 'ks=cdrom:/dev/sr0:/isolinux/%s' % dest_fname
kernel_params = getattr(self, 'kernel_params')
if 'ks=' in kernel_params:
kernel_params = re.sub('ks\=[\w\d\:\.\/]+',
ks_param,
kernel_params)
else:
kernel_params = '%s %s' % (kernel_params, ks_param)
# Standard setting is kickstart disk in /dev/sr0 and
# install cdrom in /dev/sr1. As we merge them together,
# we need to change repo configuration to /dev/sr0
if 'repo=cdrom' in kernel_params:
kernel_params = re.sub('repo\=cdrom[\:\w\d\/]*',
'repo=cdrom:/dev/sr0',
kernel_params)
self.params['kernel_params'] = ''
boot_disk = utils_disk.CdromInstallDisk(
self.cdrom_unattended,
self.tmpdir,
self.cdrom_cd1_mount,
kernel_params)
elif self.params.get('unattended_delivery_method') == 'url':
if self.unattended_server_port is None:
self.unattended_server_port = utils_misc.find_free_port(
8000,
8099,
self.url_auto_content_ip)
path = os.path.join(os.path.dirname(self.cdrom_unattended),
'ks')
boot_disk = RemoteInstall(path, self.url_auto_content_ip,
self.unattended_server_port,
dest_fname)
ks_param = 'ks=%s' % boot_disk.get_url()
kernel_params = getattr(self, 'kernel_params')
if 'ks=' in kernel_params:
kernel_params = re.sub('ks\=[\w\d\:\.\/]+',
ks_param,
kernel_params)
else:
kernel_params = '%s %s' % (kernel_params, ks_param)
# Standard setting is kickstart disk in /dev/sr0 and
# install cdrom in /dev/sr1. When we get ks via http,
# we need to change repo configuration to /dev/sr0
if 'repo=cdrom' in kernel_params:
if ((self.vm.driver_type == 'xen') and
(self.params.get('hvm_or_pv') == 'pv')):
kernel_params = re.sub('repo\=[\:\w\d\/]*',
'repo=http://%s:%s' %
(self.url_auto_content_ip,
self.url_auto_content_port),
kernel_params)
else:
kernel_params = re.sub('repo\=cdrom[\:\w\d\/]*',
'repo=cdrom:/dev/sr0',
kernel_params)
self.params['kernel_params'] = kernel_params
elif self.params.get('unattended_delivery_method') == 'cdrom':
boot_disk = utils_disk.CdromDisk(self.cdrom_unattended,
self.tmpdir)
elif self.params.get('unattended_delivery_method') == 'floppy':
boot_disk = utils_disk.FloppyDisk(self.floppy,
self.qemu_img_binary,
self.tmpdir, self.vfd_size)
else:
raise ValueError("Neither cdrom_unattended nor floppy set "
"on the config file, please verify")
answer_path = boot_disk.get_answer_file_path(dest_fname)
self.answer_kickstart(answer_path)
elif self.unattended_file.endswith('.xml'):
if "autoyast" in self.kernel_params:
# SUSE autoyast install
dest_fname = "autoinst.xml"
if self.cdrom_unattended:
boot_disk = utils_disk.CdromDisk(self.cdrom_unattended,
self.tmpdir)
#.........这里部分代码省略.........
开发者ID:kongove，项目名称:virt-test，代码行数:101，代码来源:unattended_install.py
注：本文中的virttest.utils_misc.find_free_port函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论