def setUp(self):
self.root_dir = tempfile.mkdtemp()
- def _test_if(self, daemonize_testbed, controller_access_configuration):
+ def _test_switched(self, controller_access_config = None,
+ testbed_access_config = None):
exp_desc = ExperimentDescription()
testbed_id = "netns"
user = getpass.getuser()
app.connector("node").connect(node1.connector("apps"))
app.enable_trace("stdout")
- if daemonize_testbed:
- netns_desc.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
- inst_root_dir = os.path.join(self.root_dir, "instance")
- os.mkdir(inst_root_dir)
- netns_desc.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
- netns_desc.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ if testbed_access_config:
+ for attr in testbed_access_config.attributes:
+ if attr.value:
+ netns_desc.set_attribute_value(attr.name, attr.value)
xml = exp_desc.to_xml()
- if controller_access_configuration:
- controller = proxy.create_experiment_controller(xml,
- controller_access_configuration)
- else:
- controller = ExperimentController(xml, self.root_dir)
+ controller = proxy.create_experiment_controller(xml,
+ controller_access_config)
try:
controller.start()
controller.shutdown()
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
- def test_local_if(self):
- self._test_if(
- daemonize_testbed = False,
- controller_access_configuration = None)
+ def test_switched(self):
+ self._test_switched()
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
- def test_all_daemonized_controller(self):
+ def test_daemonized_controller(self):
access_config = proxy.AccessConfiguration()
access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
- access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
- self._test_if(
- daemonize_testbed = False,
- controller_access_configuration = access_config)
+ self._test_switched(controller_access_config = access_config)
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
- def test_all_daemonized_if(self):
+ def test_daemonized_tbd(self):
access_config = proxy.AccessConfiguration()
access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
- access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ inst_root_dir = os.path.join(self.root_dir, "instance")
+ os.mkdir(inst_root_dir)
+ access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
- self._test_if(
- daemonize_testbed = True,
- controller_access_configuration = access_config)
+ self._test_switched(testbed_access_config = access_config)
@test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
- def test_all_ssh_daemonized_if(self):
+ def test_daemonized_all(self):
+ controller_access_config = proxy.AccessConfiguration()
+ controller_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
+ controller_access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
+ controller_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+
+ testbed_access_config = proxy.AccessConfiguration()
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
+ testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ inst_root_dir = os.path.join(self.root_dir, "instance")
+ os.mkdir(inst_root_dir)
+ testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
+
+ self._test_switched(
+ controller_access_config = controller_access_config,
+ testbed_access_config = testbed_access_config)
+
+ @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
+ def test_ssh_daemonized_tbd(self):
env = test_util.test_environment()
+
+ testbed_access_config = proxy.AccessConfiguration()
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
+ testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
+ testbed_access_config.set_attribute_value(DC.USE_AGENT, True)
+ inst_root_dir = os.path.join(self.root_dir, "instance")
+ os.mkdir(inst_root_dir)
+ testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
- access_config = proxy.AccessConfiguration()
- access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
- access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
- access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
- access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
- access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
- access_config.set_attribute_value(DC.USE_AGENT, True)
+ self._test_switched(
+ testbed_access_config = testbed_access_config)
+
+ def test_sudo_daemonized_tbd(self):
+ env = test_util.test_environment()
+
+ testbed_access_config = proxy.AccessConfiguration()
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
+ testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
+ testbed_access_config.set_attribute_value(DC.USE_AGENT, True)
+ testbed_access_config.set_attribute_value(DC.USE_SUDO, True)
+ inst_root_dir = os.path.join(self.root_dir, "instance")
+ os.mkdir(inst_root_dir)
+ testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
- self._test_if(
- daemonize_testbed = True,
- controller_access_configuration = access_config)
+ self._test_switched(
+ testbed_access_config = testbed_access_config)
+
+ @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
+ def test_ssh_daemonized_all(self):
+ env = test_util.test_environment()
+
+ controller_access_config = proxy.AccessConfiguration()
+ controller_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
+ controller_access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
+ controller_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ controller_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
+ controller_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
+ controller_access_config.set_attribute_value(DC.USE_AGENT, True)
+
+ testbed_access_config = proxy.AccessConfiguration()
+ testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
+ testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
+ # BUG! IT DOESN'T WORK WITH 2 LEVELS OF SSH!
+ #testbed_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
+ #testbed_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
+ #testbed_access_config.set_attribute_value(DC.USE_AGENT, True)
+ inst_root_dir = os.path.join(self.root_dir, "instance")
+ os.mkdir(inst_root_dir)
+ testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
+
+ self._test_switched(
+ controller_access_config = controller_access_config,
+ testbed_access_config = testbed_access_config)
def tearDown(self):
try:
- shutil.rmtree(self.root_dir)
+ #shutil.rmtree(self.root_dir)
+ pass
except:
# retry
time.sleep(0.1)