Merge remote-tracking branch 'origin/asyncio' into asyncio

Conflicts:
	gns3server/handlers/vpcs_handler.py
	tests/api/test_version.py
	tests/api/test_vpcs.py

tests/api/base.py

@@ -44,6 +44,9 @@ class Query:
     def get(self, path, **kwargs):
         return self._fetch("GET", path, **kwargs)
 
+    def delete(self, path, **kwargs):
+        return self._fetch("DELETE", path, **kwargs)
+
     def _get_url(self, path):
         return "http://{}:{}{}".format(self._host, self._port, path)
 

tests/api/test_version.py

@@ -21,6 +21,7 @@ It's also used for unittest the HTTP implementation.
 """
 
 from tests.utils import asyncio_patch
+from tests.api.base import server, loop, port_manager
 from gns3server.version import __version__
 
 

tests/api/test_vpcs.py

@@ -15,7 +15,9 @@
 # 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 tests.api.base import server, loop
 from tests.utils import asyncio_patch
+from gns3server import modules
 
 
 @asyncio_patch('gns3server.modules.VPCS.create_vm', return_value=84)
@@ -27,17 +29,41 @@ def test_vpcs_create(server):
     assert response.json['id'] == 84
 
 
-def test_vpcs_nio_create(server):
-    response = server.post('/vpcs/42/nio', {
-        'id': 42,
-        'nio': {
-            'type': 'nio_unix',
-            'local_file': '/tmp/test',
-            'remote_file': '/tmp/remote'
+def test_vpcs_nio_create_udp(server):
+    vm = server.post('/vpcs', {'name': 'PC TEST 1'})
+    response = server.post('/vpcs/{}/ports/0/nio'.format(vm.json["id"]), {
+            'type': 'nio_udp',
+            'lport': 4242,
+            'rport': 4343,
+            'rhost': '127.0.0.1'
         },
-        'port': 0,
-        'port_id': 0},
         example=True)
     assert response.status == 200
-    assert response.route == '/vpcs/{id}/nio'
-    assert response.json['name'] == 'PC 2'
+    assert response.route == '/vpcs/{id}/ports/{port_id}/nio'
+    assert response.json['type'] == 'nio_udp'
+
+@patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=True)
+def test_vpcs_nio_create_tap(mock, server):
+    vm = server.post('/vpcs', {'name': 'PC TEST 1'})
+    response = server.post('/vpcs/{}/ports/0/nio'.format(vm.json["id"]), {
+            'type': 'nio_tap',
+            'tap_device': 'test',
+        })
+    assert response.status == 200
+    assert response.route == '/vpcs/{vpcs_id}/ports/{port_id}/nio'
+    assert response.json['type'] == 'nio_tap'
+
+def test_vpcs_delete_nio(server):
+    vm = server.post('/vpcs', {'name': 'PC TEST 1'})
+    response = server.post('/vpcs/{}/ports/0/nio'.format(vm.json["id"]), {
+            'type': 'nio_udp',
+            'lport': 4242,
+            'rport': 4343,
+            'rhost': '127.0.0.1'
+        },
+        )
+    response = server.delete('/vpcs/{}/ports/0/nio'.format(vm.json["vpcs_id"]), example=True)
+    assert response.status == 200
+    assert response.route == '/vpcs/{id}/ports/{port_id}/nio'
+
+

tests/modules/vpcs/test_vpcs_device.py

@@ -28,36 +28,64 @@ 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, port_manager):
-    vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir), path="/bin/test")
+    vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
     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, port_manager):
     with pytest.raises(VPCSError):
-        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir), path="/bin/test")
+        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
         assert vm.name == "test"
         assert vm.id == 42
 
+@patch("gns3server.config.Config.get_section_config", return_value = {"path": "/bin/test_fake"})
 def test_vm_invalid_vpcs_path(tmpdir, port_manager):
     with pytest.raises(VPCSError):
-        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir), path="/bin/test_fake")
+        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
         assert vm.name == "test"
         assert vm.id == 42
 
 def test_start(tmpdir, loop, port_manager):
     with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
-        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir), path="/bin/test")
+        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
+        nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+
         loop.run_until_complete(asyncio.async(vm.start()))
         assert vm.is_running() == True
 
 def test_stop(tmpdir, loop, port_manager):
     process = MagicMock()
     with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
-        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir), path="/bin/test")
+        vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
+        nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+
         loop.run_until_complete(asyncio.async(vm.start()))
         assert vm.is_running() == True
         loop.run_until_complete(asyncio.async(vm.stop()))
         assert vm.is_running() == False
         process.terminate.assert_called_with()
 
+def test_add_nio_binding_udp(port_manager, tmpdir):
+    vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
+    nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+    assert nio.lport == 4242
+
+def test_add_nio_binding_tap(port_manager, tmpdir):
+    vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
+    with patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=True):
+        nio = vm.port_add_nio_binding(0, {"type": "nio_tap", "tap_device": "test"})
+        assert nio.tap_device == "test"
+
+def test_add_nio_binding_tap_no_privileged_access(port_manager, tmpdir):
+    vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
+    with patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=False):
+        with pytest.raises(VPCSError):
+            vm.port_add_nio_binding(0, {"type": "nio_tap", "tap_device": "test"})
+    assert vm._ethernet_adapter.ports[0] is not None
+
+def test_port_remove_nio_binding(port_manager, tmpdir):
+    vm = VPCSDevice("test", 42, port_manager, working_dir=str(tmpdir))
+    nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+    vm.port_remove_nio_binding(0)
+    assert vm._ethernet_adapter.ports[0] == None