Compare commits

...

4 Commits

@ -88,7 +88,7 @@ class Docker(BaseManager):
f"{stdout.decode('utf-8', errors='ignore').strip()}") f"{stdout.decode('utf-8', errors='ignore').strip()}")
except OSError as e: except OSError as e:
raise DockerError(f"Could not install busybox: {e}") raise DockerError(f"Could not install busybox: {e}")
raise DockerError("No busybox executable could be found") raise DockerError("No busybox executable could be found, please install busybox (apt install busybox-static on Debian/Ubuntu) and make sure it is in your PATH")
@staticmethod @staticmethod
def resources_path(): def resources_path():

@ -226,7 +226,7 @@ class Qemu(BaseManager):
else: else:
try: try:
output = await subprocess_check_output(qemu_path, "-version", "-nographic") output = await subprocess_check_output(qemu_path, "-version", "-nographic")
match = re.search("version\s+([0-9a-z\-\.]+)", output) match = re.search(r"version\s+([0-9a-z\-\.]+)", output)
if match: if match:
version = match.group(1) version = match.group(1)
return version return version

@ -701,7 +701,7 @@ class VMware(BaseManager):
path = ctypes.create_unicode_buffer(ctypes.wintypes.MAX_PATH) path = ctypes.create_unicode_buffer(ctypes.wintypes.MAX_PATH)
ctypes.windll.shell32.SHGetFolderPathW(None, 5, None, 0, path) ctypes.windll.shell32.SHGetFolderPathW(None, 5, None, 0, path)
documents_folder = path.value documents_folder = path.value
return ['{}\My Virtual Machines'.format(documents_folder), '{}\Virtual Machines'.format(documents_folder)] return [r'{}\My Virtual Machines'.format(documents_folder), r'{}\Virtual Machines'.format(documents_folder)]
elif sys.platform.startswith("darwin"): elif sys.platform.startswith("darwin"):
return [os.path.expanduser("~/Documents/Virtual Machines.localized")] return [os.path.expanduser("~/Documents/Virtual Machines.localized")]
else: else:

@ -189,7 +189,7 @@ class UDPLink(Link):
:returns: Node where the capture should run :returns: Node where the capture should run
""" """
ALWAYS_RUNNING_NODES_TYPE = ("cloud", "nat", "ethernet_switch", "ethernet_hub") ALWAYS_RUNNING_NODES_TYPE = ("cloud", "nat", "ethernet_switch", "ethernet_hub", "frame_relay_switch", "atm_switch")
for node in self._nodes: for node in self._nodes:
if node["node"].compute.id == "local" and node["node"].node_type in ALWAYS_RUNNING_NODES_TYPE and node["node"].status == "started": if node["node"].compute.id == "local" and node["node"].node_type in ALWAYS_RUNNING_NODES_TYPE and node["node"].status == "started":

@ -268,7 +268,7 @@ class ATMSwitchHandler:
response.json({"pcap_file_path": pcap_file_path}) response.json({"pcap_file_path": pcap_file_path})
@Route.post( @Route.post(
r"/projects/{project_id}/atm_relay_switch/nodes/{node_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/stop_capture", r"/projects/{project_id}/atm_switch/nodes/{node_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/stop_capture",
parameters={ parameters={
"project_id": "Project UUID", "project_id": "Project UUID",
"node_id": "Node UUID", "node_id": "Node UUID",
@ -290,7 +290,7 @@ class ATMSwitchHandler:
response.set_status(204) response.set_status(204)
@Route.get( @Route.get(
r"/projects/{project_id}/atm_relay_switch/nodes/{node_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/pcap", r"/projects/{project_id}/atm_switch/nodes/{node_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/pcap",
description="Stream the pcap capture file", description="Stream the pcap capture file",
parameters={ parameters={
"project_id": "Project UUID", "project_id": "Project UUID",

@ -60,7 +60,7 @@ def parse_version(version):
""" """
release_type_found = False release_type_found = False
version_infos = re.split('(\.|[a-z]+)', version) version_infos = re.split(r'(\.|[a-z]+)', version)
version = [] version = []
for info in version_infos: for info in version_infos:
if info == '.' or len(info) == 0: if info == '.' or len(info) == 0:

@ -46,7 +46,7 @@ def _get_windows_interfaces_from_registry():
hkeycard = winreg.OpenKey(hkey, network_card_id) hkeycard = winreg.OpenKey(hkey, network_card_id)
guid, _ = winreg.QueryValueEx(hkeycard, "ServiceName") guid, _ = winreg.QueryValueEx(hkeycard, "ServiceName")
netcard, _ = winreg.QueryValueEx(hkeycard, "Description") netcard, _ = winreg.QueryValueEx(hkeycard, "Description")
connection = r"SYSTEM\CurrentControlSet\Control\Network\{4D36E972-E325-11CE-BFC1-08002BE10318}" + "\{}\Connection".format(guid) connection = r"SYSTEM\CurrentControlSet\Control\Network\{4D36E972-E325-11CE-BFC1-08002BE10318}" + r"\{}\Connection".format(guid)
hkeycon = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, connection) hkeycon = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, connection)
name, _ = winreg.QueryValueEx(hkeycon, "Name") name, _ = winreg.QueryValueEx(hkeycon, "Name")
interface = r"SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces\{}".format(guid) interface = r"SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces\{}".format(guid)
@ -92,7 +92,7 @@ def get_windows_interfaces():
interfaces = [] interfaces = []
try: try:
locator = win32com.client.Dispatch("WbemScripting.SWbemLocator") locator = win32com.client.Dispatch("WbemScripting.SWbemLocator")
service = locator.ConnectServer(".", "root\cimv2") service = locator.ConnectServer(".", r"root\cimv2")
network_configs = service.InstancesOf("Win32_NetworkAdapterConfiguration") network_configs = service.InstancesOf("Win32_NetworkAdapterConfiguration")
# more info on Win32_NetworkAdapter: http://msdn.microsoft.com/en-us/library/aa394216%28v=vs.85%29.aspx # more info on Win32_NetworkAdapter: http://msdn.microsoft.com/en-us/library/aa394216%28v=vs.85%29.aspx
for adapter in service.InstancesOf("Win32_NetworkAdapter"): for adapter in service.InstancesOf("Win32_NetworkAdapter"):

@ -238,7 +238,7 @@ async def test_install_busybox_dynamic_linked():
with pytest.raises(DockerError) as e: with pytest.raises(DockerError) as e:
dst_dir = Docker.resources_path() dst_dir = Docker.resources_path()
await Docker.install_busybox(dst_dir) await Docker.install_busybox(dst_dir)
assert str(e.value) == "No busybox executable could be found" assert str(e.value) == "No busybox executable could be found, please install busybox (apt install busybox-static on Debian/Ubuntu) and make sure it is in your PATH"
@pytest.mark.asyncio @pytest.mark.asyncio
@ -249,4 +249,4 @@ async def test_install_busybox_no_executables():
with pytest.raises(DockerError) as e: with pytest.raises(DockerError) as e:
dst_dir = Docker.resources_path() dst_dir = Docker.resources_path()
await Docker.install_busybox(dst_dir) await Docker.install_busybox(dst_dir)
assert str(e.value) == "No busybox executable could be found" assert str(e.value) == "No busybox executable could be found, please install busybox (apt install busybox-static on Debian/Ubuntu) and make sure it is in your PATH"

Loading…
Cancel
Save