+++ /dev/null
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-# vim:ts=4:sw=4:et:ai:sts=4
-
-#TODO: DEF ERRORCODES
-#TODO: DEF PROTOCOL
-
-class Controller(object):
- def __init__(self):
- self._testbeds = dict()
-
- def create_testbed(self, guid, testbed_id, config, access_config = None):
- # TODO: proxy
- # guid: guid of the associated backend
- self._testbeds[guid] = testbed
-
- def destroy_testbed(self, guid):
- tesbed = self._testbeds[guid]
- tesbed.shutdown()
- del self._testbeds[guid]
-
- def forward(self, guid, instruction):
- #TODO:
- pass
-
- def forward_batch(self, guid, batch):
- raise NotImplementedError
-
- def start(self):
- raise NotImplementedError
-
- def stop(self):
- raise NotImplementedError
-
- def status(self):
- raise NotImplementedError
-
- def shutdown(self):
- raise NotImplementedError
-
# connectors -- list of available connectors for the box
self._connectors = dict()
# factory_attributes -- factory attributes for box construction
- self._factory_attributes = list()
+ self._factory_attributes = dict()
# graphical_info -- GUI position information
self.graphical_info = GraphicalInfo(str(self._guid))
attr.range, attr.allowed, attr.readonly, attr.visible,
attr.validation_function)
for attr in factory.attributes:
- self._factory_attributes.append(attr)
+ if attr.modified:
+ self._factory_attributes[attr.name] = attr.value
@property
def guid(self):
def traces(self):
return self._traces.values()
+ @property
+ def traces_name(self):
+ return self._traces.keys()
+
@property
def factory_attributes(self):
return self._factory_attributes
def routes(self):
return []
+ def trace_help(self, trace_id):
+ return self._traces[trace_id].help
+
+ def enable_trace(self, trace_id):
+ self._traces[trace_id].enabled = True
+
+ def disable_trace(self, trace_id):
+ self._traces[trace_id].enabled = False
+
def connector(self, name):
return self._connectors[name]
- def trace(self, trace_id):
- return self._traces[trace_id]
-
def destroy(self):
super(Box, self).destroy()
for c in self.connectors:
self._boxes = dict()
self.graphical_info = GraphicalInfo(str(self._guid))
+ metadata = Metadata(provider.testbed_id, provider.testbed_version)
+ for attr in metadata.testbed_attributes().attributes:
+ self.add_attribute(attr.name, attr.help, attr.type, attr.value,
+ attr.range, attr.allowed, attr.readonly, attr.visible,
+ attr.validation_function)
+
@property
def guid(self):
return self._guid
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-from nepi.core.attributes import Attribute, AttributesMap
+from nepi.core.attributes import AttributesMap
+from nepi.util.parser._xml import XmlExperimentParser
from nepi.util import validation
import sys
def add_trace(self, trace_id):
self._traces.append(trace_id)
-class TestbedConfiguration(AttributesMap):
- def __init__(self):
- super(TestbedConfiguration, self).__init__()
- self.add_attribute("HomeDirectory",
- "Path to the local directory where traces and other files \
- will be stored",
- Attribute.STRING, False, None, None, "",
- validation.is_string)
-
class TestbedInstance(object):
- def __init__(self, testbed_id, testbed_version, configuration):
+ def __init__(self, testbed_id, testbed_version):
self._testbed_id = testbed_id
self._testbed_version = testbed_version
- self._configuration = configuration
- self._home_directory = configuration.get_attribute_value(
- "HomeDirectory")
@property
- def home_directory(self):
- return self._home_directory
+ def guids(self):
+ raise NotImplementedError
+
+ def configure(self, name, value):
+ """Set a configuartion attribute for the testbed instance"""
+ raise NotImplementedError
def create(self, guid, factory_id):
"""Instructs creation of element """
"""Instructs setting an attribute on an element"""
raise NotImplementedError
- def do_create(self):
- """After do_create all instructed elements are created and
- attributes setted"""
- raise NotImplementedError
-
def connect(self, guid1, connector_type_name1, guid2,
connector_type_name2):
raise NotImplementedError
cross_testbed_id, cross_factory_id, cross_connector_type_name):
raise NotImplementedError
- def do_connect(self):
- raise NotImplementedError
-
def add_trace(self, guid, trace_id):
raise NotImplementedError
def add_route(self, guid, destination, netprefix, nexthop):
raise NotImplementedError
+ def do_setup(self):
+ """After do_setup the testbed initial configuration is done"""
+ raise NotImplementedError
+
+ def do_create(self):
+ """After do_create all instructed elements are created and
+ attributes setted"""
+ raise NotImplementedError
+
+ def do_connect(self):
+ """After do_connect all internal connections between testbed elements
+ are done"""
+ raise NotImplementedError
+
def do_configure(self):
+ """After do_configure elements are configured"""
raise NotImplementedError
def do_cross_connect(self):
+ """After do_cross_connect all external connections between different testbed
+ elements are done"""
raise NotImplementedError
- def set(self, time, guid, name, value):
+ def start(self, time):
raise NotImplementedError
- def get(self, time, guid, name):
+ def stop(self, time):
raise NotImplementedError
- def start(self, time):
+ def set(self, time, guid, name, value):
raise NotImplementedError
- def action(self, time, guid, action):
+ def get(self, time, guid, name):
raise NotImplementedError
- def stop(self, time):
+ def action(self, time, guid, action):
raise NotImplementedError
def status(self, guid):
def shutdown(self):
raise NotImplementedError
+class ExperimentController(object):
+ def __init__(self, experiment_xml):
+ self._experiment_xml = experiment_xml
+ self._testbeds = dict()
+ self._access_config = dict()
+
+ @property
+ def experiment_xml(self):
+ return self._experiment_xml
+
+ def testbed_instance(self, guid):
+ return self._testbeds[guid]
+
+ def set_testbed_access_config(self, guid, access_config):
+ self._access_config[guid] = access_config
+
+ def trace(self, testbed_guid, guid, trace_id):
+ return self._testbeds[testbed_guid].trace(guid, trace_id)
+
+ def start(self):
+ parser = XmlExperimentParser()
+ data = parser.from_xml_to_data(self._experiment_xml)
+ element_guids = list()
+ for guid in data.guids:
+ if data.is_testbed_data(guid):
+ (testbed_id, testbed_version) = data.get_testbed_data(guid)
+ instance = self._build_testbed_instance(testbed_id,
+ testbed_version)
+ for (name, value) in data.get_attribute_data(guid):
+ instance.configure(name, value)
+ self._testbeds[guid] = instance
+ else:
+ element_guids.append(guid)
+
+ for guid in element_guids:
+ (testbed_guid, factory_id) = data.get_box_data(guid)
+ instance = self._testbeds[testbed_guid]
+ instance.create(guid, factory_id)
+ for (name, value) in data.get_attribute_data(guid):
+ instance.create_set(guid, name, value)
+
+ for guid in element_guids:
+ (testbed_guid, factory_id) = data.get_box_data(guid)
+ instance = self._testbeds[testbed_guid]
+ for (connector_type_name, other_guid, other_connector_type_name) \
+ in data.get_connection_data(guid):
+ (testbed_guid, factory_id) = data.get_box_data(guid)
+ (other_testbed_guid, other_factory_id) = data.get_box_data(
+ other_guid)
+ if testbed_guid == other_testbed_guid:
+ instance.connect(guid, connector_type_name, other_guid,
+ other_connector_type_name)
+ else:
+ instance.cross_connect(guid, connector_type_name, other_guid,
+ other_testbed_id, other_factory_id, other_connector_type_name)
+ for trace_id in data.get_trace_data(guid):
+ instance.add_trace(guid, trace_id)
+ for (autoconf, address, family, netprefix, broadcast) in \
+ data.get_address_data(guid):
+ if address != None:
+ # TODO: BUG!!! Hardcodeado!!!!!! XXXXXXXXX CORREGIR!!!
+ family = 0
+ netprefix = 24
+ instance.add_adddress(guid, family, address, netprefix,
+ broadcast)
+ for (family, destination, netprefix, nexthop) in \
+ data.get_route_data(guid):
+ instance.add_route(guid, destination, netprefix, nexthop)
+
+ for instance in self._testbeds.values():
+ instance.do_setup()
+ for instance in self._testbeds.values():
+ instance.do_create()
+ instance.do_connect()
+ instance.do_configure()
+ for instances in self._testbeds.values():
+ instance.do_cross_connect()
+ for instances in self._testbeds.values():
+ instance.start()
+
+ def stop(self):
+ for instance in self._testbeds.values():
+ instance.stop()
+
+ def status(self, guid):
+ for instance in self._testbeds.values():
+ for guid_ in instance.guids:
+ if guid_ == guid:
+ return instance.status(guid)
+ raise RuntimeError("No element exists with guid %d" % guid)
+
+ def shutdown(self):
+ for instance in self._testbeds.values():
+ instance.shutdown()
+
+ def _build_testbed_instance(self, testbed_id, testbed_version):
+ mod_name = "nepi.testbeds.%s" % (testbed_id.lower())
+ if not mod_name in sys.modules:
+ __import__(mod_name)
+ module = sys.modules[mod_name]
+ return module.TestbedInstance(testbed_version)
+
#!/usr/bin/env python
# -*- coding: utf-8 -*-
+from nepi.core.attributes import AttributesMap
import sys
class VersionedMetadataInfo(object):
@property
def attributes(self):
""" dictionary of dictionaries of all available attributes.
- "attribute_id": dict({
+ attribute_id: dict({
"name": attribute name,
"help": help text,
"type": attribute type,
})
"""
raise NotImplementedError
-
+
+ @property
+ def testbed_attributes(self):
+ """ dictionary of attributes for testbed instance configuration
+ attributes_id = dict({
+ "name": attribute name,
+ "help": help text,
+ "type": attribute type,
+ "value": default attribute value,
+ "range": (maximum, minimun) values else None if not defined,
+ "allowed": array of posible values,
+ "readonly": whether the attribute is read only for the user,
+ "visible": whether the attribute is visible for the user,
+ "validation_function": validation function for the attribute
+ })
+ ]
+ """
+ raise NotImplementedError
+
class Metadata(object):
def __init__(self, testbed_id, version):
self._version = version
def factories_order(self):
return self._metadata.factories_order
+ def testbed_attributes(self):
+ attributes = AttributesMap()
+ for attribute_info in self._metadata.testbed_attributes.values():
+ name = attribute_info["name"]
+ help = attribute_info["help"]
+ type = attribute_info["type"]
+ value = attribute_info["value"]
+ range = attribute_info["range"]
+ allowed = attribute_info["allowed"]
+ readonly = attribute_info["readonly"]
+ visible = attribute_info["visible"]
+ validation_function = attribute_info["validation_function"]
+ attributes.add_attribute(name, help, type, value,
+ range, allowed, readonly, visible, validation_function)
+ return attributes
+
def build_design_factories(self):
from nepi.core.design import Factory
factories = list()
# -*- coding: utf-8 -*-
from nepi.core import execute
-from nepi.core.attributes import Attribute
from nepi.core.metadata import Metadata
from nepi.util import validation
from nepi.util.constants import AF_INET, AF_INET6, STATUS_UNDETERMINED
TIME_NOW = "0s"
class TestbedInstance(execute.TestbedInstance):
- def __init__(self, testbed_id, testbed_version, configuration):
- super(TestbedInstance, self).__init__(testbed_id, testbed_version,
- configuration)
+ def __init__(self, testbed_id, testbed_version):
+ super(TestbedInstance, self).__init__(testbed_id, testbed_version)
+ # testbed attributes for validation
+ self._attributes = None
+ # element factories for validation
self._factories = dict()
- self._elements = dict()
+
+ # experiment construction instructions
self._create = dict()
- self._set = dict()
+ self._create_set = dict()
self._connect = dict()
self._cross_connect = dict()
self._add_trace = dict()
self._add_address = dict()
- self._add_route = dict()
+ self._add_route = dict()
+ self._configure = dict()
+
+ # log of set operations
+ self._set = dict()
+ # log of actions
+ self._actions = dict()
+
+ # testbed element instances
+ self._elements = dict()
self._metadata = Metadata(self._testbed_id, self._testbed_version)
for factory in self._metadata.build_execute_factories():
self._factories[factory.factory_id] = factory
+ self._attributes = self._metadata.testbed_attributes()
+
+ @property
+ def guids(self):
+ return self._create.keys()
@property
def elements(self):
return self._elements
+ def configure(self, name, value):
+ if not self._attributes.has_attribute(name):
+ raise RuntimeError("Invalid attribute %s for testbed" % name)
+ # Validation
+ self._attributes.set_attribute_value(name, value)
+ self._configure[name] = value
+
def create(self, guid, factory_id):
if factory_id not in self._factories:
raise RuntimeError("Invalid element type %s for Netns version %s" %
raise RuntimeError("Invalid attribute %s for element type %s" %
(name, factory_id))
factory.set_attribute_value(name, value)
- if guid not in self._set:
- self._set[guid] = dict()
- self._set[guid][name] = value
+ if guid not in self._create_set:
+ self._create_set[guid] = dict()
+ self._create_set[guid][name] = value
def connect(self, guid1, connector_type_name1, guid2,
connector_type_name2):
self._add_route[guid] = list()
self._add_route[guid].append((destination, netprefix, nexthop))
+ def do_setup(self):
+ raise NotImplementedError
+
def do_create(self):
guids = dict()
# order guids (elements) according to factory_id
continue
factory = self._factories[factory_id]
for guid in guids[factory_id]:
- parameters = dict() if guid not in self._set else \
- self._set[guid]
+ parameters = dict() if guid not in self._create_set else \
+ self._create_set[guid]
factory.create_function(self, guid, parameters)
for name, value in parameters.iteritems():
self.set(TIME_NOW, guid, name, value)
code_to_connect(element, cross_guid)
def set(self, time, guid, name, value):
- raise NotImplementedError
+ if not guid in self._create:
+ raise RuntimeError("Element guid %d doesn't exist" % guid)
+ factory_id = self._create[guid]
+ factory = self._factories[factory_id]
+ if not factory.has_attribute(name):
+ raise RuntimeError("Invalid attribute %s for element type %s" %
+ (name, factory_id))
+ factory.set_attribute_value(name, value)
+ if guid not in self._set:
+ self._set[guid] = dict()
+ if time not in self._set[guid]:
+ self._set[guid][time] = dict()
+ self._set[guid][time][name] = value
def get(self, time, guid, name):
raise NotImplementedError
if start_function:
traces = [] if guid not in self._add_trace else \
self._add_trace[guid]
- parameters = dict() if guid not in self._set else \
- self._set[guid]
+ parameters = dict() if guid not in self._create_set else \
+ self._create_set[guid]
start_function(self, guid, parameters, traces)
def action(self, time, guid, action):
cross_count = len(self._cross_connect[guid][connection_type_name])
return count + cross_count
-
# -*- coding: utf-8 -*-
from constants import TESTBED_ID
-from execute import TestbedConfiguration, TestbedInstance
+from execute import TestbedInstance
# -*- coding: utf-8 -*-
from constants import TESTBED_ID
-from nepi.core import execute
-from nepi.core import execute_impl
-from nepi.core.attributes import Attribute
+from nepi.core import testbed_impl
from nepi.core.metadata import Metadata
from nepi.util import validation
from nepi.util.constants import AF_INET, AF_INET6
import os
-class TestbedConfiguration(execute.TestbedConfiguration):
- def __init__(self):
- super(TestbedConfiguration, self).__init__()
- self.add_attribute("EnableDebug", "Enable netns debug output",
- Attribute.BOOL, False, None, None, False, validation.is_bool)
-
-class TestbedInstance(execute_impl.TestbedInstance):
- def __init__(self, testbed_version, configuration):
- super(TestbedInstance, self).__init__(TESTBED_ID, testbed_version,
- configuration)
- self._netns = self._load_netns_module(configuration)
+class TestbedInstance(testbed_impl.TestbedInstance):
+ def __init__(self, testbed_version):
+ super(TestbedInstance, self).__init__(TESTBED_ID, testbed_version)
+ self._netns = None
+ self._home_directory = None
self._traces = dict()
+ @property
+ def home_directory(self):
+ return self._home_directory
+
@property
def netns(self):
return self._netns
+ def do_setup(self):
+ self._home_directory = self._attributes.\
+ get_attribute_value("homeDirectory")
+ self._netns = self._load_netns_module()
+
def do_configure(self):
# TODO: add traces!
# configure addressess
nexthop = nexthop)
def set(self, time, guid, name, value):
+ super(TestbedInstance, self).set(time, guid, name, value)
# TODO: take on account schedule time for the task
element = self._elements[guid]
if element:
def follow_trace(self, trace_id, trace):
self._traces[trace_id] = trace
- def _load_netns_module(self, configuration):
+ def _load_netns_module(self):
# TODO: Do something with the configuration!!!
import sys
__import__("netns")
netns_mod = sys.modules["netns"]
# enable debug
- enable_debug = configuration.get_attribute_value("EnableDebug")
+ enable_debug = self._attributes.get_attribute_value("enableDebug")
if enable_debug:
netns_mod.environ.set_log_level(netns_mod.environ.LOG_DEBUG)
return netns_mod
"visible": False,
"validation_function": validation.is_integer
}),
-
})
traces = dict({
}),
})
+testbed_attributes = dict({
+ "enable_debug": dict({
+ "name": "enableDebug",
+ "help": "Enable netns debug output",
+ "type": Attribute.BOOL,
+ "value": False,
+ "range": None,
+ "allowed": None,
+ "readonly": False,
+ "visible": True,
+ "validation_function": validation.is_bool
+ }),
+ "home_directory": dict({
+ "name": "homeDirectory",
+ "help": "Path to the directory where traces and other files \
+ will be stored",
+ "type": Attribute.STRING,
+ "value": False,
+ "range": None,
+ "allowed": None,
+ "readonly": False,
+ "visible": True,
+ "validation_function": validation.is_string
+ })
+ })
+
class VersionedMetadataInfo(metadata.VersionedMetadataInfo):
@property
def connections_types(self):
def factories_info(self):
return factories_info
+ @property
+ def testbed_attributes(self):
+ return testbed_attributes
+
STATUS_RUNNING = 1
STATUS_FINISHED = 2
STATUS_UNDETERMINED = 3
+
+
if connections_tag.hasChildNodes():
parent_tag.appendChild(connections_tag)
- def from_xml(self, experiment_description, xml):
+ def from_xml_to_data(self, xml):
data = ExperimentData()
doc = minidom.parseString(xml)
testbeds_tag = doc.getElementsByTagName("testbeds")[0]
for element_tag in element_tag_list:
if element_tag.nodeType == doc.ELEMENT_NODE:
self.box_data_from_xml(element_tag, testbed_guid, data)
+ return data
+
+ def from_xml(self, experiment_description, xml):
+ data = self.from_xml_to_data(xml)
self.from_data(experiment_description, data)
def testbed_data_from_xml(self, tag, data):
data = self.data[guid]
if not "traces" in data:
return []
- return [trace_name for trace_name in data["traces"]]
+ return [trace_id for trace_id in data["traces"]]
def get_connection_data(self, guid):
data = self.data[guid]
data.add_graphical_info_data(guid, g_info.x, g_info.y, g_info.width,
g_info.height, g_info.label)
- def factory_attributes_to_data(self, data, guid, attributes):
- for attribute in attributes:
- if attribute.modified:
- data.add_factory_attribute_data(guid, attribute.name,
- attribute.value)
+ def factory_attributes_to_data(self, data, guid, factory_attributes):
+ for name, value in factory_attributes.iteritems():
+ data.add_factory_attribute_data(guid, name, value)
def attributes_to_data(self, data, guid, attributes):
for attribute in attributes:
def traces_to_data(self, data, guid, traces):
for trace in traces:
if trace.enabled:
- data.add_trace_data(guid, trace.name)
+ data.add_trace_data(guid, trace.trace_id)
def connections_to_data(self, data, guid, connectors):
for connector in connectors:
testbed_id = "netns"
netns_provider = FactoriesProvider(testbed_id, testbed_version)
netns_desc = exp_desc.add_testbed_description(netns_provider)
+ netns_desc.set_attribute_value("enableDebug", True)
node1 = netns_desc.create("Node")
node2 = netns_desc.create("Node")
iface1 = netns_desc.create("NodeInterface")
def test_run_ping_if(self):
user = getpass.getuser()
testbed_version = "01"
- config = netns.TestbedConfiguration()
- config.set_attribute_value("HomeDirectory", self._home_dir)
- instance = netns.TestbedInstance(testbed_version, config)
+ instance = netns.TestbedInstance(testbed_version)
+ instance.configure("homeDirectory", self._home_dir)
instance.create(2, "Node")
instance.create(3, "Node")
instance.create(4, "NodeInterface")
instance.add_trace(7, "stdout")
instance.connect(7, "node", 2, "apps")
+ instance.do_setup()
instance.do_create()
instance.do_connect()
instance.do_configure()
def test_run_ping_p2pif(self):
user = getpass.getuser()
testbed_version = "01"
- config = netns.TestbedConfiguration()
- config.set_attribute_value("HomeDirectory", self._home_dir)
- instance = netns.TestbedInstance(testbed_version, config)
+ instance = netns.TestbedInstance(testbed_version)
+ instance.configure("homeDirectory", self._home_dir)
instance.create(2, "Node")
instance.create(3, "Node")
instance.create(4, "P2PNodeInterface")
instance.create_set(6, "user", user)
instance.add_trace(6, "stdout")
instance.connect(6, "node", 2, "apps")
+
+ instance.do_setup()
instance.do_create()
instance.do_connect()
instance.do_configure()
def test_run_ping_routing(self):
user = getpass.getuser()
testbed_version = "01"
- config = netns.TestbedConfiguration()
- config.set_attribute_value("HomeDirectory", self._home_dir)
- instance = netns.TestbedInstance(testbed_version, config)
+ instance = netns.TestbedInstance(testbed_version)
+ instance.configure("homeDirectory", self._home_dir)
instance.create(2, "Node")
instance.create(3, "Node")
instance.create(4, "Node")
instance.add_route(2, "10.0.1.0", 24, "10.0.0.2")
instance.add_route(4, "10.0.0.0", 24, "10.0.1.1")
+ instance.do_setup()
instance.do_create()
instance.do_connect()
instance.do_configure()
--- /dev/null
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import getpass
+from nepi.core.design import ExperimentDescription, FactoriesProvider
+from nepi.core.execute import ExperimentController
+from nepi.util.constants import AF_INET, STATUS_FINISHED
+import os
+import shutil
+import test_util
+import time
+import unittest
+import uuid
+
+class NetnsIntegrationTestCase(unittest.TestCase):
+ def setUp(self):
+ self._home_dir = os.path.join(os.getenv("HOME"), ".nepi",
+ str(uuid.uuid1()))
+ os.makedirs(self._home_dir)
+
+ @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_desc = exp_desc.add_testbed_description(netns_provider)
+ netns_desc.set_attribute_value("homeDirectory", self._home_dir)
+ #netns_desc.set_attribute_value("enableDebug", True)
+ node1 = netns_desc.create("Node")
+ node2 = netns_desc.create("Node")
+ iface1 = netns_desc.create("NodeInterface")
+ iface1.set_attribute_value("up", True)
+ node1.connector("devs").connect(iface1.connector("node"))
+ ip1 = iface1.add_address()
+ ip1.set_attribute_value("Address", "10.0.0.1")
+ iface2 = netns_desc.create("NodeInterface")
+ iface2.set_attribute_value("up", True)
+ node2.connector("devs").connect(iface2.connector("node"))
+ ip2 = iface2.add_address()
+ ip2.set_attribute_value("Address", "10.0.0.2")
+ switch = netns_desc.create("Switch")
+ switch.set_attribute_value("up", True)
+ iface1.connector("switch").connect(switch.connector("devs"))
+ iface2.connector("switch").connect(switch.connector("devs"))
+ app = netns_desc.create("Application")
+ app.set_attribute_value("command", "ping -qc1 10.0.0.2")
+ app.set_attribute_value("user", user)
+ app.connector("node").connect(node1.connector("apps"))
+ app.enable_trace("stdout")
+ xml = exp_desc.to_xml()
+
+ controller = ExperimentController(xml)
+ controller.start()
+ while controller.status(app.guid) != STATUS_FINISHED:
+ time.sleep(0.5)
+ ping_result = controller.trace(netns_desc.guid, app.guid, "stdout")
+ comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
+
+--- 10.0.0.2 ping statistics ---
+1 packets transmitted, 1 received, 0% packet loss, time 0ms
+"""
+ self.assertTrue(ping_result.startswith(comp_result))
+ controller.stop()
+ controller.shutdown()
+
+ def tearDown(self):
+ shutil.rmtree(self._home_dir)
+
+if __name__ == '__main__':
+ unittest.main()
+
git://git.onelab.eu / nepi.git / commitdiff