arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-28 03:08:14 +00:00
Code Issues Releases Wiki Activity

Dirty stop start for VPCS

Browse Source
This commit is contained in:
Julien Duponchelle 2015-01-14 18:52:02 +01:00
parent 482fdf9031
commit 6c35cc304e
15 changed files with 811 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
gns3server/handlers/vpcs_handler.py
View File

@ -19,16 +19,13 @@ from ..web.route import Route
from ..schemas.vpcs import VPCS_CREATE_SCHEMA
from ..schemas.vpcs import VPCS_OBJECT_SCHEMA
from ..schemas.vpcs import VPCS_ADD_NIO_SCHEMA
from ..modules import VPCS
from ..modules.vpcs import VPCS
class VPCSHandler(object):
@classmethod
@Route.post(
r"/vpcs",
parameters={
"vpcs_id": "Id of VPCS instance"
},
status_codes={
201: "Success of creation of VPCS",
409: "Conflict"
@ -43,6 +40,38 @@ class VPCSHandler(object):
"vpcs_id": vm.id,
"console": 4242})
@classmethod
@Route.post(
r"/vpcs/{vpcs_id}/start",
parameters={
"vpcs_id": "Id of VPCS instance"
},
status_codes={
201: "Success of creation of VPCS",
},
description="Start VPCS",
)
def create(request, response):
vpcs_manager = VPCS.instance()
vm = yield from vpcs_manager.start_vm(int(request.match_info['vpcs_id']))
response.json({})
@classmethod
@Route.post(
r"/vpcs/{vpcs_id}/stop",
parameters={
"vpcs_id": "Id of VPCS instance"
},
status_codes={
201: "Success of stopping VPCS",
},
description="Stop VPCS",
)
def create(request, response):
vpcs_manager = VPCS.instance()
vm = yield from vpcs_manager.stop_vm(int(request.match_info['vpcs_id']))
response.json({})
@classmethod
@Route.get(
r"/vpcs/{vpcs_id}",

10
gns3server/modules/vm_manager.py → gns3server/modules/base_manager.py
View File

@ -19,12 +19,12 @@
import asyncio
import aiohttp
from .vm_error import VMError
from .device_error import DeviceError
class VMManager:
class BaseManager:
"""
Base class for all VMManager.
Base class for all Manager.
Responsible of management of a VM pool
"""
@ -69,10 +69,10 @@ class VMManager:
identifier = i
break
if identifier == 0:
raise VMError("Maximum number of VM instances reached")
raise DeviceError("Maximum number of VM instances reached")
else:
if identifier in self._vms:
raise VMError("VM identifier {} is already used by another VM instance".format(identifier))
raise DeviceError("VM identifier {} is already used by another VM instance".format(identifier))
vm = self._VM_CLASS(vmname, identifier)
yield from vm.wait_for_creation()
self._vms[vm.id] = vm

73
gns3server/modules/base_vm.py
View File

@ -17,18 +17,74 @@
import asyncio
from .vm_error import VMError
from .device_error import DeviceError
from .attic import find_unused_port
import logging
log = logging.getLogger(__name__)
class BaseVM:
_allocated_console_ports = []
def __init__(self, name, identifier):
self._loop = asyncio.get_event_loop()
self._allocate_console()
self._queue = asyncio.Queue()
self._name = name
self._id = identifier
self._created = asyncio.Future()
self._worker = asyncio.async(self._run())
log.info("{type} device {name} [id={id}] has been created".format(
type=self.__class__.__name__,
name=self._name,
id=self._id))
def _allocate_console(self):
if not self._console:
# allocate a console port
try:
self._console = find_unused_port(self._console_start_port_range,
self._console_end_port_range,
self._console_host,
ignore_ports=self._allocated_console_ports)
except Exception as e:
raise DeviceError(e)
if self._console in self._allocated_console_ports:
raise DeviceError("Console port {} is already used by another device".format(console))
self._allocated_console_ports.append(self._console)
@property
def console(self):
"""
Returns the TCP console port.
:returns: console port (integer)
"""
return self._console
@console.setter
def console(self, console):
"""
Sets the TCP console port.
:param console: console port (integer)
"""
if console in self._allocated_console_ports:
raise VPCSError("Console port {} is already used by another VPCS device".format(console))
self._allocated_console_ports.remove(self._console)
self._console = console
self._allocated_console_ports.append(self._console)
log.info("{type} {name} [id={id}]: console port set to {port}".format(
type=self.__class__.__name__,
name=self._name,
id=self._id,
port=console))
@property
def id(self):
"""
@ -65,7 +121,7 @@ class BaseVM:
try:
yield from self._create()
self._created.set_result(True)
except VMError as e:
except DeviceError as e:
self._created.set_exception(e)
return
@ -75,7 +131,7 @@ class BaseVM:
try:
yield from asyncio.wait_for(self._execute(subcommand, args), timeout=timeout)
except asyncio.TimeoutError:
raise VMError("{} has timed out after {} seconds!".format(subcommand, timeout))
raise DeviceError("{} has timed out after {} seconds!".format(subcommand, timeout))
future.set_result(True)
except Exception as e:
future.set_exception(e)
@ -83,6 +139,15 @@ class BaseVM:
def wait_for_creation(self):
return self._created
@asyncio.coroutine
def start():
"""
Starts the VM process.
"""
raise NotImplementedError
def put(self, *args):
"""
Add to the processing queue of the VM
@ -95,5 +160,5 @@ class BaseVM:
args.insert(0, future)
self._queue.put_nowait(args)
except asyncio.qeues.QueueFull:
raise VMError("Queue is full")
raise DeviceError("Queue is full")
return future

2
gns3server/modules/vm_error.py → gns3server/modules/device_error.py
View File

@ -16,5 +16,5 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
class VMError(Exception):
class DeviceError(Exception):
pass

7
gns3server/modules/vpcs/__init__.py
View File

@ -19,12 +19,9 @@
VPCS server module.
"""
from ..vm_manager import VMManager
from ..base_manager import BaseManager
from .vpcs_device import VPCSDevice
class VPCS(VMManager):
class VPCS(BaseManager):
_VM_CLASS = VPCSDevice
def create_vm(self, name):
return super().create_vm(name)

0
gns3server/modules/vpcs/adapters/__init__.py Normal file
View File

104
gns3server/modules/vpcs/adapters/adapter.py Normal file
View File

@ -0,0 +1,104 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2014 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
class Adapter(object):
"""
Base class for adapters.
:param interfaces: number of interfaces supported by this adapter.
"""
def __init__(self, interfaces=1):
self._interfaces = interfaces
self._ports = {}
for port_id in range(0, interfaces):
self._ports[port_id] = None
def removable(self):
"""
Returns True if the adapter can be removed from a slot
and False if not.
:returns: boolean
"""
return True
def port_exists(self, port_id):
"""
Checks if a port exists on this adapter.
:returns: True is the port exists,
False otherwise.
"""
if port_id in self._ports:
return True
return False
def add_nio(self, port_id, nio):
"""
Adds a NIO to a port on this adapter.
:param port_id: port ID (integer)
:param nio: NIO instance
"""
self._ports[port_id] = nio
def remove_nio(self, port_id):
"""
Removes a NIO from a port on this adapter.
:param port_id: port ID (integer)
"""
self._ports[port_id] = None
def get_nio(self, port_id):
"""
Returns the NIO assigned to a port.
:params port_id: port ID (integer)
:returns: NIO instance
"""
return self._ports[port_id]
@property
def ports(self):
"""
Returns port to NIO mapping
:returns: dictionary port -> NIO
"""
return self._ports
@property
def interfaces(self):
"""
Returns the number of interfaces supported by this adapter.
:returns: number of interfaces
"""
return self._interfaces

31
gns3server/modules/vpcs/adapters/ethernet_adapter.py Normal file
View File

@ -0,0 +1,31 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2014 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .adapter import Adapter
class EthernetAdapter(Adapter):
"""
VPCS Ethernet adapter.
"""
def __init__(self):
Adapter.__init__(self, interfaces=1)
def __str__(self):
return "VPCS Ethernet adapter"

0
gns3server/modules/vpcs/nios/__init__.py Normal file
View File

46
gns3server/modules/vpcs/nios/nio_tap.py Normal file
View File

@ -0,0 +1,46 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2013 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Interface for TAP NIOs (UNIX based OSes only).
"""
class NIO_TAP(object):
"""
TAP NIO.
:param tap_device: TAP device name (e.g. tap0)
"""
def __init__(self, tap_device):
self._tap_device = tap_device
@property
def tap_device(self):
"""
Returns the TAP device used by this NIO.
:returns: the TAP device name
"""
return self._tap_device
def __str__(self):
return "NIO TAP"

72
gns3server/modules/vpcs/nios/nio_udp.py Normal file
View File

@ -0,0 +1,72 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2013 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Interface for UDP NIOs.
"""
class NIO_UDP(object):
"""
UDP NIO.
:param lport: local port number
:param rhost: remote address/host
:param rport: remote port number
"""
_instance_count = 0
def __init__(self, lport, rhost, rport):
self._lport = lport
self._rhost = rhost
self._rport = rport
@property
def lport(self):
"""
Returns the local port
:returns: local port number
"""
return self._lport
@property
def rhost(self):
"""
Returns the remote host
:returns: remote address/host
"""
return self._rhost
@property
def rport(self):
"""
Returns the remote port
:returns: remote port number
"""
return self._rport
def __str__(self):
return "NIO UDP"

363
gns3server/modules/vpcs/vpcs_device.py
View File

@ -15,9 +15,370 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
VPCS device management (creates command line, processes, files etc.) in
order to run an VPCS instance.
"""
import os
import sys
import subprocess
import signal
import shutil
import re
import asyncio
from pkg_resources import parse_version
from .vpcs_error import VPCSError
from .adapters.ethernet_adapter import EthernetAdapter
from .nios.nio_udp import NIO_UDP
from .nios.nio_tap import NIO_TAP
from ..base_vm import BaseVM
import logging
log = logging.getLogger(__name__)
class VPCSDevice(BaseVM):
pass
"""
VPCS device implementation.
:param name: name of this VPCS device
:param vpcs_id: VPCS instance ID
:param path: path to VPCS executable
:param working_dir: path to a working directory
:param console: TCP console port
:param console_host: IP address to bind for console connections
:param console_start_port_range: TCP console port range start
:param console_end_port_range: TCP console port range end
"""
def __init__(self, name, vpcs_id,
path = None,
working_dir = None,
console=None,
console_host="0.0.0.0",
console_start_port_range=4512,
console_end_port_range=5000):
#self._path = path
#self._working_dir = working_dir
# TODO: Hardcodded for testing
self._path = "/usr/local/bin/vpcs"
self._working_dir = "/tmp"
self._console = console
self._console_host = console_host
self._command = []
self._process = None
self._vpcs_stdout_file = ""
self._started = False
self._console_start_port_range = console_start_port_range
self._console_end_port_range = console_end_port_range
# VPCS settings
self._script_file = ""
self._ethernet_adapter = EthernetAdapter() # one adapter with 1 Ethernet interface
# working_dir_path = os.path.join(working_dir, "vpcs", "pc-{}".format(self._id))
#
# if vpcs_id and not os.path.isdir(working_dir_path):
# raise VPCSError("Working directory {} doesn't exist".format(working_dir_path))
#
# # create the device own working directory
# self.working_dir = working_dir_path
#
super().__init__(name, vpcs_id)
@asyncio.coroutine
def _create(self):
"""Called when run loop is started"""
self._check_requirement()
def _check_requirement(self):
"""Check if VPCS is available with the correct version"""
if not self._path:
raise VPCSError("No path to a VPCS executable has been set")
if not os.path.isfile(self._path):
raise VPCSError("VPCS program '{}' is not accessible".format(self._path))
if not os.access(self._path, os.X_OK):
raise VPCSError("VPCS program '{}' is not executable".format(self._path))
yield from self._check_vpcs_version()
def defaults(self):
"""
Returns all the default attribute values for VPCS.
:returns: default values (dictionary)
"""
vpcs_defaults = {"name": self._name,
"script_file": self._script_file,
"console": self._console}
return vpcs_defaults
@classmethod
def reset(cls):
"""
Resets allocated instance list.
"""
cls._instances.clear()
cls._allocated_console_ports.clear()
@property
def name(self):
"""
Returns the name of this VPCS device.
:returns: name
"""
return self._name
@name.setter
def name(self, new_name):
"""
Sets the name of this VPCS device.
:param new_name: name
"""
if self._script_file:
# update the startup.vpc
config_path = os.path.join(self._working_dir, "startup.vpc")
if os.path.isfile(config_path):
try:
with open(config_path, "r+", errors="replace") as f:
old_config = f.read()
new_config = old_config.replace(self._name, new_name)
f.seek(0)
f.write(new_config)
except OSError as e:
raise VPCSError("Could not amend the configuration {}: {}".format(config_path, e))
log.info("VPCS {name} [id={id}]: renamed to {new_name}".format(name=self._name,
id=self._id,
new_name=new_name))
self._name = new_name
@asyncio.coroutine
def _check_vpcs_version(self):
"""
Checks if the VPCS executable version is >= 0.5b1.
"""
#TODO: should be async
try:
output = subprocess.check_output([self._path, "-v"], cwd=self._working_dir)
match = re.search("Welcome to Virtual PC Simulator, version ([0-9a-z\.]+)", output.decode("utf-8"))
if match:
version = match.group(1)
if parse_version(version) < parse_version("0.5b1"):
raise VPCSError("VPCS executable version must be >= 0.5b1")
else:
raise VPCSError("Could not determine the VPCS version for {}".format(self._path))
except (OSError, subprocess.SubprocessError) as e:
raise VPCSError("Error while looking for the VPCS version: {}".format(e))
@asyncio.coroutine
def start(self):
"""
Starts the VPCS process.
"""
if not self.is_running():
# if not self._ethernet_adapter.get_nio(0):
# raise VPCSError("This VPCS instance must be connected in order to start")
self._command = self._build_command()
try:
log.info("starting VPCS: {}".format(self._command))
self._vpcs_stdout_file = os.path.join(self._working_dir, "vpcs.log")
log.info("logging to {}".format(self._vpcs_stdout_file))
flags = 0
if sys.platform.startswith("win32"):
flags = subprocess.CREATE_NEW_PROCESS_GROUP
with open(self._vpcs_stdout_file, "w") as fd:
self._process = yield from asyncio.create_subprocess_exec(*self._command,
stdout=fd,
stderr=subprocess.STDOUT,
cwd=self._working_dir,
creationflags=flags)
log.info("VPCS instance {} started PID={}".format(self._id, self._process.pid))
self._started = True
except (OSError, subprocess.SubprocessError) as e:
vpcs_stdout = self.read_vpcs_stdout()
log.error("could not start VPCS {}: {}\n{}".format(self._path, e, vpcs_stdout))
raise VPCSError("could not start VPCS {}: {}\n{}".format(self._path, e, vpcs_stdout))
@asyncio.coroutine
def stop(self):
"""
Stops the VPCS process.
"""
# stop the VPCS process
if self.is_running():
log.info("stopping VPCS instance {} PID={}".format(self._id, self._process.pid))
if sys.platform.startswith("win32"):
self._process.send_signal(signal.CTRL_BREAK_EVENT)
else:
self._process.terminate()
self._process.wait()
self._process = None
self._started = False
def read_vpcs_stdout(self):
"""
Reads the standard output of the VPCS process.
Only use when the process has been stopped or has crashed.
"""
output = ""
if self._vpcs_stdout_file:
try:
with open(self._vpcs_stdout_file, errors="replace") as file:
output = file.read()
except OSError as e:
log.warn("could not read {}: {}".format(self._vpcs_stdout_file, e))
return output
def is_running(self):
"""
Checks if the VPCS process is running
:returns: True or False
"""
if self._process:
return True
return False
def port_add_nio_binding(self, port_id, nio):
"""
Adds a port NIO binding.
:param port_id: port ID
:param nio: NIO instance to add to the slot/port
"""
if not self._ethernet_adapter.port_exists(port_id):
raise VPCSError("Port {port_id} doesn't exist in adapter {adapter}".format(adapter=self._ethernet_adapter,
port_id=port_id))
self._ethernet_adapter.add_nio(port_id, nio)
log.info("VPCS {name} [id={id}]: {nio} added to port {port_id}".format(name=self._name,
id=self._id,
nio=nio,
port_id=port_id))
def port_remove_nio_binding(self, port_id):
"""
Removes a port NIO binding.
:param port_id: port ID
:returns: NIO instance
"""
if not self._ethernet_adapter.port_exists(port_id):
raise VPCSError("Port {port_id} doesn't exist in adapter {adapter}".format(adapter=self._ethernet_adapter,
port_id=port_id))
nio = self._ethernet_adapter.get_nio(port_id)
self._ethernet_adapter.remove_nio(port_id)
log.info("VPCS {name} [id={id}]: {nio} removed from port {port_id}".format(name=self._name,
id=self._id,
nio=nio,
port_id=port_id))
return nio
def _build_command(self):
"""
Command to start the VPCS process.
(to be passed to subprocess.Popen())
VPCS command line:
usage: vpcs [options] [scriptfile]
Option:
-h print this help then exit
-v print version information then exit
-i num number of vpc instances to start (default is 9)
-p port run as a daemon listening on the tcp 'port'
-m num start byte of ether address, default from 0
-r file load and execute script file
compatible with older versions, DEPRECATED.
-e tap mode, using /dev/tapx by default (linux only)
-u udp mode, default
udp mode options:
-s port local udp base port, default from 20000
-c port remote udp base port (dynamips udp port), default from 30000
-t ip remote host IP, default 127.0.0.1
tap mode options:
-d device device name, works only when -i is set to 1
hypervisor mode option:
-H port run as the hypervisor listening on the tcp 'port'
If no 'scriptfile' specified, vpcs will read and execute the file named
'startup.vpc' if it exsits in the current directory.
"""
command = [self._path]
command.extend(["-p", str(self._console)]) # listen to console port
nio = self._ethernet_adapter.get_nio(0)
if nio:
if isinstance(nio, NIO_UDP):
# UDP tunnel
command.extend(["-s", str(nio.lport)]) # source UDP port
command.extend(["-c", str(nio.rport)]) # destination UDP port
command.extend(["-t", nio.rhost]) # destination host
elif isinstance(nio, NIO_TAP):
# TAP interface
command.extend(["-e"])
command.extend(["-d", nio.tap_device])
command.extend(["-m", str(self._id)]) # the unique ID is used to set the MAC address offset
command.extend(["-i", "1"]) # option to start only one VPC instance
command.extend(["-F"]) # option to avoid the daemonization of VPCS
if self._script_file:
command.extend([self._script_file])
return command
@property
def script_file(self):
"""
Returns the script-file for this VPCS instance.
:returns: path to script-file
"""
return self._script_file
@script_file.setter
def script_file(self, script_file):
"""
Sets the script-file for this VPCS instance.
:param script_file: path to base-script-file
"""
self._script_file = script_file
log.info("VPCS {name} [id={id}]: script_file set to {config}".format(name=self._name,
id=self._id,
config=self._script_file))

40
gns3server/modules/vpcs/vpcs_error.py Normal file
View File

@ -0,0 +1,40 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2013 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Custom exceptions for VPCS module.
"""
from ..device_error import DeviceError
class VPCSError(DeviceError):
def __init__(self, message, original_exception=None):
Exception.__init__(self, message)
if isinstance(message, Exception):
message = str(message)
self._message = message
self._original_exception = original_exception
def __repr__(self):
return self._message
def __str__(self):
return self._message

5
gns3server/web/route.py
View File

@ -20,6 +20,7 @@ import jsonschema
import asyncio
import aiohttp
from ..modules.device_error import DeviceError
from .response import Response
@ -97,6 +98,10 @@ class Route(object):
response = Response(route=route)
response.set_status(e.status)
response.json({"message": e.text, "status": e.status})
except DeviceError as e:
response = Response(route=route)
response.set_status(400)
response.json({"message": str(e), "status": 400})
return response
cls._routes.append((method, cls._path, control_schema))

41
tests/modules/vpcs/test_vpcs_device.py Normal file
View File

@ -0,0 +1,41 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
from unittest.mock import patch
from gns3server.modules.vpcs.vpcs_device import VPCSDevice
from gns3server.modules.vpcs.vpcs_error import VPCSError
@patch("subprocess.check_output", return_value="Welcome to Virtual PC Simulator, version 0.6".encode("utf-8"))
def test_vm(tmpdir):
vm = VPCSDevice("test", 42, working_dir=str(tmpdir), path="/bin/test")
assert vm.name == "test"
assert vm.id == 42
@patch("subprocess.check_output", return_value="Welcome to Virtual PC Simulator, version 0.1".encode("utf-8"))
def test_vm_invalid_vpcs_version(tmpdir):
with pytest.raises(VPCSError):
vm = VPCSDevice("test", 42, working_dir=str(tmpdir), path="/bin/test")
assert vm.name == "test"
assert vm.id == 42
def test_vm_invalid_vpcs_path(tmpdir):
with pytest.raises(VPCSError):
vm = VPCSDevice("test", 42, working_dir=str(tmpdir), path="/bin/test_fake")
assert vm.name == "test"
assert vm.id == 42