-#!/usr/bin/env python
# -*- coding: utf-8 -*-
class Attribute(object):
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.core.attributes import Attribute, AttributesMap
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.core.attributes import AttributesMap, Attribute
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.core.attributes import Attribute, AttributesMap
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.core import execute
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_ID
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
TESTBED_ID = "netns"
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from factories_metadata import wifi_standards, l4_protocols, \
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
TESTBED_ID = "ns3"
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.core import testbed_impl
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.util import tags
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
traces = dict({
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
# TODO: Allow netrefs!
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
TESTBED_ID = "planetlab"
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
import itertools
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
import weakref
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
AF_INET = 0
-# vim:ts=4:sw=4:et:ai:sts=4
import os, subprocess, os.path
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
-# vim:ts=4:sw=4:et:ai:sts=4
class GraphicalInfo(object):
""" This class allows to describe the position and dimensions of a
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
class GuidGenerator(object):
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
+
import struct
import random
import socket
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
import threading
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.core.attributes import Attribute
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
import base64
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.util.constants import DeploymentConfiguration as DC
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
MOBILE = "mobile"
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
import ipaddr
class DesignTestCase(unittest.TestCase):
def setUp(self):
+ # hack to add the mock testbed on the correct module path
sys.modules["nepi.testbeds.mock.metadata"] = mock.metadata
def test_design(self):
#!/usr/bin/env python
# -*- coding: utf-8 -*-
+import getpass
from nepi.util import tags
from nepi.util.constants import ApplicationStatus as AS
-import getpass
import mock
import mock.metadata
import sys
#!/usr/bin/env python
# -*- coding: utf-8 -*-
+import getpass
from nepi.core.design import ExperimentDescription, FactoriesProvider
from nepi.util import proxy
from nepi.util.constants import DeploymentConfiguration as DC
-import getpass
import mock
import mock.metadata
import mock2
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
TESTBED_ID = "mock"
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
TESTBED_ID = "mock2"
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from constants import TESTBED_ID, TESTBED_VERSION
-#!/usr/bin/env python
-
import nepi.util.environ
import ctypes
import imp
class NetnsDesignTestCase(unittest.TestCase):
def test_design_if(self):
- exp_desc = ExperimentDescription()
testbed_id = "netns"
+ exp_desc = ExperimentDescription()
provider = FactoriesProvider(testbed_id)
+
tstbd_desc = exp_desc.add_testbed_description(provider)
tstbd_desc.set_attribute_value("enableDebug", True)
node1 = tstbd_desc.create("Node")
app = tstbd_desc.create("Application")
app.set_attribute_value("command", "ping -qc10 10.0.0.2")
app.connector("node").connect(node1.connector("apps"))
+
xml = exp_desc.to_xml()
exp_desc2 = ExperimentDescription()
exp_desc2.from_xml(xml)
instance.do_configure()
instance.do_prestart()
instance.start()
+
while instance.status(7) != AS.STATUS_FINISHED:
time.sleep(0.5)
ping_result = instance.trace(7, "stdout")
instance.do_configure()
instance.do_prestart()
instance.start()
+
while instance.status(6) != AS.STATUS_FINISHED:
time.sleep(0.5)
ping_result = instance.trace(6, "stdout")
instance.do_configure()
instance.do_prestart()
instance.start()
+
while instance.status(11) != AS.STATUS_FINISHED:
time.sleep(0.5)
ping_result = instance.trace(11, "stdout")
instance.do_configure()
instance.do_prestart()
instance.start()
+
while instance.status(6) != AS.STATUS_FINISHED:
time.sleep(0.5)
pcap_result = instance.trace(2, "pcap")
def _test_switched(self, controller_access_config = None,
testbed_access_config = None):
- exp_desc = ExperimentDescription()
testbed_id = "netns"
+ exp_desc = ExperimentDescription()
user = getpass.getuser()
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_id = "ns3"
+ exp_desc = ExperimentDescription()
provider = FactoriesProvider(testbed_id)
tstbd_desc = exp_desc.add_testbed_description(provider)
instance.do_configure()
instance.do_prestart()
instance.start()
+
while instance.status(17) != AS.STATUS_FINISHED:
time.sleep(0.1)
ping_result = instance.trace(14, "P2PAsciiTrace")
instance.do_configure()
instance.do_prestart()
instance.start()
+
while instance.status(27) != AS.STATUS_FINISHED:
time.sleep(0.1)
ping_result = instance.trace(24, "P2PAsciiTrace")
def _test_fd_net_device(self, daemonize_testbed,
controller_access_configuration):
- exp_desc = ExperimentDescription()
testbed_id = "ns3"
+ exp_desc = ExperimentDescription()
ns3_provider = FactoriesProvider(testbed_id)
ns3_desc1 = exp_desc.add_testbed_description(ns3_provider)
root_dir1 = os.path.join(self.root_dir, "1")
class PlanetlabDesignTestCase(unittest.TestCase):
def make_test_design(self):
- exp_desc = ExperimentDescription()
testbed_id = "planetlab"
+ exp_desc = ExperimentDescription()
provider = FactoriesProvider(testbed_id)
tstbd_desc = exp_desc.add_testbed_description(provider)
tstbd_desc.set_attribute_value("slice", "inria_nepi")
from nepi.util.constants import ApplicationStatus as AS
from nepi.testbeds import planetlab
import os
+import re
import shutil
+import sys
import tempfile
+import test_util
import time
import unittest
-import re
-import test_util
-import sys
class PlanetLabExecuteTestCase(unittest.TestCase):
testbed_id = "planetlab"
# -*- coding: utf-8 -*-
import getpass
+import logging
from nepi.core.design import ExperimentDescription, FactoriesProvider
from nepi.core.execute import ExperimentController
from nepi.util import proxy
from nepi.util.constants import DeploymentConfiguration as DC
import os
+import re
import shutil
+import sys
import tempfile
import test_util
import time
import unittest
-import re
-import sys
-import logging
class PlanetLabIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"
from nepi.util import proxy
from nepi.util.constants import DeploymentConfiguration as DC, ATTR_NEPI_TESTBED_ENVIRONMENT_SETUP
import os
+import re
import shutil
import tempfile
import test_util
import time
import unittest
-import re
class PlanetLabMultiIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"
from nepi.util import proxy
from nepi.util.constants import DeploymentConfiguration as DC, ATTR_NEPI_TESTBED_ENVIRONMENT_SETUP
import os
+import re
import shutil
import tempfile
import test_util
import time
import unittest
-import re
class PlanetLabMultiIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"
from nepi.util import proxy
from nepi.util.constants import DeploymentConfiguration as DC, ATTR_NEPI_TESTBED_ENVIRONMENT_SETUP
import os
+import re
import shutil
import tempfile
import test_util
import time
import unittest
-import re
class PlanetLabCrossIntegrationTestCase(unittest.TestCase):
testbed_id = "planetlab"