if not self._server_process:
self._spawn_server_process()
self._client_process = self._spawn_nc_in_namespace(
- self.client_namespace.namespace,
+ self.client_namespace,
address=self.client_address)
return self._client_process
def _spawn_server_process(self):
self._server_process = self._spawn_nc_in_namespace(
- self.server_namespace.namespace,
+ self.server_namespace,
address=self.server_address,
listen=True)
import testtools
-from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_manager
from neutron.agent.linux import utils
from neutron.tests import base
def _test_with_nc(self, fw_manager, direction, port, udp):
netcat = helpers.NetcatTester(
- ip_lib.IPWrapper(self.client.namespace),
- ip_lib.IPWrapper(self.server.namespace),
+ self.client.namespace, self.server.namespace,
self.server.ip, self.port, run_as_root=True, udp=udp)
self.addCleanup(netcat.stop_processes)
protocol = 'tcp'
router.process(self.agent)
router_ns = ip_lib.IPWrapper(namespace=router.ns_name)
- netcat = helpers.NetcatTester(router_ns, router_ns,
+ netcat = helpers.NetcatTester(router.ns_name, router.ns_name,
server_address, port,
client_address=client_address,
run_as_root=True,
protocol_port = helpers.get_free_namespace_port(dst_machine.namespace)
# client sends to fip
netcat = helpers.NetcatTester(
- ip_lib.IPWrapper(src_machine.namespace),
- ip_lib.IPWrapper(dst_machine.namespace),
+ src_machine.namespace, dst_machine.namespace,
dst_machine.ip, protocol_port, client_address=dst_fip,
run_as_root=True, udp=False)
self.addCleanup(netcat.stop_processes)