class Roads09Ns3PLExample(object):
testbed_id = "planetlab"
- testbed_version = "01"
slicename = "inria_nepi"
plchost = "nepiplc.pl.sophia.inria.fr"
def make_experiment_desc(self):
testbed_id = self.testbed_id
- testbed_version = self.testbed_version
slicename = self.slicename
plchost = self.plchost
pl_ssh_key = os.environ.get(
pl_user, pl_pwd = pl_auth()
exp_desc = ExperimentDescription()
- pl_provider = FactoriesProvider(testbed_id, testbed_version)
+ pl_provider = FactoriesProvider(testbed_id)
pl_desc = exp_desc.add_testbed_description(pl_provider)
pl_desc.set_attribute_value("homeDirectory", self.root_dir)
pl_desc.set_attribute_value("slice", slicename)
def make_ns_in_pl(self, pl, exp, node1, iface1, root):
ns3_testbed_id = "ns3"
- ns3_testbed_version = "3_9_RC3"
# Add NS3 support in node1
plnepi = pl.create("NepiDependency")
plns3.connector("node").connect(node1.connector("deps"))
# Create NS3 testbed running in node1
- ns3_provider = FactoriesProvider(ns3_testbed_id, ns3_testbed_version)
+ ns3_provider = FactoriesProvider(ns3_testbed_id)
ns3_desc = exp.add_testbed_description(ns3_provider)
ns3_desc.set_attribute_value("rootDirectory", root)
ns3_desc.set_attribute_value("SimulatorImplementationType", "ns3::RealtimeSimulatorImpl")
exp_desc = ExperimentDescription()
## NS3 Testbed instance description ##
- testbed_version = "3_9"
testbed_id = "ns3"
- ns3_provider = FactoriesProvider(testbed_id, testbed_version)
+ ns3_provider = FactoriesProvider(testbed_id)
ns3_desc = exp_desc.add_testbed_description(ns3_provider)
ns3_desc.set_attribute_value("homeDirectory", self.root_dir)
ns3_desc.set_attribute_value("SimulatorImplementationType", "ns3::RealtimeSimulatorImpl")
phy2.connector("chan").connect(wifichan.connector("phys"))
## NETNS testbed description 1 ##
- testbed_version = "01"
testbed_id = "netns"
- netns_provider = FactoriesProvider(testbed_id, testbed_version)
+ netns_provider = FactoriesProvider(testbed_id)
netns_desc1 = exp_desc.add_testbed_description(netns_provider)
netns_desc1.set_attribute_value("homeDirectory", self.root_dir)
#netns_desc1.set_attribute_value("enableDebug", True)
def run(self):
exp_desc = ExperimentDescription()
- testbed_version = "3_9"
testbed_id = "ns3"
- ns3_provider = FactoriesProvider(testbed_id, testbed_version)
+ ns3_provider = FactoriesProvider(testbed_id)
ns3_desc = exp_desc.add_testbed_description(ns3_provider)
ns3_desc.set_attribute_value("homeDirectory", self.root_dir)
ns3_desc.set_attribute_value("StopTime", "7.1s")
self._connectors = self._traces = self._factory_attributes = None
class FactoriesProvider(object):
- def __init__(self, testbed_id, testbed_version):
+ def __init__(self, testbed_id):
super(FactoriesProvider, self).__init__()
self._testbed_id = testbed_id
- self._testbed_version = testbed_version
self._factories = dict()
- metadata = Metadata(testbed_id, testbed_version)
+ metadata = Metadata(testbed_id)
for factory in metadata.build_factories():
self.add_factory(factory)
+ self._testbed_version = metadata.testbed_version
+
@property
def testbed_id(self):
return self._testbed_id
self._boxes = dict()
self.graphical_info = GraphicalInfo()
- metadata = Metadata(provider.testbed_id, provider.testbed_version)
+ metadata = Metadata(provider.testbed_id)
for attr in metadata.testbed_attributes().attributes:
self.add_attribute(attr.name, attr.help, attr.type, attr.value,
attr.range, attr.allowed, attr.flags,
# force recovery mode
deployment_config.set_attribute_value("recover",recover)
- testbed = proxy.create_testbed_controller(testbed_id,
- testbed_version, deployment_config)
+ testbed = proxy.create_testbed_controller(testbed_id, testbed_version,
+ deployment_config)
for (name, value) in data.get_attribute_data(guid):
testbed.defer_configure(name, value)
self._testbeds[guid] = testbed
DeploymentConfiguration as DC, \
AttributeCategories as AC
-class VersionedMetadataInfo(object):
+class MetadataInfo(object):
@property
def connector_types(self):
""" dictionary of dictionaries with allowed connection information.
"""
raise NotImplementedError
+ @property
+ def testbed_id(self):
+ """ ID for the testbed """
+ raise NotImplementedError
+
+ @property
+ def testbed_version(self):
+ """ version for the testbed """
+ raise NotImplementedError
+
class Metadata(object):
# These attributes should be added to all boxes
STANDARD_BOX_ATTRIBUTES = dict({
STANDARD_TESTBED_ATTRIBUTES.update(DEPLOYMENT_ATTRIBUTES.copy())
- def __init__(self, testbed_id, version):
- self._version = version
+ def __init__(self, testbed_id):
self._testbed_id = testbed_id
- metadata_module = self._load_versioned_metadata_module()
- self._metadata = metadata_module.VersionedMetadataInfo()
+ metadata_module = self._load_metadata_module()
+ self._metadata = metadata_module.MetadataInfo()
+ if testbed_id != self._metadata.testbed_id:
+ raise RuntimeError("Bad testbed id. Asked for %s, got %s" % \
+ (testbed_id, self._metadata.testbed_id ))
@property
def create_order(self):
def start_order(self):
return self._metadata.start_order
+ @property
+ def testbed_version(self):
+ return self._metadata.testbed_version
+
+ @property
+ def testbed_id(self):
+ return self._testbed_id
+
def testbed_attributes(self):
attributes = AttributesMap()
testbed_attributes = self._testbed_attributes()
factories.append(factory)
return factories
- def _load_versioned_metadata_module(self):
- mod_name = "nepi.testbeds.%s.metadata_v%s" % (self._testbed_id.lower(),
- self._version)
+ def _load_metadata_module(self):
+ mod_name = "nepi.testbeds.%s.metadata" % (self._testbed_id.lower())
if not mod_name in sys.modules:
__import__(mod_name)
return sys.modules[mod_name]
# testbed element instances
self._elements = dict()
- self._metadata = Metadata(self._testbed_id, self._testbed_version)
+ self._metadata = Metadata(self._testbed_id)
+ if self._metadata.testbed_version != testbed_version:
+ raise RuntimeError("Bad testbed version on testbed %s. Asked for %s, got %s" % \
+ (testbed_id, testbed_version, self._metadata.testbed_version))
for factory in self._metadata.build_factories():
self._factories[factory.factory_id] = factory
self._attributes = self._metadata.testbed_attributes()
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_ID
from execute import TestbedController
# -*- coding: utf-8 -*-
TESTBED_ID = "netns"
+TESTBED_VERSION = "0.1"
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import testbed_impl
from nepi.util.constants import TIME_NOW
import os
finally:
processcond.release()
- def __init__(self, testbed_version):
- super(TestbedController, self).__init__(TESTBED_ID, testbed_version)
+ def __init__(self):
+ super(TestbedController, self).__init__(TESTBED_ID, TESTBED_VERSION)
self._netns = None
self._home_directory = None
self._traces = dict()
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import metadata
from nepi.core.attributes import Attribute
from nepi.util import tags, validation
}),
})
-class VersionedMetadataInfo(metadata.VersionedMetadataInfo):
+class MetadataInfo(metadata.MetadataInfo):
@property
def connector_types(self):
return connector_types
def testbed_attributes(self):
return testbed_attributes
+ @property
+ def testbed_id(self):
+ return TESTBED_ID
+
+ @property
+ def testbed_version(self):
+ return TESTBED_VERSION
+
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from execute import TestbedController
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from factories_metadata_v3_9 import wifi_standards, l4_protocols, \
+from factories_metadata import wifi_standards, l4_protocols, \
service_flow_direction, service_flow_scheduling_type
import validation as ns3_validation
from nepi.core.attributes import Attribute
# -*- coding: utf-8 -*-
TESTBED_ID = "ns3"
+TESTBED_VERSION = "3.9"
from nepi.core import testbed_impl
from nepi.core.attributes import Attribute
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.util.constants import TIME_NOW, TestbedStatus as TS
import os
import sys
import weakref
def init():
- if 'ns3' in sys.modules:
- return
+ if 'ns3' in sys.modules:
+ return
- import ctypes
- import imp
+ import ctypes
+ import imp
- bindings = os.environ["NEPI_NS3BINDINGS"] \
+ bindings = os.environ["NEPI_NS3BINDINGS"] \
if "NEPI_NS3BINDINGS" in os.environ else None
- libfile = os.environ["NEPI_NS3LIBRARY"] \
+ libfile = os.environ["NEPI_NS3LIBRARY"] \
if "NEPI_NS3LIBRARY" in os.environ else None
- if libfile:
- ctypes.CDLL(libfile, ctypes.RTLD_GLOBAL)
+ if libfile:
+ ctypes.CDLL(libfile, ctypes.RTLD_GLOBAL)
+ """
+ files = os.listdir(build_path)
+ regex = re.compile("(.*\.so)$")
+ libs = [m.group(1) for filename in files for m in [regex.search(filename)] if m]
+ i = 0
+ while len(libs) > 0:
+ i += 1
+ lib = libs[ i % len(libs)]
+ libfile = os.path.join(build_path, lib)
+ try:
+ ctypes.CDLL(libfile, ctypes.RTLD_GLOBAL)
+ libs.remove(lib)
+ except:
+ pass
+ """
path = [ os.path.dirname(__file__) ] + sys.path
if bindings:
LOCAL_TYPES = tuple(LOCAL_FACTORIES.values())
- def __init__(self, testbed_version):
- super(TestbedController, self).__init__(TESTBED_ID, testbed_version)
+ def __init__(self):
+ super(TestbedController, self).__init__(TESTBED_ID, TESTBED_VERSION)
self._ns3 = None
self._home_directory = None
self._traces = dict()
--- /dev/null
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from constants import TESTBED_ID, TESTBED_VERSION
+from nepi.core import metadata
+
+class MetadataInfo(metadata.MetadataInfo):
+ @property
+ def connector_types(self):
+ from connection_metadata import connector_types
+ return connector_types
+
+ @property
+ def connections(self):
+ from connection_metadata import connections
+ return connections
+
+ @property
+ def attributes(self):
+ from attributes_metadata import attributes
+ return attributes
+
+ @property
+ def traces(self):
+ from traces_metadata import traces
+ return traces
+
+ @property
+ def create_order(self):
+ from factories_metadata import factories_order
+ return factories_order
+
+ @property
+ def configure_order(self):
+ from factories_metadata import factories_order
+ return factories_order
+
+ @property
+ def factories_info(self):
+ from factories_metadata import factories_info
+ return factories_info
+
+ @property
+ def testbed_attributes(self):
+ from attributes_metadata import testbed_attributes
+ return testbed_attributes
+
+ @property
+ def testbed_id(self):
+ return TESTBED_ID
+
+ @property
+ def testbed_version(self):
+ return TESTBED_VERSION
+
+++ /dev/null
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-from nepi.core import metadata
-
-class VersionedMetadataInfo(metadata.VersionedMetadataInfo):
- @property
- def connector_types(self):
- from connection_metadata_v3_9 import connector_types
- return connector_types
-
- @property
- def connections(self):
- from connection_metadata_v3_9 import connections
- return connections
-
- @property
- def attributes(self):
- from attributes_metadata_v3_9 import attributes
- return attributes
-
- @property
- def traces(self):
- from traces_metadata_v3_9 import traces
- return traces
-
- @property
- def create_order(self):
- from factories_metadata_v3_9 import factories_order
- return factories_order
-
- @property
- def configure_order(self):
- from factories_metadata_v3_9 import factories_order
- return factories_order
-
- @property
- def factories_info(self):
- from factories_metadata_v3_9 import factories_info
- return factories_info
-
- @property
- def testbed_attributes(self):
- from attributes_metadata_v3_9 import testbed_attributes
- return testbed_attributes
-
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from execute import TestbedController
# -*- coding: utf-8 -*-
TESTBED_ID = "planetlab"
+TESTBED_VERSION = "0.1"
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import testbed_impl
from nepi.util.constants import TIME_NOW
from nepi.util.graphtools import mst
pass
class TestbedController(testbed_impl.TestbedController):
- def __init__(self, testbed_version):
- super(TestbedController, self).__init__(TESTBED_ID, testbed_version)
+ def __init__(self):
+ super(TestbedController, self).__init__(TESTBED_ID, TESTBED_VERSION)
self._home_directory = None
self.slicename = None
self._traces = dict()
import time
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import metadata
from nepi.core.attributes import Attribute
from nepi.util import tags, validation
}),
})
-class VersionedMetadataInfo(metadata.VersionedMetadataInfo):
+class MetadataInfo(metadata.MetadataInfo):
@property
def connector_types(self):
return connector_types
def testbed_attributes(self):
return testbed_attributes
+ @property
+ def testbed_id(self):
+ return TESTBED_ID
+
+ @property
+ def testbed_version(self):
+ return TESTBED_VERSION
+
def testbed_from_data(self, experiment_description, guid, data):
from nepi.core.design import FactoriesProvider
(testbed_id, testbed_version) = data.get_testbed_data(guid)
- provider = FactoriesProvider(testbed_id, testbed_version)
+ provider = FactoriesProvider(testbed_id)
+ if provider.testbed_version != testbed_version:
+ raise RuntimeError("Bad testbed version on testbed %s. Asked for %s, got %s" % \
+ (testbed_id, testbed_version, provider.testbed_version))
experiment_description.add_testbed_description(provider, guid)
testbed_description = experiment_description.testbed_description(guid)
self.graphical_info_from_data(testbed_description, data)
if not mod_name in sys.modules:
__import__(mod_name)
module = sys.modules[mod_name]
- return module.TestbedController(testbed_version)
+ tc = module.TestbedController()
+ if tc.testbed_version != testbed_version:
+ raise RuntimeError("Bad testbed version on testbed %s. Asked for %s, got %s" % \
+ (testbed_id, testbed_version, tc.testbed_version))
+ return tc
# Just a namespace class
class Marshalling:
from nepi.core.design import ExperimentDescription, FactoriesProvider
from nepi.util import tags
-import mock.metadata_v01
+import mock.metadata
import sys
import unittest
class DesignTestCase(unittest.TestCase):
def setUp(self):
- sys.modules["nepi.testbeds.mock.metadata_v01"] = mock.metadata_v01
+ sys.modules["nepi.testbeds.mock.metadata"] = mock.metadata
def test_design(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id = "mock"
- provider = FactoriesProvider(testbed_id, testbed_version)
+ provider = FactoriesProvider(testbed_id)
desc = exp_desc.add_testbed_description(provider)
desc.set_attribute_value("fake", True)
node1 = desc.create("Node")
from nepi.util.constants import ApplicationStatus as AS
import getpass
import mock
-import mock.metadata_v01
+import mock.metadata
import sys
import time
import unittest
class ExecuteTestCase(unittest.TestCase):
def setUp(self):
- sys.modules["nepi.testbeds.mock.metadata_v01"] = mock.metadata_v01
+ sys.modules["nepi.testbeds.mock.metadata"] = mock.metadata
def test_execute(self):
- testbed_version = "01"
- testbed_id = "mock"
- instance = mock.TestbedController(testbed_version)
+ instance = mock.TestbedController()
instance.defer_configure("fake", True)
instance.defer_create(2, "Node")
instance.defer_create(3, "Node")
from nepi.util.constants import DeploymentConfiguration as DC
import getpass
import mock
-import mock.metadata_v01
+import mock.metadata
import mock2
-import mock2.metadata_v01
+import mock2.metadata
import os
import shutil
import sys
class ExecuteTestCase(unittest.TestCase):
def setUp(self):
- sys.modules["nepi.testbeds.mock.metadata_v01"] = mock.metadata_v01
+ sys.modules["nepi.testbeds.mock.metadata"] = mock.metadata
sys.modules["nepi.testbeds.mock"] = mock
- sys.modules["nepi.testbeds.mock2.metadata_v01"] = mock2.metadata_v01
+ sys.modules["nepi.testbeds.mock2.metadata"] = mock2.metadata
sys.modules["nepi.testbeds.mock2"] = mock2
self.root_dir = tempfile.mkdtemp()
time.sleep(0.1)
shutil.rmtree(self.root_dir)
- def make_testbed(self, exp_desc, testbed_id, testbed_version):
- provider = FactoriesProvider(testbed_id, testbed_version)
+ def make_testbed(self, exp_desc, testbed_id):
+ provider = FactoriesProvider(testbed_id)
desc = exp_desc.add_testbed_description(provider)
desc.set_attribute_value("fake", True)
node1 = desc.create("Node")
def make_test_experiment(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id = "mock"
- return self.make_testbed(exp_desc, testbed_id, testbed_version)
+ return self.make_testbed(exp_desc, testbed_id)
def make_cross_test_experiment(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id1 = "mock"
testbed_id2 = "mock2"
exp_desc, desc1, app1, node11, node12, iface11, iface12 = \
- self.make_testbed(exp_desc, testbed_id1, testbed_version)
+ self.make_testbed(exp_desc, testbed_id1)
exp_desc, desc2, app2, node21, node22, iface21, iface22 = \
- self.make_testbed(exp_desc, testbed_id2, testbed_version)
+ self.make_testbed(exp_desc, testbed_id2)
iface12.connector("cross").connect(iface21.connector("cross"))
return exp_desc, desc1, desc2, iface12, iface21
self.assertTrue(fake_result.startswith(comp_result))
self.assertEquals(controller.get_testbed_id(node1.guid), "mock")
- self.assertEquals(controller.get_testbed_version(node1.guid), "01")
+ self.assertEquals(controller.get_testbed_version(node1.guid), "0.1")
self.assertEquals(controller.get_factory_id(node1.guid), "Node")
controller.stop()
self.assertTrue(fake_result.startswith(comp_result))
self.assertEquals(controller.get_testbed_id(node1.guid), "mock")
- self.assertEquals(controller.get_testbed_version(node1.guid), "01")
+ self.assertEquals(controller.get_testbed_version(node1.guid), "0.1")
self.assertEquals(controller.get_factory_id(node1.guid), "Node")
controller.stop()
self.assertTrue(fake_result.startswith(comp_result))
self.assertEquals(controller.get_testbed_id(node1.guid), "mock")
- self.assertEquals(controller.get_testbed_version(node1.guid), "01")
+ self.assertEquals(controller.get_testbed_version(node1.guid), "0.1")
self.assertEquals(controller.get_factory_id(node1.guid), "Node")
controller.stop()
self.assertTrue(fake_result.startswith(comp_result))
self.assertEquals(controller.get_testbed_id(node1.guid), "mock")
- self.assertEquals(controller.get_testbed_version(node1.guid), "01")
+ self.assertEquals(controller.get_testbed_version(node1.guid), "0.1")
self.assertEquals(controller.get_factory_id(node1.guid), "Node")
traces_info = controller.traces_info()
self.assertTrue(fake_result.startswith(comp_result))
self.assertEquals(controller.get_testbed_id(node1.guid), "mock")
- self.assertEquals(controller.get_testbed_version(node1.guid), "01")
+ self.assertEquals(controller.get_testbed_version(node1.guid), "0.1")
self.assertEquals(controller.get_factory_id(node1.guid), "Node")
# controller dies
addr.set_attribute_value("Address", "10.0.0.2")
desc2 = exp_desc.add_testbed_description(
- FactoriesProvider("mock2", "01") )
+ FactoriesProvider("mock2") )
desc2.set_attribute_value(DC.DEPLOYMENT_HOST, "{#[some].addr[0].[Address]#}")
# DC.DEPLOYMENT_HOST should be ignored if DC.DEPLOYMENT_CONNECTION is not set
# But it should be resolved anyway
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from execute import TestbedController
# -*- coding: utf-8 -*-
TESTBED_ID = "mock"
-
+TESTBED_VERSION = "0.1"
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import testbed_impl
class TestbedController(testbed_impl.TestbedController):
- def __init__(self, testbed_version):
- super(TestbedController, self).__init__(TESTBED_ID, testbed_version)
+ def __init__(self):
+ super(TestbedController, self).__init__(TESTBED_ID, TESTBED_VERSION)
def do_configure(self):
pass
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import metadata
from nepi.core.attributes import Attribute
from nepi.util import validation, tags
}),
})
-class VersionedMetadataInfo(metadata.VersionedMetadataInfo):
+class MetadataInfo(metadata.MetadataInfo):
@property
def connector_types(self):
return connector_types
def testbed_attributes(self):
return testbed_attributes
+ @property
+ def testbed_id(self):
+ return TESTBED_ID
+
+ @property
+ def testbed_version(self):
+ return TESTBED_VERSION
+
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from execute import TestbedController
# -*- coding: utf-8 -*-
TESTBED_ID = "mock2"
+TESTBED_VERSION = "0.1"
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import testbed_impl
class TestbedController(testbed_impl.TestbedController):
- def __init__(self, testbed_version):
- super(TestbedController, self).__init__(TESTBED_ID, testbed_version)
+ def __init__(self):
+ super(TestbedController, self).__init__(TESTBED_ID, TESTBED_VERSION)
def do_configure(self):
pass
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from constants import TESTBED_ID
+from constants import TESTBED_ID, TESTBED_VERSION
from nepi.core import metadata
from nepi.core.attributes import Attribute
from nepi.util import tags, validation
}),
})
-class VersionedMetadataInfo(metadata.VersionedMetadataInfo):
+class MetadataInfo(metadata.MetadataInfo):
@property
def connector_types(self):
return connector_types
def testbed_attributes(self):
return testbed_attributes
+ @property
+ def testbed_id(self):
+ return TESTBED_ID
+
+ @property
+ def testbed_version(self):
+ return TESTBED_VERSION
+
class NetnsDesignTestCase(unittest.TestCase):
def test_design_if(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id = "netns"
- provider = FactoriesProvider(testbed_id, testbed_version)
+ provider = FactoriesProvider(testbed_id)
tstbd_desc = exp_desc.add_testbed_description(provider)
tstbd_desc.set_attribute_value("enableDebug", True)
node1 = tstbd_desc.create("Node")
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
def test_run_ping_if(self):
user = getpass.getuser()
- testbed_version = "01"
- instance = netns.TestbedController(testbed_version)
+ instance = netns.TestbedController()
instance.defer_configure("homeDirectory", self.root_dir)
instance.defer_create(2, "Node")
instance.defer_create(3, "Node")
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
def test_run_ping_p2pif(self):
user = getpass.getuser()
- testbed_version = "01"
- instance = netns.TestbedController(testbed_version)
+ instance = netns.TestbedController()
instance.defer_configure("homeDirectory", self.root_dir)
instance.defer_create(2, "Node")
instance.defer_create(3, "Node")
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
def test_run_ping_routing(self):
user = getpass.getuser()
- testbed_version = "01"
- instance = netns.TestbedController(testbed_version)
+ instance = netns.TestbedController()
instance.defer_configure("homeDirectory", self.root_dir)
instance.defer_create(2, "Node")
instance.defer_create(3, "Node")
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
def test_node_pcap_trace(self):
user = getpass.getuser()
- testbed_version = "01"
- instance = netns.TestbedController(testbed_version)
+ instance = netns.TestbedController()
instance.defer_configure("homeDirectory", self.root_dir)
instance.defer_create(2, "Node")
instance.defer_add_trace(2, "pcap")
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
def test_local_if(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id = "netns"
user = getpass.getuser()
- netns_provider = FactoriesProvider(testbed_id, testbed_version)
+ netns_provider = FactoriesProvider(testbed_id)
netns_desc = exp_desc.add_testbed_description(netns_provider)
netns_desc.set_attribute_value("homeDirectory", self.root_dir)
#netns_desc.set_attribute_value("enableDebug", True)
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
def test_all_daemonized_if(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id = "netns"
user = getpass.getuser()
- netns_provider = FactoriesProvider(testbed_id, testbed_version)
+ netns_provider = FactoriesProvider(testbed_id)
netns_desc = exp_desc.add_testbed_description(netns_provider)
netns_desc.set_attribute_value("homeDirectory", self.root_dir)
#netns_desc.set_attribute_value("enableDebug", True)
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
def test_all_ssh_daemonized_if(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id = "netns"
env = test_util.test_environment()
user = getpass.getuser()
- netns_provider = FactoriesProvider(testbed_id, testbed_version)
+ netns_provider = FactoriesProvider(testbed_id)
netns_desc = exp_desc.add_testbed_description(netns_provider)
netns_desc.set_attribute_value("homeDirectory", self.root_dir)
#netns_desc.set_attribute_value("enableDebug", True)
class Ns3DesignTestCase(unittest.TestCase):
def test_design_if(self):
exp_desc = ExperimentDescription()
- testbed_version = "3_9"
testbed_id = "ns3"
- provider = FactoriesProvider(testbed_id, testbed_version)
+ provider = FactoriesProvider(testbed_id)
tstbd_desc = exp_desc.add_testbed_description(provider)
node1 = tstbd_desc.create("ns3::Node")
@test_util.skipUnless(test_util.ns3_usable(),
"Test requires working ns-3 bindings")
def test_run_ping_if(self):
- testbed_version = "3_9"
- instance = ns3.TestbedController(testbed_version)
+ instance = ns3.TestbedController()
instance.defer_configure("homeDirectory", self.root_dir)
instance.defer_create(2, "ns3::Node")
instance.defer_create(3, "ns3::Ipv4L3Protocol")
@test_util.skipUnless(test_util.ns3_usable(),
"Test requires working ns-3 bindings")
def test_run_ping_routing(self):
- testbed_version = "3_9"
- instance = ns3.TestbedController(testbed_version)
+ instance = ns3.TestbedController()
instance.defer_configure("homeDirectory", self.root_dir)
instance.defer_create(2, "ns3::Node")
instance.defer_create(3, "ns3::Ipv4L3Protocol")
"Test requires working ns-3 bindings")
def test_local_if(self):
exp_desc = ExperimentDescription()
- testbed_version = "3_9"
testbed_id = "ns3"
- ns3_provider = FactoriesProvider(testbed_id, testbed_version)
+ ns3_provider = FactoriesProvider(testbed_id)
ns3_desc = exp_desc.add_testbed_description(ns3_provider)
ns3_desc.set_attribute_value("homeDirectory", self.root_dir)
"Test requires working ns-3 bindings")
def test_all_daemonized_if(self):
exp_desc = ExperimentDescription()
- testbed_version = "3_9"
testbed_id = "ns3"
- ns3_provider = FactoriesProvider(testbed_id, testbed_version)
+ ns3_provider = FactoriesProvider(testbed_id)
ns3_desc = exp_desc.add_testbed_description(ns3_provider)
ns3_desc.set_attribute_value("homeDirectory", self.root_dir)
class PlanetlabDesignTestCase(unittest.TestCase):
def make_test_design(self):
exp_desc = ExperimentDescription()
- testbed_version = "01"
testbed_id = "planetlab"
- provider = FactoriesProvider(testbed_id, testbed_version)
+ provider = FactoriesProvider(testbed_id)
tstbd_desc = exp_desc.add_testbed_description(provider)
tstbd_desc.set_attribute_value("slice", "inria_nepi")
node1 = tstbd_desc.create("Node")
class PlanetLabExecuteTestCase(unittest.TestCase):
testbed_id = "planetlab"
- testbed_version = "01"
slicename = "inria_nepi"
plchost = "nepiplc.pl.sophia.inria.fr"
def make_instance(self):
testbed_id = self.testbed_id
- testbed_version = self.testbed_version
slicename = self.slicename
plchost = self.plchost
- instance = planetlab.TestbedController(testbed_version)
+ instance = planetlab.TestbedController()
pl_ssh_key = os.environ.get(
"PL_SSH_KEY",
"%s/.ssh/id_rsa_planetlab" % (os.environ['HOME'],) )
class PlanetLabIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"
- testbed_version = "01"
slicename = "inria_nepi"
plchost = "nepiplc.pl.sophia.inria.fr"
def make_experiment_desc(self):
testbed_id = self.testbed_id
- testbed_version = self.testbed_version
slicename = self.slicename
plchost = self.plchost
pl_ssh_key = os.environ.get(
pl_user, pl_pwd = test_util.pl_auth()
exp_desc = ExperimentDescription()
- pl_provider = FactoriesProvider(testbed_id, testbed_version)
+ pl_provider = FactoriesProvider(testbed_id)
pl_desc = exp_desc.add_testbed_description(pl_provider)
pl_desc.set_attribute_value("homeDirectory", self.root_dir)
pl_desc.set_attribute_value("slice", slicename)
class PlanetLabMultiIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"
- testbed_version = "01"
slicename1 = "inria_nepi"
plchost1 = "nepiplc.pl.sophia.inria.fr"
def make_experiment_desc(self):
testbed_id = self.testbed_id
- testbed_version = self.testbed_version
slicename1 = self.slicename1
plchost1 = self.plchost1
pl_user, pl_pwd = test_util.pl_auth()
exp_desc = ExperimentDescription()
- pl_provider = FactoriesProvider(testbed_id, testbed_version)
+ pl_provider = FactoriesProvider(testbed_id)
pl_desc = exp_desc.add_testbed_description(pl_provider)
pl_desc.set_attribute_value("homeDirectory", self.root_dir)
pl_desc.set_attribute_value("slice", slicename1)
def make_ns_in_pl(self, pl, exp, node1, iface1, root):
ns3_testbed_id = "ns3"
- ns3_testbed_version = "3_9"
# Add NS3 support in node1
plnepi = pl.create("NepiDependency")
plns3.connector("node").connect(node1.connector("deps"))
# Create NS3 testbed running in node1
- ns3_provider = FactoriesProvider(ns3_testbed_id, ns3_testbed_version)
+ ns3_provider = FactoriesProvider(ns3_testbed_id)
ns3_desc = exp.add_testbed_description(ns3_provider)
ns3_desc.set_attribute_value("rootDirectory", root)
ns3_desc.set_attribute_value("SimulatorImplementationType", "ns3::RealtimeSimulatorImpl")
class PlanetLabMultiIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"
- testbed_version = "01"
slicename1 = "inria_nepi"
plchost1 = "nepiplc.pl.sophia.inria.fr"
def make_experiment_desc(self):
testbed_id = self.testbed_id
- testbed_version = self.testbed_version
slicename1 = self.slicename1
plchost1 = self.plchost1
pl_user, pl_pwd = test_util.pl_auth()
exp_desc = ExperimentDescription()
- pl_provider = FactoriesProvider(testbed_id, testbed_version)
+ pl_provider = FactoriesProvider(testbed_id)
pl_desc = exp_desc.add_testbed_description(pl_provider)
pl_desc.set_attribute_value("homeDirectory", self.root_dir)
pl_desc.set_attribute_value("slice", slicename1)
class PlanetLabCrossIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"
- testbed_version = "01"
slicename = "inria_nepi"
plchost = "nepiplc.pl.sophia.inria.fr"
def make_experiment_desc(self):
testbed_id = self.testbed_id
- testbed_version = self.testbed_version
slicename = self.slicename
plchost = self.plchost
pl_ssh_key = os.environ.get(
pl_user, pl_pwd = test_util.pl_auth()
exp_desc = ExperimentDescription()
- pl_provider = FactoriesProvider(testbed_id, testbed_version)
+ pl_provider = FactoriesProvider(testbed_id)
pl_desc = exp_desc.add_testbed_description(pl_provider)
pl_desc.set_attribute_value("homeDirectory", self.root_dir)
pl_desc.set_attribute_value("slice", slicename)
def make_ns_in_pl(self, pl, exp, node1, iface1, root):
ns3_testbed_id = "ns3"
- ns3_testbed_version = "3_9"
# Add NS3 support in node1
plnepi = pl.create("NepiDependency")
plns3.connector("node").connect(node1.connector("deps"))
# Create NS3 testbed running in node1
- ns3_provider = FactoriesProvider(ns3_testbed_id, ns3_testbed_version)
+ ns3_provider = FactoriesProvider(ns3_testbed_id)
ns3_desc = exp.add_testbed_description(ns3_provider)
ns3_desc.set_attribute_value("rootDirectory", root)
ns3_desc.set_attribute_value("SimulatorImplementationType", "ns3::RealtimeSimulatorImpl")
"Test is expensive, requires NEPI_FULL_TESTS=yes")
def test_ns3_in_pl(self):
ns3_testbed_id = "ns3"
- ns3_testbed_version = "3_9"
pl, exp = self.make_experiment_desc()
git://git.onelab.eu / nepi.git / commitdiff