mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-11-03 05:51:15 +02:00
Move old test to old_tests directory
This commit is contained in:
Julien Duponchelle committed
1 parent
efad58a2af
commit
aab944fb6c
31 files changed
+13
-10
No files matched your search
@@ -51,7 +51,7 @@ def test_version_invalid_input_schema(server):
|
||||
assert response.status == 400
|
||||
|
||||
|
||||
@asyncio_patch("demoserver.handlers.version_handler.VersionHandler", return_value={})
|
||||
@asyncio_patch("gns3server.handlers.version_handler.VersionHandler", return_value={})
|
||||
def test_version_invalid_output_schema():
|
||||
query = {'version': "0.4.2"}
|
||||
response = server.post('/version', query)
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
/c3725.image
|
||||
@@ -1,29 +0,0 @@
|
||||
from gns3server.modules.dynamips import HypervisorManager
|
||||
import pytest
|
||||
import os
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def hypervisor(request):
|
||||
|
||||
dynamips_path = '/usr/bin/dynamips'
|
||||
print("\nStarting Dynamips Hypervisor: {}".format(dynamips_path))
|
||||
manager = HypervisorManager(dynamips_path, "/tmp", "127.0.0.1")
|
||||
hypervisor = manager.start_new_hypervisor()
|
||||
|
||||
def stop():
|
||||
print("\nStopping Dynamips Hypervisor")
|
||||
manager.stop_all_hypervisors()
|
||||
|
||||
request.addfinalizer(stop)
|
||||
return hypervisor
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def image(request):
|
||||
|
||||
cwd = os.path.dirname(os.path.abspath(__file__))
|
||||
image_path = os.path.join(cwd, "c3725.image")
|
||||
if not os.path.isfile(image_path):
|
||||
return None
|
||||
return image_path
|
||||
Binary file not shown.
@@ -1,62 +0,0 @@
|
||||
from gns3server.modules.dynamips import ATMBridge
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def atm_bridge(request, hypervisor):
|
||||
|
||||
atm_bridge = ATMBridge(hypervisor, "ATM bridge")
|
||||
request.addfinalizer(atm_bridge.delete)
|
||||
return atm_bridge
|
||||
|
||||
|
||||
def test_atm_bridge_exists(atm_bridge):
|
||||
|
||||
assert atm_bridge.list()
|
||||
|
||||
|
||||
def test_rename_atm_bridge(atm_bridge):
|
||||
|
||||
atm_bridge.name = "new ATM bridge"
|
||||
assert atm_bridge.name == "new ATM bridge"
|
||||
|
||||
|
||||
def test_add_remove_nio(atm_bridge):
|
||||
|
||||
nio = NIO_Null(atm_bridge.hypervisor)
|
||||
atm_bridge.add_nio(nio, 0) # add NIO on port 0
|
||||
assert atm_bridge.nios
|
||||
atm_bridge.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_add_nio_already_allocated_port(atm_bridge):
|
||||
|
||||
nio = NIO_Null(atm_bridge.hypervisor)
|
||||
atm_bridge.add_nio(nio, 0) # add NIO on port 0
|
||||
with pytest.raises(DynamipsError):
|
||||
atm_bridge.add_nio(nio, 0)
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_remove_nio_non_allocated_port(atm_bridge):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
atm_bridge.remove_nio(0) # remove NIO from port 0
|
||||
|
||||
|
||||
def test_bridge(atm_bridge):
|
||||
|
||||
nio1 = NIO_Null(atm_bridge.hypervisor)
|
||||
atm_bridge.add_nio(nio1, 0) # add NIO on port 0 (Ethernet NIO)
|
||||
nio2 = NIO_Null(atm_bridge.hypervisor)
|
||||
atm_bridge.add_nio(nio1, 1) # add NIO on port 1 (ATM NIO)
|
||||
atm_bridge.configure(0, 1, 10, 10) # configure Ethernet port 0 -> ATM port 1 with VC 10:10
|
||||
assert atm_bridge.mapping[0] == (1, 10, 10)
|
||||
atm_bridge.unconfigure()
|
||||
atm_bridge.remove_nio(0)
|
||||
atm_bridge.remove_nio(1)
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
@@ -1,83 +0,0 @@
|
||||
from gns3server.modules.dynamips import ATMSwitch
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def atmsw(request, hypervisor):
|
||||
|
||||
atmsw = ATMSwitch(hypervisor, "ATM switch")
|
||||
request.addfinalizer(atmsw.delete)
|
||||
return atmsw
|
||||
|
||||
|
||||
def test_atmsw_exists(atmsw):
|
||||
|
||||
assert atmsw.list()
|
||||
|
||||
|
||||
def test_rename_atmsw(atmsw):
|
||||
|
||||
atmsw.name = "new ATM switch"
|
||||
assert atmsw.name == "new ATM switch"
|
||||
|
||||
|
||||
def test_add_remove_nio(atmsw):
|
||||
|
||||
nio = NIO_Null(atmsw.hypervisor)
|
||||
atmsw.add_nio(nio, 0) # add NIO on port 0
|
||||
assert atmsw.nios
|
||||
atmsw.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_add_nio_already_allocated_port(atmsw):
|
||||
|
||||
nio = NIO_Null(atmsw.hypervisor)
|
||||
atmsw.add_nio(nio, 0) # add NIO on port 0
|
||||
with pytest.raises(DynamipsError):
|
||||
atmsw.add_nio(nio, 0)
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_remove_nio_non_allocated_port(atmsw):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
atmsw.remove_nio(0) # remove NIO from port 0
|
||||
|
||||
|
||||
def test_vp(atmsw):
|
||||
|
||||
nio1 = NIO_Null(atmsw.hypervisor)
|
||||
atmsw.add_nio(nio1, 0) # add NIO on port 0
|
||||
nio2 = NIO_Null(atmsw.hypervisor)
|
||||
atmsw.add_nio(nio1, 1) # add NIO on port 1
|
||||
atmsw.map_vp(0, 10, 1, 20) # port 0 VP 10 to port 1 VP 20 (unidirectional)
|
||||
atmsw.map_vp(1, 20, 0, 10) # port 1 VP 20 to port 0 VP 10 (unidirectional)
|
||||
assert atmsw.mapping[(0, 10)] == (1, 20)
|
||||
assert atmsw.mapping[(1, 20)] == (0, 10)
|
||||
atmsw.unmap_vp(0, 10, 1, 20) # port 0 VP 10 to port 1 VP 20 (unidirectional)
|
||||
atmsw.unmap_vp(1, 20, 0, 10) # port 1 VP 20 to port 0 VP 10 (unidirectional)
|
||||
atmsw.remove_nio(0)
|
||||
atmsw.remove_nio(1)
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
|
||||
|
||||
def test_pvc(atmsw):
|
||||
|
||||
nio1 = NIO_Null(atmsw.hypervisor)
|
||||
atmsw.add_nio(nio1, 0) # add NIO on port 0
|
||||
nio2 = NIO_Null(atmsw.hypervisor)
|
||||
atmsw.add_nio(nio1, 1) # add NIO on port 1
|
||||
atmsw.map_pvc(0, 10, 10, 1, 20, 20) # port 0 VC 10:10 to port 1 VP 20:20 (unidirectional)
|
||||
atmsw.map_pvc(1, 20, 20, 0, 10, 10) # port 1 VC 20:20 to port 0 VC 10:10 (unidirectional)
|
||||
assert atmsw.mapping[(0, 10, 10)] == (1, 20, 20)
|
||||
assert atmsw.mapping[(1, 20, 20)] == (0, 10, 10)
|
||||
atmsw.unmap_pvc(0, 10, 10, 1, 20, 20) # port 0 VC 10:10 to port 1 VP 20:20 (unidirectional)
|
||||
atmsw.unmap_pvc(1, 20, 20, 0, 10, 10) # port 1 VC 20:20 to port 0 VC 10:10 (unidirectional)
|
||||
atmsw.remove_nio(0)
|
||||
atmsw.remove_nio(1)
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
@@ -1,31 +0,0 @@
|
||||
from gns3server.modules.dynamips import Bridge
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bridge(request, hypervisor):
|
||||
|
||||
bridge = Bridge(hypervisor, "bridge")
|
||||
request.addfinalizer(bridge.delete)
|
||||
return bridge
|
||||
|
||||
|
||||
def test_bridge_exists(bridge):
|
||||
|
||||
assert bridge.list()
|
||||
|
||||
|
||||
def test_rename_bridge(bridge):
|
||||
|
||||
bridge.name = "new bridge"
|
||||
assert bridge.name == "new bridge"
|
||||
|
||||
|
||||
def test_add_remove_nio(bridge):
|
||||
|
||||
nio = NIO_Null(bridge.hypervisor)
|
||||
bridge.add_nio(nio)
|
||||
assert bridge.nios
|
||||
bridge.remove_nio(nio)
|
||||
nio.delete()
|
||||
@@ -1,167 +0,0 @@
|
||||
from gns3server.modules.dynamips import C1700
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
from gns3server.modules.dynamips import WIC_2T
|
||||
from gns3server.modules.dynamips import WIC_1ENET
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router_c1700(request, hypervisor):
|
||||
|
||||
router = C1700(hypervisor, "c1700 router")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_router_exists(router_c1700):
|
||||
|
||||
assert router_c1700.platform == "c1700"
|
||||
assert router_c1700.list()
|
||||
|
||||
|
||||
def test_chassis_1721(hypervisor):
|
||||
|
||||
router = C1700(hypervisor, "1721 chassis", chassis="1721")
|
||||
assert router.chassis == "1721"
|
||||
assert str(router.slots[0]) == "C1700-MB-1FE"
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_1721(router_c1700):
|
||||
|
||||
assert router_c1700.chassis == "1720" # default chassis
|
||||
router_c1700.chassis = "1721"
|
||||
assert router_c1700.chassis == "1721"
|
||||
|
||||
|
||||
def test_chassis_1750(hypervisor):
|
||||
|
||||
router = C1700(hypervisor, "1750 chassis", chassis="1750")
|
||||
assert router.chassis == "1750"
|
||||
assert str(router.slots[0]) == "C1700-MB-1FE"
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_1750(router_c1700):
|
||||
|
||||
assert router_c1700.chassis == "1720" # default chassis
|
||||
router_c1700.chassis = "1750"
|
||||
assert router_c1700.chassis == "1750"
|
||||
|
||||
|
||||
def test_chassis_1751(hypervisor):
|
||||
|
||||
router = C1700(hypervisor, "1751 chassis", chassis="1751")
|
||||
assert router.chassis == "1751"
|
||||
assert str(router.slots[0]) == "C1700-MB-1FE"
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_1751(router_c1700):
|
||||
|
||||
assert router_c1700.chassis == "1720" # default chassis
|
||||
router_c1700.chassis = "1751"
|
||||
assert router_c1700.chassis == "1751"
|
||||
|
||||
|
||||
def test_chassis_1760(hypervisor):
|
||||
|
||||
router = C1700(hypervisor, "1760 chassis", chassis="1760")
|
||||
assert router.chassis == "1760"
|
||||
assert str(router.slots[0]) == "C1700-MB-1FE"
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_1760(router_c1700):
|
||||
|
||||
assert router_c1700.chassis == "1720" # default chassis
|
||||
router_c1700.chassis = "1760"
|
||||
assert router_c1700.chassis == "1760"
|
||||
|
||||
|
||||
def test_iomem(router_c1700):
|
||||
|
||||
assert router_c1700.iomem == 15 # default value
|
||||
router_c1700.iomem = 20
|
||||
assert router_c1700.iomem == 20
|
||||
|
||||
|
||||
def test_mac_addr(router_c1700):
|
||||
|
||||
assert router_c1700.mac_addr != None
|
||||
router_c1700.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router_c1700.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router_c1700):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c1700.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router_c1700):
|
||||
|
||||
assert router_c1700.system_id == "FTX0945W0MY" # default value
|
||||
router_c1700.system_id = "FTX0945W0MO"
|
||||
assert router_c1700.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router_c1700):
|
||||
|
||||
router_c1700.get_hardware_info() # FIXME: Dynamips doesn't return anything
|
||||
|
||||
|
||||
def test_install_remove_wic(router_c1700):
|
||||
|
||||
wic = WIC_2T()
|
||||
router_c1700.install_wic(0, wic) # install in WIC slot 0
|
||||
assert router_c1700.slots[0].wics[0]
|
||||
wic = WIC_1ENET()
|
||||
router_c1700.install_wic(1, wic) # install in WIC slot 1
|
||||
assert router_c1700.slots[0].wics[1]
|
||||
router_c1700.uninstall_wic(0) # uninstall WIC from slot 0
|
||||
assert not router_c1700.slots[0].wics[0]
|
||||
|
||||
|
||||
def test_install_wic_into_wrong_slot(router_c1700):
|
||||
|
||||
wic = WIC_2T()
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c1700.install_wic(2, wic) # install in WIC slot 2
|
||||
|
||||
|
||||
def test_install_wic_into_already_occupied_slot(router_c1700):
|
||||
|
||||
wic = WIC_2T()
|
||||
router_c1700.install_wic(0, wic) # install in WIC slot 0
|
||||
wic = WIC_1ENET()
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c1700.install_wic(0, wic) # install in WIC slot 0
|
||||
|
||||
|
||||
def test_wic_add_remove_nio_binding(router_c1700):
|
||||
|
||||
nio = NIO_Null(router_c1700.hypervisor)
|
||||
wic = WIC_2T()
|
||||
router_c1700.install_wic(0, wic) # install WIC in slot 0
|
||||
router_c1700.slot_add_nio_binding(0, 17, nio) # slot 0/17 (slot 0, wic 0, port 1)
|
||||
assert router_c1700.slots[0].ports[17] == nio
|
||||
assert router_c1700.get_slot_nio_bindings(slot_id=0)
|
||||
router_c1700.slot_remove_nio_binding(0, 17) # slot 0/17 (slot 0, wic 0, port 1)
|
||||
assert not router_c1700.get_slot_nio_bindings(slot_id=0)
|
||||
assert not router_c1700.slots[0].ports[17] == nio
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_wic_add_remove_nio_binding_for_chassis_1760(hypervisor):
|
||||
|
||||
router = C1700(hypervisor, "1760 chassis", chassis="1760")
|
||||
nio = NIO_Null(router.hypervisor)
|
||||
wic = WIC_2T()
|
||||
router.install_wic(1, wic) # install WIC in slot 1
|
||||
router.slot_add_nio_binding(0, 32, nio) # slot 0/17 (slot 0, wic 1, port 0)
|
||||
router.slot_remove_nio_binding(0, 32)
|
||||
assert not router.get_slot_nio_bindings(slot_id=0)
|
||||
nio.delete()
|
||||
router.delete()
|
||||
@@ -1,216 +0,0 @@
|
||||
from gns3server.modules.dynamips import C2600
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
from gns3server.modules.dynamips import NM_1E
|
||||
from gns3server.modules.dynamips import NM_4E
|
||||
from gns3server.modules.dynamips import NM_1FE_TX
|
||||
from gns3server.modules.dynamips import NM_16ESW
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router_c2600(request, hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "c2600 router")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_router_exists(router_c2600):
|
||||
|
||||
assert router_c2600.platform == "c2600"
|
||||
assert router_c2600.list()
|
||||
|
||||
|
||||
def test_chassis_2611(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2611 chassis", chassis="2611")
|
||||
assert router.chassis == "2611"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2611"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2611(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2611"
|
||||
assert router_c2600.chassis == "2611"
|
||||
|
||||
|
||||
def test_chassis_2620(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2620 chassis", chassis="2620")
|
||||
assert router.chassis == "2620"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2620"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2620(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2620"
|
||||
assert router_c2600.chassis == "2620"
|
||||
|
||||
|
||||
def test_chassis_2621(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2621 chassis", chassis="2621")
|
||||
assert router.chassis == "2621"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2621"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2621(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2621"
|
||||
assert router_c2600.chassis == "2621"
|
||||
|
||||
|
||||
def test_chassis_2610XM(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2610XM chassis", chassis="2610XM")
|
||||
assert router.chassis == "2610XM"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2610XM"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2610XM(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2610XM"
|
||||
assert router_c2600.chassis == "2610XM"
|
||||
|
||||
|
||||
def test_chassis_2611XM(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2611XM chassis", chassis="2611XM")
|
||||
assert router.chassis == "2611XM"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2611XM"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2611XM(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2611XM"
|
||||
assert router_c2600.chassis == "2611XM"
|
||||
|
||||
|
||||
def test_chassis_2620XM(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2620XM chassis", chassis="2620XM")
|
||||
assert router.chassis == "2620XM"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2620XM"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2620XM(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2620XM"
|
||||
assert router_c2600.chassis == "2620XM"
|
||||
|
||||
|
||||
def test_chassis_2621XM(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2621XM chassis", chassis="2621XM")
|
||||
assert router.chassis == "2621XM"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2621XM"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2621XM(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2621XM"
|
||||
assert router_c2600.chassis == "2621XM"
|
||||
|
||||
|
||||
def test_chassis_2650XM(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2650XM chassis", chassis="2650XM")
|
||||
assert router.chassis == "2650XM"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2650XM"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2650XM(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2650XM"
|
||||
assert router_c2600.chassis == "2650XM"
|
||||
|
||||
|
||||
def test_chassis_2651XM(hypervisor):
|
||||
|
||||
router = C2600(hypervisor, "2651XM chassis", chassis="2651XM")
|
||||
assert router.chassis == "2651XM"
|
||||
assert isinstance(router.slots[0], router.integrated_adapters["2651XM"])
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_2651XM(router_c2600):
|
||||
|
||||
assert router_c2600.chassis == "2610" # default chassis
|
||||
router_c2600.chassis = "2651XM"
|
||||
assert router_c2600.chassis == "2651XM"
|
||||
|
||||
|
||||
def test_iomem(router_c2600):
|
||||
|
||||
assert router_c2600.iomem == 15 # default value
|
||||
router_c2600.iomem = 20
|
||||
assert router_c2600.iomem == 20
|
||||
|
||||
|
||||
def test_mac_addr(router_c2600):
|
||||
|
||||
assert router_c2600.mac_addr != None
|
||||
router_c2600.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router_c2600.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router_c2600):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c2600.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router_c2600):
|
||||
|
||||
assert router_c2600.system_id == "FTX0945W0MY" # default value
|
||||
router_c2600.system_id = "FTX0945W0MO"
|
||||
assert router_c2600.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router_c2600):
|
||||
|
||||
router_c2600.get_hardware_info() # FIXME: Dynamips doesn't return anything
|
||||
|
||||
|
||||
def test_slot_add_NM_1E(router_c2600):
|
||||
|
||||
adapter = NM_1E()
|
||||
router_c2600.slot_add_binding(1, adapter)
|
||||
assert router_c2600.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_4E(router_c2600):
|
||||
|
||||
adapter = NM_4E()
|
||||
router_c2600.slot_add_binding(1, adapter)
|
||||
assert router_c2600.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_1FE_TX(router_c2600):
|
||||
|
||||
adapter = NM_1FE_TX()
|
||||
router_c2600.slot_add_binding(1, adapter)
|
||||
assert router_c2600.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_16ESW(router_c2600):
|
||||
|
||||
adapter = NM_16ESW()
|
||||
router_c2600.slot_add_binding(1, adapter)
|
||||
assert router_c2600.slots[1] == adapter
|
||||
@@ -1,73 +0,0 @@
|
||||
from gns3server.modules.dynamips import C2691
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
from gns3server.modules.dynamips import NM_1FE_TX
|
||||
from gns3server.modules.dynamips import NM_4T
|
||||
from gns3server.modules.dynamips import NM_16ESW
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router_c2691(request, hypervisor):
|
||||
|
||||
router = C2691(hypervisor, "c2691 router")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_router_exists(router_c2691):
|
||||
|
||||
assert router_c2691.platform == "c2691"
|
||||
assert router_c2691.list()
|
||||
|
||||
|
||||
def test_iomem(router_c2691):
|
||||
|
||||
assert router_c2691.iomem == 5 # default value
|
||||
router_c2691.iomem = 10
|
||||
assert router_c2691.iomem == 10
|
||||
|
||||
|
||||
def test_mac_addr(router_c2691):
|
||||
|
||||
assert router_c2691.mac_addr != None
|
||||
router_c2691.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router_c2691.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router_c2691):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c2691.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router_c2691):
|
||||
|
||||
assert router_c2691.system_id == "FTX0945W0MY" # default value
|
||||
router_c2691.system_id = "FTX0945W0MO"
|
||||
assert router_c2691.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router_c2691):
|
||||
|
||||
router_c2691.get_hardware_info() # FIXME: Dynamips doesn't return anything
|
||||
|
||||
|
||||
def test_slot_add_NM_1FE_TX(router_c2691):
|
||||
|
||||
adapter = NM_1FE_TX()
|
||||
router_c2691.slot_add_binding(1, adapter)
|
||||
assert router_c2691.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_4T(router_c2691):
|
||||
|
||||
adapter = NM_4T()
|
||||
router_c2691.slot_add_binding(1, adapter)
|
||||
assert router_c2691.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_16ESW(router_c2691):
|
||||
|
||||
adapter = NM_16ESW()
|
||||
router_c2691.slot_add_binding(1, adapter)
|
||||
assert router_c2691.slots[1] == adapter
|
||||
@@ -1,118 +0,0 @@
|
||||
from gns3server.modules.dynamips import C3600
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
from gns3server.modules.dynamips import NM_1E
|
||||
from gns3server.modules.dynamips import NM_4E
|
||||
from gns3server.modules.dynamips import NM_1FE_TX
|
||||
from gns3server.modules.dynamips import NM_16ESW
|
||||
from gns3server.modules.dynamips import NM_4T
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router_c3600(request, hypervisor):
|
||||
|
||||
router = C3600(hypervisor, "c3600 router")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_router_exist(router_c3600):
|
||||
|
||||
assert router_c3600.platform == "c3600"
|
||||
assert router_c3600.list()
|
||||
|
||||
|
||||
def test_chassis_3620(hypervisor):
|
||||
|
||||
router = C3600(hypervisor, "3620 chassis", chassis="3620")
|
||||
assert router.chassis == "3620"
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_3620(router_c3600):
|
||||
|
||||
assert router_c3600.chassis == "3640" # default chassis
|
||||
router_c3600.chassis = "3620"
|
||||
assert router_c3600.chassis == "3620"
|
||||
|
||||
|
||||
def test_chassis_3660(hypervisor):
|
||||
|
||||
router = C3600(hypervisor, "3660 chassis", chassis="3660")
|
||||
assert router.chassis == "3660"
|
||||
assert str(router.slots[0]) == "Leopard-2FE"
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_chassis_change_to_3660(router_c3600):
|
||||
|
||||
assert router_c3600.chassis == "3640" # default chassis
|
||||
router_c3600.chassis = "3660"
|
||||
assert router_c3600.chassis == "3660"
|
||||
|
||||
|
||||
def test_iomem(router_c3600):
|
||||
|
||||
assert router_c3600.iomem == 5 # default value
|
||||
router_c3600.iomem = 10
|
||||
assert router_c3600.iomem == 10
|
||||
|
||||
|
||||
def test_mac_addr(router_c3600):
|
||||
|
||||
assert router_c3600.mac_addr != None
|
||||
router_c3600.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router_c3600.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router_c3600):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c3600.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router_c3600):
|
||||
|
||||
assert router_c3600.system_id == "FTX0945W0MY" # default value
|
||||
router_c3600.system_id = "FTX0945W0MO"
|
||||
assert router_c3600.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router_c3600):
|
||||
|
||||
router_c3600.get_hardware_info() # FIXME: Dynamips doesn't return anything
|
||||
|
||||
|
||||
def test_slot_add_NM_1E(router_c3600):
|
||||
|
||||
adapter = NM_1E()
|
||||
router_c3600.slot_add_binding(1, adapter)
|
||||
assert router_c3600.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_4E(router_c3600):
|
||||
|
||||
adapter = NM_4E()
|
||||
router_c3600.slot_add_binding(1, adapter)
|
||||
assert router_c3600.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_1FE_TX(router_c3600):
|
||||
|
||||
adapter = NM_1FE_TX()
|
||||
router_c3600.slot_add_binding(1, adapter)
|
||||
assert router_c3600.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_16ESW(router_c3600):
|
||||
|
||||
adapter = NM_16ESW()
|
||||
router_c3600.slot_add_binding(1, adapter)
|
||||
assert router_c3600.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_4T(router_c3600):
|
||||
|
||||
adapter = NM_4T()
|
||||
router_c3600.slot_add_binding(1, adapter)
|
||||
assert router_c3600.slots[1] == adapter
|
||||
@@ -1,73 +0,0 @@
|
||||
from gns3server.modules.dynamips import C3725
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
from gns3server.modules.dynamips import NM_1FE_TX
|
||||
from gns3server.modules.dynamips import NM_4T
|
||||
from gns3server.modules.dynamips import NM_16ESW
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router_c3725(request, hypervisor):
|
||||
|
||||
router = C3725(hypervisor, "c3725 router")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_router_exists(router_c3725):
|
||||
|
||||
assert router_c3725.platform == "c3725"
|
||||
assert router_c3725.list()
|
||||
|
||||
|
||||
def test_iomem(router_c3725):
|
||||
|
||||
assert router_c3725.iomem == 5 # default value
|
||||
router_c3725.iomem = 10
|
||||
assert router_c3725.iomem == 10
|
||||
|
||||
|
||||
def test_mac_addr(router_c3725):
|
||||
|
||||
assert router_c3725.mac_addr != None
|
||||
router_c3725.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router_c3725.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router_c3725):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c3725.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router_c3725):
|
||||
|
||||
assert router_c3725.system_id == "FTX0945W0MY" # default value
|
||||
router_c3725.system_id = "FTX0945W0MO"
|
||||
assert router_c3725.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router_c3725):
|
||||
|
||||
router_c3725.get_hardware_info() # FIXME: Dynamips doesn't return anything
|
||||
|
||||
|
||||
def test_slot_add_NM_1FE_TX(router_c3725):
|
||||
|
||||
adapter = NM_1FE_TX()
|
||||
router_c3725.slot_add_binding(1, adapter)
|
||||
assert router_c3725.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_4T(router_c3725):
|
||||
|
||||
adapter = NM_4T()
|
||||
router_c3725.slot_add_binding(1, adapter)
|
||||
assert router_c3725.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_16ESW(router_c3725):
|
||||
|
||||
adapter = NM_16ESW()
|
||||
router_c3725.slot_add_binding(1, adapter)
|
||||
assert router_c3725.slots[1] == adapter
|
||||
@@ -1,73 +0,0 @@
|
||||
from gns3server.modules.dynamips import C3745
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
from gns3server.modules.dynamips import NM_1FE_TX
|
||||
from gns3server.modules.dynamips import NM_4T
|
||||
from gns3server.modules.dynamips import NM_16ESW
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router_c3745(request, hypervisor):
|
||||
|
||||
router = C3745(hypervisor, "c3745 router")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_router_exists(router_c3745):
|
||||
|
||||
assert router_c3745.platform == "c3745"
|
||||
assert router_c3745.list()
|
||||
|
||||
|
||||
def test_iomem(router_c3745):
|
||||
|
||||
assert router_c3745.iomem == 5 # default value
|
||||
router_c3745.iomem = 10
|
||||
assert router_c3745.iomem == 10
|
||||
|
||||
|
||||
def test_mac_addr(router_c3745):
|
||||
|
||||
assert router_c3745.mac_addr != None
|
||||
router_c3745.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router_c3745.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router_c3745):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c3745.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router_c3745):
|
||||
|
||||
assert router_c3745.system_id == "FTX0945W0MY" # default value
|
||||
router_c3745.system_id = "FTX0945W0MO"
|
||||
assert router_c3745.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router_c3745):
|
||||
|
||||
router_c3745.get_hardware_info() # FIXME: Dynamips doesn't return anything
|
||||
|
||||
|
||||
def test_slot_add_NM_1FE_TX(router_c3745):
|
||||
|
||||
adapter = NM_1FE_TX()
|
||||
router_c3745.slot_add_binding(1, adapter)
|
||||
assert router_c3745.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_4T(router_c3745):
|
||||
|
||||
adapter = NM_4T()
|
||||
router_c3745.slot_add_binding(1, adapter)
|
||||
assert router_c3745.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_NM_16ESW(router_c3745):
|
||||
|
||||
adapter = NM_16ESW()
|
||||
router_c3745.slot_add_binding(1, adapter)
|
||||
assert router_c3745.slots[1] == adapter
|
||||
@@ -1,188 +0,0 @@
|
||||
from gns3server.modules.dynamips import C7200
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
from gns3server.modules.dynamips import PA_2FE_TX
|
||||
from gns3server.modules.dynamips import PA_4E
|
||||
from gns3server.modules.dynamips import PA_4T
|
||||
from gns3server.modules.dynamips import PA_8E
|
||||
from gns3server.modules.dynamips import PA_8T
|
||||
from gns3server.modules.dynamips import PA_A1
|
||||
from gns3server.modules.dynamips import PA_FE_TX
|
||||
from gns3server.modules.dynamips import PA_GE
|
||||
from gns3server.modules.dynamips import PA_POS_OC3
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router_c7200(request, hypervisor):
|
||||
|
||||
router = C7200(hypervisor, "c7200 router")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_router_exists(router_c7200):
|
||||
|
||||
assert router_c7200.platform == "c7200"
|
||||
assert router_c7200.list()
|
||||
|
||||
|
||||
def test_npe(router_c7200):
|
||||
|
||||
assert router_c7200.npe == "npe-400" # default value
|
||||
router_c7200.npe = "npe-200"
|
||||
assert router_c7200.npe == "npe-200"
|
||||
|
||||
|
||||
def test_midplane(router_c7200):
|
||||
|
||||
assert router_c7200.midplane == "vxr" # default value
|
||||
router_c7200.midplane = "std"
|
||||
assert router_c7200.midplane == "std"
|
||||
|
||||
|
||||
def test_sensors(router_c7200):
|
||||
|
||||
assert router_c7200.sensors == [22, 22, 22, 22] # default values (everything at 22C)
|
||||
router_c7200.sensors = [25, 25, 25, 25]
|
||||
assert router_c7200.sensors == [25, 25, 25, 25]
|
||||
|
||||
|
||||
def test_power_supplies(router_c7200):
|
||||
|
||||
assert router_c7200.power_supplies == [1, 1] # default values (1 = powered on)
|
||||
router_c7200.power_supplies = [0, 0]
|
||||
assert router_c7200.power_supplies == [0, 0]
|
||||
|
||||
|
||||
def test_mac_addr(router_c7200):
|
||||
|
||||
assert router_c7200.mac_addr != None
|
||||
router_c7200.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router_c7200.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router_c7200):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c7200.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router_c7200):
|
||||
|
||||
assert router_c7200.system_id == "FTX0945W0MY" # default value
|
||||
router_c7200.system_id = "FTX0945W0MO"
|
||||
assert router_c7200.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router_c7200):
|
||||
|
||||
router_c7200.get_hardware_info() # FIXME: Dynamips doesn't return anything
|
||||
|
||||
|
||||
def test_slot_add_PA_2FE_TX(router_c7200):
|
||||
|
||||
adapter = PA_2FE_TX()
|
||||
router_c7200.slot_add_binding(1, adapter)
|
||||
assert router_c7200.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_4E(router_c7200):
|
||||
|
||||
adapter = PA_4E()
|
||||
router_c7200.slot_add_binding(2, adapter)
|
||||
assert router_c7200.slots[2] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_4T(router_c7200):
|
||||
|
||||
adapter = PA_4T()
|
||||
router_c7200.slot_add_binding(3, adapter)
|
||||
assert router_c7200.slots[3] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_8E(router_c7200):
|
||||
|
||||
adapter = PA_8E()
|
||||
router_c7200.slot_add_binding(4, adapter)
|
||||
assert router_c7200.slots[4] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_8T(router_c7200):
|
||||
|
||||
adapter = PA_8T()
|
||||
router_c7200.slot_add_binding(5, adapter)
|
||||
assert router_c7200.slots[5] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_A1(router_c7200):
|
||||
|
||||
adapter = PA_A1()
|
||||
router_c7200.slot_add_binding(1, adapter)
|
||||
assert router_c7200.slots[1] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_FE_TX(router_c7200):
|
||||
|
||||
adapter = PA_FE_TX()
|
||||
router_c7200.slot_add_binding(2, adapter)
|
||||
assert router_c7200.slots[2] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_GE(router_c7200):
|
||||
|
||||
adapter = PA_GE()
|
||||
router_c7200.slot_add_binding(3, adapter)
|
||||
assert router_c7200.slots[3] == adapter
|
||||
|
||||
|
||||
def test_slot_add_PA_POS_OC3(router_c7200):
|
||||
|
||||
adapter = PA_POS_OC3()
|
||||
router_c7200.slot_add_binding(4, adapter)
|
||||
assert router_c7200.slots[4] == adapter
|
||||
|
||||
|
||||
def test_slot_add_into_already_occupied_slot(router_c7200):
|
||||
|
||||
adapter = PA_FE_TX()
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c7200.slot_add_binding(0, adapter)
|
||||
|
||||
|
||||
def test_slot_add_into_wrong_slot(router_c7200):
|
||||
|
||||
adapter = PA_FE_TX()
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c7200.slot_add_binding(10, adapter)
|
||||
|
||||
|
||||
def test_slot_remove_adapter(router_c7200):
|
||||
|
||||
adapter = PA_FE_TX()
|
||||
router_c7200.slot_add_binding(1, adapter)
|
||||
router_c7200.slot_remove_binding(1)
|
||||
assert router_c7200.slots[1] == None
|
||||
|
||||
|
||||
def test_slot_add_remove_nio_binding(router_c7200):
|
||||
|
||||
adapter = PA_FE_TX()
|
||||
router_c7200.slot_add_binding(1, adapter)
|
||||
nio = NIO_Null(router_c7200.hypervisor)
|
||||
router_c7200.slot_add_nio_binding(1, 0, nio) # slot 1/0
|
||||
assert router_c7200.get_slot_nio_bindings(slot_id=1)
|
||||
assert router_c7200.slots[1].ports[0] == nio
|
||||
router_c7200.slot_remove_nio_binding(1, 0) # slot 1/0
|
||||
assert not router_c7200.get_slot_nio_bindings(slot_id=0)
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_slot_add_nio_to_wrong_port(router_c7200):
|
||||
|
||||
adapter = PA_FE_TX()
|
||||
router_c7200.slot_add_binding(1, adapter)
|
||||
nio = NIO_Null(router_c7200.hypervisor)
|
||||
with pytest.raises(DynamipsError):
|
||||
router_c7200.slot_add_nio_binding(1, 1, nio) # slot 1/1
|
||||
nio.delete()
|
||||
@@ -1,87 +0,0 @@
|
||||
from gns3server.modules.dynamips import EthernetSwitch
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ethsw(request, hypervisor):
|
||||
|
||||
ethsw = EthernetSwitch(hypervisor, "Ethernet switch")
|
||||
request.addfinalizer(ethsw.delete)
|
||||
return ethsw
|
||||
|
||||
|
||||
def test_ethsw_exists(ethsw):
|
||||
|
||||
assert ethsw.list()
|
||||
|
||||
|
||||
def test_rename_ethsw(ethsw):
|
||||
|
||||
ethsw.name = "new Ethernet switch"
|
||||
assert ethsw.name == "new Ethernet switch"
|
||||
|
||||
|
||||
def test_add_remove_nio(ethsw):
|
||||
|
||||
nio = NIO_Null(ethsw.hypervisor)
|
||||
ethsw.add_nio(nio, 0) # add NIO on port 0
|
||||
assert ethsw.nios
|
||||
ethsw.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_add_nio_already_allocated_port(ethsw):
|
||||
|
||||
nio = NIO_Null(ethsw.hypervisor)
|
||||
ethsw.add_nio(nio, 0) # add NIO on port 0
|
||||
with pytest.raises(DynamipsError):
|
||||
ethsw.add_nio(nio, 0)
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_remove_nio_non_allocated_port(ethsw):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
ethsw.remove_nio(0) # remove NIO from port 0
|
||||
|
||||
|
||||
def test_set_access_port(ethsw):
|
||||
|
||||
nio = NIO_Null(ethsw.hypervisor)
|
||||
ethsw.add_nio(nio, 0) # add NIO on port 0
|
||||
ethsw.set_access_port(0, 10) # set port 0 as access in VLAN 10
|
||||
assert ethsw.mapping[0] == ("access", 10)
|
||||
ethsw.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_set_dot1q_port(ethsw):
|
||||
|
||||
nio = NIO_Null(ethsw.hypervisor)
|
||||
ethsw.add_nio(nio, 0) # add NIO on port 0
|
||||
ethsw.set_dot1q_port(0, 1) # set port 0 as 802.1Q trunk with native VLAN 1
|
||||
assert ethsw.mapping[0] == ("dot1q", 1)
|
||||
ethsw.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_set_qinq_port(ethsw):
|
||||
|
||||
nio = NIO_Null(ethsw.hypervisor)
|
||||
ethsw.add_nio(nio, 0) # add NIO on port 0
|
||||
ethsw.set_qinq_port(0, 100) # set port 0 as QinQ trunk with outer VLAN 100
|
||||
assert ethsw.mapping[0] == ("qinq", 100)
|
||||
ethsw.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_get_mac_addr_table(ethsw):
|
||||
|
||||
assert not ethsw.get_mac_addr_table() # MAC address table should be empty
|
||||
|
||||
|
||||
def test_clear_mac_addr_table(ethsw):
|
||||
|
||||
ethsw.clear_mac_addr_table()
|
||||
@@ -1,65 +0,0 @@
|
||||
from gns3server.modules.dynamips import FrameRelaySwitch
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def frsw(request, hypervisor):
|
||||
|
||||
frsw = FrameRelaySwitch(hypervisor, "Frane Relay switch")
|
||||
request.addfinalizer(frsw.delete)
|
||||
return frsw
|
||||
|
||||
|
||||
def test_frsw_exists(frsw):
|
||||
|
||||
assert frsw.list()
|
||||
|
||||
|
||||
def test_rename_frsw(frsw):
|
||||
|
||||
frsw.name = "new Frame Relay switch"
|
||||
assert frsw.name == "new Frame Relay switch"
|
||||
|
||||
|
||||
def test_add_remove_nio(frsw):
|
||||
|
||||
nio = NIO_Null(frsw.hypervisor)
|
||||
frsw.add_nio(nio, 0) # add NIO on port 0
|
||||
assert frsw.nios
|
||||
frsw.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_add_nio_already_allocated_port(frsw):
|
||||
|
||||
nio = NIO_Null(frsw.hypervisor)
|
||||
frsw.add_nio(nio, 0) # add NIO on port 0
|
||||
with pytest.raises(DynamipsError):
|
||||
frsw.add_nio(nio, 0)
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_remove_nio_non_allocated_port(frsw):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
frsw.remove_nio(0) # remove NIO from port 0
|
||||
|
||||
|
||||
def test_vc(frsw):
|
||||
|
||||
nio1 = NIO_Null(frsw.hypervisor)
|
||||
frsw.add_nio(nio1, 0) # add NIO on port 0
|
||||
nio2 = NIO_Null(frsw.hypervisor)
|
||||
frsw.add_nio(nio1, 1) # add NIO on port 1
|
||||
frsw.map_vc(0, 10, 1, 20) # port 0 DLCI 10 to port 1 DLCI 20 (unidirectional)
|
||||
frsw.map_vc(1, 20, 0, 10) # port 1 DLCI 20 to port 0 DLCI 10 (unidirectional)
|
||||
assert frsw.mapping[(0, 10)] == (1, 20)
|
||||
assert frsw.mapping[(1, 20)] == (0, 10)
|
||||
frsw.unmap_vc(0, 10, 1, 20) # port 0 DLCI 10 to port 1 DLCI 20 (unidirectional)
|
||||
frsw.unmap_vc(1, 20, 0, 10) # port 1 DLCI 20 to port 0 DLCI 10 (unidirectional)
|
||||
frsw.remove_nio(0)
|
||||
frsw.remove_nio(1)
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
@@ -1,25 +0,0 @@
|
||||
from gns3server.modules.dynamips import Hub
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hub(request, hypervisor):
|
||||
|
||||
hub = Hub(hypervisor, "hub")
|
||||
request.addfinalizer(hub.delete)
|
||||
return hub
|
||||
|
||||
|
||||
def test_hub_exists(hub):
|
||||
|
||||
assert hub.list()
|
||||
|
||||
|
||||
def test_add_remove_nio(hub):
|
||||
|
||||
nio = NIO_Null(hub.hypervisor)
|
||||
hub.add_nio(nio, 0) # add NIO to port 0
|
||||
assert hub.mapping[0] == nio
|
||||
hub.remove_nio(0) # remove NIO from port 0
|
||||
nio.delete()
|
||||
@@ -1,41 +0,0 @@
|
||||
from gns3server.modules.dynamips import Hypervisor
|
||||
import time
|
||||
|
||||
|
||||
def test_is_started(hypervisor):
|
||||
|
||||
assert hypervisor.is_running()
|
||||
|
||||
|
||||
def test_port(hypervisor):
|
||||
|
||||
assert hypervisor.port == 7200
|
||||
|
||||
|
||||
def test_host(hypervisor):
|
||||
|
||||
assert hypervisor.host == "0.0.0.0"
|
||||
|
||||
|
||||
def test_working_dir(hypervisor):
|
||||
|
||||
assert hypervisor.working_dir == "/tmp"
|
||||
|
||||
|
||||
def test_path(hypervisor):
|
||||
|
||||
dynamips_path = '/usr/bin/dynamips'
|
||||
assert hypervisor.path == dynamips_path
|
||||
|
||||
|
||||
def test_stdout():
|
||||
|
||||
# try to launch Dynamips on the same port
|
||||
# this will fail so that we can read its stdout/stderr
|
||||
dynamips_path = '/usr/bin/dynamips'
|
||||
hypervisor = Hypervisor(dynamips_path, "/tmp", "127.0.0.1", 7200)
|
||||
hypervisor.start()
|
||||
# give some time for Dynamips to start
|
||||
time.sleep(0.1)
|
||||
output = hypervisor.read_stdout()
|
||||
assert output
|
||||
@@ -1,52 +0,0 @@
|
||||
from gns3server.modules.dynamips import Router
|
||||
from gns3server.modules.dynamips import HypervisorManager
|
||||
import pytest
|
||||
import os
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def hypervisor_manager(request):
|
||||
|
||||
dynamips_path = '/usr/bin/dynamips'
|
||||
print("\nStarting Dynamips Hypervisor: {}".format(dynamips_path))
|
||||
manager = HypervisorManager(dynamips_path, "/tmp", "127.0.0.1")
|
||||
|
||||
#manager.start_new_hypervisor()
|
||||
|
||||
def stop():
|
||||
print("\nStopping Dynamips Hypervisor")
|
||||
manager.stop_all_hypervisors()
|
||||
|
||||
request.addfinalizer(stop)
|
||||
return manager
|
||||
|
||||
|
||||
def test_allocate_hypervisor_for_router(hypervisor_manager):
|
||||
|
||||
hypervisor_manager.allocate_hypervisor_per_device = False
|
||||
# default of 1GB of RAM per hypervisor instance
|
||||
assert hypervisor_manager.memory_usage_limit_per_hypervisor == 1024
|
||||
hypervisor = hypervisor_manager.allocate_hypervisor_for_router("c3725.image", 512)
|
||||
assert hypervisor.is_running()
|
||||
hypervisor = hypervisor_manager.allocate_hypervisor_for_router("c3725.image", 256)
|
||||
assert hypervisor.memory_load == 768
|
||||
hypervisor = hypervisor_manager.allocate_hypervisor_for_router("c3725.image", 512)
|
||||
assert hypervisor.memory_load == 512
|
||||
assert len(hypervisor_manager.hypervisors) == 2
|
||||
|
||||
|
||||
def test_unallocate_hypervisor_for_router(hypervisor_manager):
|
||||
|
||||
assert len(hypervisor_manager.hypervisors) == 2
|
||||
hypervisor = hypervisor_manager.hypervisors[0]
|
||||
assert hypervisor.memory_load == 768
|
||||
router = Router(hypervisor, "router", "c3725") # default is 128MB of RAM
|
||||
hypervisor_manager.unallocate_hypervisor_for_router(router)
|
||||
assert hypervisor.memory_load == 640
|
||||
hypervisor.decrease_memory_load(512) # forces memory load down to 128
|
||||
assert hypervisor.memory_load == 128
|
||||
router.delete()
|
||||
hypervisor_manager.unallocate_hypervisor_for_router(router)
|
||||
# router is deleted and memory load to 0 now, one hypervisor must
|
||||
# have been shutdown
|
||||
assert len(hypervisor_manager.hypervisors) == 1
|
||||
@@ -1,139 +0,0 @@
|
||||
from gns3server.modules.dynamips import NIO_UDP
|
||||
from gns3server.modules.dynamips import NIO_UDP_auto
|
||||
from gns3server.modules.dynamips import NIO_FIFO
|
||||
from gns3server.modules.dynamips import NIO_Mcast
|
||||
from gns3server.modules.dynamips import NIO_Null
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
import pytest
|
||||
|
||||
# TODO: test UNIX, TAP, VDE, generic Ethernet and Linux Ethernet NIOs
|
||||
|
||||
|
||||
def test_nio_udp(hypervisor):
|
||||
|
||||
nio1 = NIO_UDP(hypervisor, 10001, "127.0.0.1", 10002)
|
||||
assert nio1.lport == 10001
|
||||
nio2 = NIO_UDP(hypervisor, 10002, "127.0.0.1", 10001)
|
||||
assert nio2.lport == 10002
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
|
||||
|
||||
def test_nio_udp_auto(hypervisor):
|
||||
|
||||
nio1 = NIO_UDP_auto(hypervisor, "127.0.0.1", 10001, 10010)
|
||||
assert nio1.lport == 10001
|
||||
nio2 = NIO_UDP_auto(hypervisor, "127.0.0.1", 10001, 10010)
|
||||
assert nio2.lport == 10002
|
||||
nio1.connect("127.0.0.1", nio2.lport)
|
||||
nio2.connect("127.0.0.1", nio1.lport)
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
|
||||
|
||||
def test_nio_fifo(hypervisor):
|
||||
|
||||
nio1 = NIO_FIFO(hypervisor)
|
||||
nio2 = NIO_FIFO(hypervisor)
|
||||
nio1.crossconnect(nio2)
|
||||
assert nio1.list()
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
|
||||
|
||||
def test_nio_mcast(hypervisor):
|
||||
|
||||
nio1 = NIO_Mcast(hypervisor, "232.0.0.1", 10001)
|
||||
assert nio1.group == "232.0.0.1"
|
||||
assert nio1.port == 10001
|
||||
nio1.ttl = 254
|
||||
assert nio1.ttl == 254
|
||||
nio2 = NIO_UDP(hypervisor, 10002, "232.0.0.1", 10001)
|
||||
nio1.delete()
|
||||
nio2.delete()
|
||||
|
||||
|
||||
def test_nio_null(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
assert nio.list()
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_rename_nio(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
assert nio.name.startswith("nio_null")
|
||||
nio.rename("test")
|
||||
assert nio.name == "test"
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_debug_nio(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
nio.debug(1)
|
||||
nio.debug(0)
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_bind_unbind_filter(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
nio.bind_filter("both", "freq_drop")
|
||||
assert nio.input_filter == ("freq_drop", None)
|
||||
assert nio.output_filter == ("freq_drop", None)
|
||||
nio.unbind_filter("both")
|
||||
nio.bind_filter("in", "capture")
|
||||
assert nio.input_filter == ("capture", None)
|
||||
nio.unbind_filter("in")
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_bind_unknown_filter(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
with pytest.raises(DynamipsError):
|
||||
nio.bind_filter("both", "my_filter")
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_unbind_with_no_filter_applied(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
with pytest.raises(DynamipsError):
|
||||
nio.unbind_filter("out")
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_setup_filter(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
nio.bind_filter("in", "freq_drop")
|
||||
nio.setup_filter("in", "5") # drop every 5th packet
|
||||
assert nio.input_filter == ("freq_drop", "5")
|
||||
nio.unbind_filter("in")
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_get_stats(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
assert nio.get_stats() == "0 0 0 0" # nothing has been transmitted or received
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_reset_stats(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
nio.reset_stats()
|
||||
nio.delete()
|
||||
|
||||
|
||||
def test_set_bandwidth(hypervisor):
|
||||
|
||||
nio = NIO_Null(hypervisor)
|
||||
assert nio.bandwidth == None # no constraint by default
|
||||
nio.set_bandwidth(1000) # bandwidth = 1000 Kb/s
|
||||
assert nio.bandwidth == 1000
|
||||
nio.delete()
|
||||
@@ -1,232 +0,0 @@
|
||||
from gns3server.modules.dynamips import Router
|
||||
from gns3server.modules.dynamips import DynamipsError
|
||||
import sys
|
||||
import pytest
|
||||
import tempfile
|
||||
import base64
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def router(request, hypervisor):
|
||||
|
||||
router = Router(hypervisor, "router", platform="c3725")
|
||||
request.addfinalizer(router.delete)
|
||||
return router
|
||||
|
||||
|
||||
def test_hypervisor_is_started(hypervisor):
|
||||
|
||||
assert hypervisor.is_running()
|
||||
|
||||
|
||||
def test_create_and_delete_router(hypervisor):
|
||||
|
||||
router = Router(hypervisor, "test my router")
|
||||
assert router.id >= 0
|
||||
assert router.name == "test my router"
|
||||
assert router.platform == "c7200" # default platform
|
||||
assert not router.is_running()
|
||||
router.delete()
|
||||
with pytest.raises(DynamipsError):
|
||||
router.get_status()
|
||||
|
||||
|
||||
def test_rename_router(hypervisor):
|
||||
|
||||
router = Router(hypervisor, "my router to rename")
|
||||
assert router.name == "my router to rename"
|
||||
router.name = "my_router"
|
||||
assert router.name == "my_router"
|
||||
router.delete()
|
||||
|
||||
|
||||
def test_image(router):
|
||||
|
||||
# let's pretend this file is an IOS image
|
||||
with tempfile.NamedTemporaryFile() as ios_image:
|
||||
router.image = ios_image.name
|
||||
assert router.image == ios_image.name
|
||||
|
||||
|
||||
def test_set_config(router):
|
||||
|
||||
with tempfile.NamedTemporaryFile() as startup_config:
|
||||
startup_config.write(b"hostname test_config\n")
|
||||
router.set_config(startup_config.name)
|
||||
|
||||
|
||||
def test_push_config(router):
|
||||
|
||||
startup_config = base64.b64encode(b"hostname test_config\n").decode("utf-8")
|
||||
private_config = base64.b64encode(b"private config\n").decode("utf-8")
|
||||
router.push_config(startup_config, private_config)
|
||||
router_startup_config, router_private_config = router.extract_config()
|
||||
assert startup_config == router_startup_config
|
||||
assert private_config == router_private_config
|
||||
|
||||
|
||||
def test_status(router, image):
|
||||
# don't test if we have no IOS image
|
||||
if not image:
|
||||
return
|
||||
|
||||
assert router.get_status() == "inactive"
|
||||
router.ram = 256
|
||||
router.image = image
|
||||
router.start()
|
||||
assert router.is_running()
|
||||
router.suspend()
|
||||
assert router.get_status() == "suspended"
|
||||
router.resume()
|
||||
assert router.is_running()
|
||||
router.stop()
|
||||
assert router.get_status() == "inactive"
|
||||
|
||||
|
||||
def test_ram(router):
|
||||
|
||||
assert router.ram == 128 # default ram
|
||||
router.ram = 256
|
||||
assert router.ram == 256
|
||||
|
||||
|
||||
def test_nvram(router):
|
||||
|
||||
assert router.nvram == 128 # default nvram
|
||||
router.nvram = 256
|
||||
assert router.nvram == 256
|
||||
|
||||
|
||||
def test_mmap(router):
|
||||
|
||||
assert router.mmap == True # default value
|
||||
router.mmap = False
|
||||
assert router.mmap == False
|
||||
|
||||
|
||||
def test_sparsemem(router):
|
||||
|
||||
assert router.sparsemem == True # default value
|
||||
router.sparsemem = False
|
||||
assert router.sparsemem == False
|
||||
|
||||
|
||||
def test_clock_divisor(router):
|
||||
|
||||
assert router.clock_divisor == 8 # default value
|
||||
router.clock_divisor = 4
|
||||
assert router.clock_divisor == 4
|
||||
|
||||
|
||||
def test_idlepc(router):
|
||||
|
||||
assert router.idlepc == "" # no default value
|
||||
router.idlepc = "0x60c086a8"
|
||||
assert router.idlepc == "0x60c086a8"
|
||||
|
||||
|
||||
def test_idlemax(router):
|
||||
|
||||
assert router.idlemax == 500 # default value
|
||||
router.idlemax = 1500
|
||||
assert router.idlemax == 1500
|
||||
|
||||
|
||||
def test_idlesleep(router):
|
||||
|
||||
assert router.idlesleep == 30 # default value
|
||||
router.idlesleep = 15
|
||||
assert router.idlesleep == 15
|
||||
|
||||
|
||||
def test_exec_area(router):
|
||||
|
||||
if sys.platform.startswith("win"):
|
||||
assert router.exec_area == 16 # default value
|
||||
else:
|
||||
assert router.exec_area == 64 # default value
|
||||
router.exec_area = 48
|
||||
assert router.exec_area == 48
|
||||
|
||||
|
||||
def test_disk0(router):
|
||||
|
||||
assert router.disk0 == 0 # default value
|
||||
router.disk0 = 16
|
||||
assert router.disk0 == 16
|
||||
|
||||
|
||||
def test_disk1(router):
|
||||
|
||||
assert router.disk1 == 0 # default value
|
||||
router.disk1 = 16
|
||||
assert router.disk1 == 16
|
||||
|
||||
|
||||
def test_confreg(router):
|
||||
|
||||
assert router.confreg == "0x2102" # default value
|
||||
router.confreg = "0x2142"
|
||||
assert router.confreg == "0x2142"
|
||||
|
||||
|
||||
def test_console(router):
|
||||
|
||||
assert router.console == 2001
|
||||
new_console_port = router.console + 100
|
||||
router.console = new_console_port
|
||||
assert router.console == new_console_port
|
||||
|
||||
|
||||
def test_aux(router):
|
||||
|
||||
assert router.aux == 2501
|
||||
new_aux_port = router.aux + 100
|
||||
router.aux = new_aux_port
|
||||
assert router.aux == new_aux_port
|
||||
|
||||
|
||||
def test_cpu_info(router):
|
||||
|
||||
router.get_cpu_info() # nothing is returned by the hypervisor, cannot test?
|
||||
|
||||
|
||||
def test_cpu_usage(router):
|
||||
|
||||
usage = router.get_cpu_usage()
|
||||
assert usage == 0 # router isn't running, so usage must be 0
|
||||
|
||||
|
||||
def test_get_slot_bindings(router):
|
||||
|
||||
assert router.get_slot_bindings()[0] == "0/0: GT96100-FE"
|
||||
|
||||
|
||||
def test_get_slot_nio_bindings(router):
|
||||
|
||||
router.get_slot_nio_bindings(slot_id=0)
|
||||
|
||||
|
||||
def test_mac_addr(router):
|
||||
|
||||
assert router.mac_addr != None
|
||||
router.mac_addr = "aa:aa:aa:aa:aa:aa"
|
||||
assert router.mac_addr == "aa:aa:aa:aa:aa:aa"
|
||||
|
||||
|
||||
def test_bogus_mac_addr(router):
|
||||
|
||||
with pytest.raises(DynamipsError):
|
||||
router.mac_addr = "zz:zz:zz:zz:zz:zz"
|
||||
|
||||
|
||||
def test_system_id(router):
|
||||
|
||||
assert router.system_id == "FTX0945W0MY" # default value
|
||||
router.system_id = "FTX0945W0MO"
|
||||
assert router.system_id == "FTX0945W0MO"
|
||||
|
||||
|
||||
def test_get_hardware_info(router):
|
||||
|
||||
router.get_hardware_info()
|
||||
@@ -1,65 +0,0 @@
|
||||
from tornado.testing import AsyncHTTPTestCase
|
||||
#from gns3server.plugins.dynamips import Dynamips
|
||||
#from gns3server._compat import urlencode
|
||||
from functools import partial
|
||||
import tornado.web
|
||||
import json
|
||||
import tempfile
|
||||
|
||||
|
||||
# class TestVMHandler(AsyncHTTPTestCase):
|
||||
#
|
||||
# def setUp(self):
|
||||
#
|
||||
# AsyncHTTPTestCase.setUp(self)
|
||||
# self.post_request = partial(self.http_client.fetch,
|
||||
# self.get_url("/api/vms/dynamips"),
|
||||
# self.stop,
|
||||
# method="POST")
|
||||
#
|
||||
# def get_app(self):
|
||||
# return tornado.web.Application(Dynamips().handlers())
|
||||
#
|
||||
# def test_endpoint(self):
|
||||
# self.http_client.fetch(self.get_url("/api/vms/dynamips"), self.stop)
|
||||
# response = self.wait()
|
||||
# assert response.code == 200
|
||||
#
|
||||
# def test_upload(self):
|
||||
#
|
||||
# try:
|
||||
# from poster.encode import multipart_encode
|
||||
# except ImportError:
|
||||
# # poster isn't available for Python 3, let's just ignore the test
|
||||
# return
|
||||
#
|
||||
# file_to_upload = tempfile.NamedTemporaryFile()
|
||||
# data, headers = multipart_encode({"file1": file_to_upload})
|
||||
# body = ""
|
||||
# for d in data:
|
||||
# body += d
|
||||
#
|
||||
# response = self.fetch('/api/vms/dynamips/storage/upload',
|
||||
# headers=headers,
|
||||
# body=body,
|
||||
# method='POST')
|
||||
#
|
||||
# assert response.code == 200
|
||||
#
|
||||
# def get_new_ioloop(self):
|
||||
# return tornado.ioloop.IOLoop.instance()
|
||||
#
|
||||
# def test_create_vm(self):
|
||||
#
|
||||
# post_data = {"name": "R1",
|
||||
# "platform": "c3725",
|
||||
# "console": 2000,
|
||||
# "aux": 3000,
|
||||
# "image": "c3725.bin",
|
||||
# "ram": 128}
|
||||
#
|
||||
# self.post_request(body=json.dumps(post_data))
|
||||
# response = self.wait()
|
||||
# assert(response.headers['Content-Type'].startswith('application/json'))
|
||||
# expected = {"success": True}
|
||||
# assert response.body.decode("utf-8") == json.dumps(expected)
|
||||
@@ -1,41 +0,0 @@
|
||||
from gns3server.modules.iou import IOUDevice
|
||||
import os
|
||||
import pytest
|
||||
|
||||
|
||||
def no_iou():
|
||||
cwd = os.path.dirname(os.path.abspath(__file__))
|
||||
iou_path = os.path.join(cwd, "i86bi_linux-ipbase-ms-12.4.bin")
|
||||
|
||||
if os.path.isfile(iou_path):
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def iou(request):
|
||||
|
||||
cwd = os.path.dirname(os.path.abspath(__file__))
|
||||
iou_path = os.path.join(cwd, "i86bi_linux-ipbase-ms-12.4.bin")
|
||||
iou_device = IOUDevice("IOU1", iou_path, "/tmp")
|
||||
iou_device.start()
|
||||
request.addfinalizer(iou_device.delete)
|
||||
return iou_device
|
||||
|
||||
|
||||
@pytest.mark.skipif(no_iou(), reason="IOU Image not available")
|
||||
def test_iou_is_started(iou):
|
||||
|
||||
print(iou.command())
|
||||
assert iou.id == 1 # we should have only one IOU running!
|
||||
assert iou.is_running()
|
||||
|
||||
|
||||
@pytest.mark.skipif(no_iou(), reason="IOU Image not available")
|
||||
def test_iou_restart(iou):
|
||||
|
||||
iou.stop()
|
||||
assert not iou.is_running()
|
||||
iou.start()
|
||||
assert iou.is_running()
|
||||
@@ -1,91 +0,0 @@
|
||||
import uuid
|
||||
from tornado.testing import AsyncTestCase
|
||||
from tornado.escape import json_encode, json_decode
|
||||
from ws4py.client.tornadoclient import TornadoWebSocketClient
|
||||
import gns3server.jsonrpc as jsonrpc
|
||||
|
||||
"""
|
||||
Tests for JSON-RPC protocol over Websockets
|
||||
"""
|
||||
|
||||
|
||||
class JSONRPC(AsyncTestCase):
|
||||
|
||||
URL = "ws://127.0.0.1:8000/"
|
||||
|
||||
def test_request(self):
|
||||
|
||||
params = {"echo": "test"}
|
||||
request = jsonrpc.JSONRPCRequest("dynamips.echo", params)
|
||||
AsyncWSRequest(self.URL, self.io_loop, self.stop, str(request))
|
||||
response = self.wait()
|
||||
json_response = json_decode(response)
|
||||
assert json_response["jsonrpc"] == 2.0
|
||||
assert json_response["id"] == request.id
|
||||
assert json_response["result"] == params
|
||||
|
||||
def test_request_with_invalid_method(self):
|
||||
|
||||
message = {"echo": "test"}
|
||||
request = jsonrpc.JSONRPCRequest("dynamips.non_existent", message)
|
||||
AsyncWSRequest(self.URL, self.io_loop, self.stop, str(request))
|
||||
response = self.wait()
|
||||
json_response = json_decode(response)
|
||||
assert json_response["error"].get("code") == -32601
|
||||
assert json_response["id"] == request.id
|
||||
|
||||
def test_request_with_invalid_version(self):
|
||||
|
||||
request = {"jsonrpc": 1.0, "method": "dynamips.echo", "id": 1}
|
||||
AsyncWSRequest(self.URL, self.io_loop, self.stop, json_encode(request))
|
||||
response = self.wait()
|
||||
json_response = json_decode(response)
|
||||
assert json_response["id"] is None
|
||||
assert json_response["error"].get("code") == -32600
|
||||
|
||||
def test_request_with_invalid_json(self):
|
||||
|
||||
request = "my non JSON request"
|
||||
AsyncWSRequest(self.URL, self.io_loop, self.stop, request)
|
||||
response = self.wait()
|
||||
json_response = json_decode(response)
|
||||
assert json_response["id"] is None
|
||||
assert json_response["error"].get("code") == -32700
|
||||
|
||||
def test_request_with_invalid_jsonrpc_field(self):
|
||||
|
||||
request = {"jsonrpc": "2.0", "method_bogus": "dynamips.echo", "id": 1}
|
||||
AsyncWSRequest(self.URL, self.io_loop, self.stop, json_encode(request))
|
||||
response = self.wait()
|
||||
json_response = json_decode(response)
|
||||
assert json_response["id"] is None
|
||||
assert json_response["error"].get("code") == -32700
|
||||
|
||||
def test_request_with_no_params(self):
|
||||
|
||||
request = jsonrpc.JSONRPCRequest("dynamips.echo")
|
||||
AsyncWSRequest(self.URL, self.io_loop, self.stop, str(request))
|
||||
response = self.wait()
|
||||
json_response = json_decode(response)
|
||||
assert json_response["id"] == request.id
|
||||
assert json_response["error"].get("code") == -32602
|
||||
|
||||
|
||||
class AsyncWSRequest(TornadoWebSocketClient):
|
||||
"""
|
||||
Very basic Websocket client for tests
|
||||
"""
|
||||
|
||||
def __init__(self, url, io_loop, callback, message):
|
||||
TornadoWebSocketClient.__init__(self, url, io_loop=io_loop)
|
||||
self._callback = callback
|
||||
self._message = message
|
||||
self.connect()
|
||||
|
||||
def opened(self):
|
||||
self.send(self._message, binary=False)
|
||||
|
||||
def received_message(self, message):
|
||||
self.close()
|
||||
if self._callback:
|
||||
self._callback(message.data)
|
||||
@@ -1,40 +0,0 @@
|
||||
from tornado.testing import AsyncHTTPTestCase
|
||||
from tornado.escape import json_decode
|
||||
from gns3server.server import VersionHandler
|
||||
from gns3server.version import __version__
|
||||
import tornado.web
|
||||
|
||||
"""
|
||||
Tests for the web server version handler
|
||||
"""
|
||||
|
||||
|
||||
class TestVersionHandler(AsyncHTTPTestCase):
|
||||
|
||||
URL = "/version"
|
||||
|
||||
def get_app(self):
|
||||
|
||||
return tornado.web.Application([(self.URL, VersionHandler)])
|
||||
|
||||
def test_endpoint(self):
|
||||
"""
|
||||
Tests if the response HTTP code is 200 (success)
|
||||
"""
|
||||
|
||||
self.http_client.fetch(self.get_url(self.URL), self.stop)
|
||||
response = self.wait()
|
||||
assert response.code == 200
|
||||
|
||||
def test_received_version(self):
|
||||
"""
|
||||
Tests if the returned content type is JSON and
|
||||
if the received version is the same as the server
|
||||
"""
|
||||
|
||||
self.http_client.fetch(self.get_url(self.URL), self.stop)
|
||||
response = self.wait()
|
||||
assert response.headers['Content-Type'].startswith('application/json')
|
||||
assert response.body
|
||||
body = json_decode(response.body)
|
||||
assert body['version'] == __version__
|
||||
@@ -1,33 +0,0 @@
|
||||
from gns3server.modules.vpcs import VPCSDevice
|
||||
import os
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def vpcs(request):
|
||||
|
||||
if os.path.isfile("/usr/bin/vpcs"):
|
||||
vpcs_path = "/usr/bin/vpcs"
|
||||
else:
|
||||
cwd = os.path.dirname(os.path.abspath(__file__))
|
||||
vpcs_path = os.path.join(cwd, "vpcs")
|
||||
vpcs_device = VPCSDevice("VPCS1", vpcs_path, "/tmp")
|
||||
vpcs_device.port_add_nio_binding(0, 'nio_tap:tap0')
|
||||
vpcs_device.start()
|
||||
request.addfinalizer(vpcs_device.delete)
|
||||
return vpcs_device
|
||||
|
||||
|
||||
def test_vpcs_is_started(vpcs):
|
||||
|
||||
print(vpcs.command())
|
||||
assert vpcs.id == 1 # we should have only one VPCS running!
|
||||
assert vpcs.is_running()
|
||||
|
||||
|
||||
def test_vpcs_restart(vpcs):
|
||||
|
||||
vpcs.stop()
|
||||
assert not vpcs.is_running()
|
||||
vpcs.start()
|
||||
assert vpcs.is_running()
|
||||
Reference in new issue
Block a user