arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-12-01 04:38:12 +00:00
Code Issues Releases Wiki Activity

Fix test to support Python 3.7 Ref https://github.com/GNS3/gns3-gui/issues/2566

Browse Source
This commit is contained in:
grossmj 2018-08-25 15:50:08 +07:00
parent 902de3dd47
commit f5dc635baa
4 changed files with 32 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
gns3server/handlers/api/controller/project_handler.py
View File

@ -16,6 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
import os import os
import sys
import aiohttp import aiohttp
import asyncio import asyncio
import tempfile import tempfile
@ -333,7 +334,18 @@ class ProjectHandler:
# We write the content to a temporary location and after we extract it all. # We write the content to a temporary location and after we extract it all.
# It could be more optimal to stream this but it is not implemented in Python. # It could be more optimal to stream this but it is not implemented in Python.
# Spooled means the file is temporary kept in memory until max_size is reached # Spooled means the file is temporary kept in memory until max_size is reached
# Cannot use tempfile.SpooledTemporaryFile(max_size=10000) in Python 3.7 due
# to a bug https://bugs.python.org/issue26175
try: try:
if sys.version_info >= (3, 7) and sys.version_info < (3, 8):
with tempfile.TemporaryFile() as temp:
while True:
chunk = yield from request.content.read(1024)
if not chunk:
break
temp.write(chunk)
project = yield from import_project(controller, request.match_info["project_id"], temp, location=path, name=name)
else:
with tempfile.SpooledTemporaryFile(max_size=10000) as temp: with tempfile.SpooledTemporaryFile(max_size=10000) as temp:
while True: while True:
chunk = yield from request.content.read(1024) chunk = yield from request.content.read(1024)

6
tests/compute/qemu/test_qemu_vm.py
View File

@ -150,9 +150,9 @@ def test_termination_callback(vm, async_run):
async_run(vm._termination_callback(0)) async_run(vm._termination_callback(0))
assert vm.status == "stopped" assert vm.status == "stopped"
async_run(queue.get(0)) #  Ping async_run(queue.get(1)) #  Ping
(action, event, kwargs) = async_run(queue.get(0)) (action, event, kwargs) = async_run(queue.get(1))
assert action == "node.updated" assert action == "node.updated"
assert event == vm assert event == vm
@ -170,7 +170,7 @@ def test_termination_callback_error(vm, tmpdir, async_run):
async_run(vm._termination_callback(1)) async_run(vm._termination_callback(1))
assert vm.status == "stopped" assert vm.status == "stopped"
async_run(queue.get(0)) # Ping async_run(queue.get(1)) # Ping
(action, event, kwargs) = queue.get_nowait() (action, event, kwargs) = queue.get_nowait()
assert action == "node.updated" assert action == "node.updated"

10
tests/compute/traceng/test_traceng_vm.py
View File

@ -66,7 +66,7 @@ def test_start(loop, vm, async_run):
process.returncode = None process.returncode = None
with NotificationManager.instance().queue() as queue: with NotificationManager.instance().queue() as queue:
async_run(queue.get(0)) # Ping async_run(queue.get(1)) # Ping
vm.ip_address = "192.168.1.1" vm.ip_address = "192.168.1.1"
with patch("sys.platform", return_value="win"): with patch("sys.platform", return_value="win"):
@ -88,7 +88,7 @@ def test_start(loop, vm, async_run):
'192.168.1.2') '192.168.1.2')
assert vm.is_running() assert vm.is_running()
assert vm.command_line == ' '.join(mock_exec.call_args[0]) assert vm.command_line == ' '.join(mock_exec.call_args[0])
(action, event, kwargs) = async_run(queue.get(0)) (action, event, kwargs) = async_run(queue.get(1))
assert action == "node.updated" assert action == "node.updated"
assert event == vm assert event == vm
@ -120,10 +120,10 @@ def test_stop(loop, vm, async_run):
process.terminate.assert_called_with() process.terminate.assert_called_with()
async_run(queue.get(0)) #  Ping async_run(queue.get(1)) #  Ping
async_run(queue.get(0)) #  Started async_run(queue.get(1)) #  Started
(action, event, kwargs) = async_run(queue.get(0)) (action, event, kwargs) = async_run(queue.get(1))
assert action == "node.updated" assert action == "node.updated"
assert event == vm assert event == vm

10
tests/compute/vpcs/test_vpcs_vm.py
View File

@ -91,7 +91,7 @@ def test_start(loop, vm, async_run):
process.returncode = None process.returncode = None
with NotificationManager.instance().queue() as queue: with NotificationManager.instance().queue() as queue:
async_run(queue.get(0)) # Ping async_run(queue.get(1)) # Ping
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True): with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec: with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec:
@ -113,7 +113,7 @@ def test_start(loop, vm, async_run):
'127.0.0.1') '127.0.0.1')
assert vm.is_running() assert vm.is_running()
assert vm.command_line == ' '.join(mock_exec.call_args[0]) assert vm.command_line == ' '.join(mock_exec.call_args[0])
(action, event, kwargs) = async_run(queue.get(0)) (action, event, kwargs) = async_run(queue.get(1))
assert action == "node.updated" assert action == "node.updated"
assert event == vm assert event == vm
@ -177,10 +177,10 @@ def test_stop(loop, vm, async_run):
else: else:
process.terminate.assert_called_with() process.terminate.assert_called_with()
async_run(queue.get(0)) #  Ping async_run(queue.get(1)) #  Ping
async_run(queue.get(0)) #  Started async_run(queue.get(1)) #  Started
(action, event, kwargs) = async_run(queue.get(0)) (action, event, kwargs) = async_run(queue.get(1))
assert action == "node.updated" assert action == "node.updated"
assert event == vm assert event == vm