#! /usr/bin/python3 # SPDX-License-Identifier: GPL-2.0-or-later # # tasst - Test A Simple Socket Transport # library of test helpers for passt & pasta # # tasst/transfer.py - Helpers for testing data transfers # # Copyright Red Hat # Author: David Gibson import ipaddress import time import avocado from tasst import Tasst, TasstSubData from tasst.address import LOOPBACK4, LOOPBACK6 from tasst.site import Site from tasst.nstool import UnshareSite # HACK: how long to wait for the server to be ready and listening (s) SERVER_READY_DELAY = 0.05 # 1/20th of a second # socat needs IPv6 addresses in square brackets def socat_fmt(ip): if isinstance(ip, ipaddress.IPv6Address): return '[{}]'.format(ip) elif isinstance(ip, ipaddress.IPv4Address): return '{}'.format(ip) else: raise TypeError class BaseTransferTasst(Tasst): def socat_upload(self, datafile, cs, ss, connect, listen): server = ss.bg('socat -u {} STDOUT'.format(listen), verbose=False) time.sleep(SERVER_READY_DELAY) cs.fg('socat -u OPEN:{} {}'.format(datafile, connect)) res = server.run() self.assertEquals(res.exit_status, 0) srcdata = cs.output('cat {}'.format(datafile), verbose=False) self.assertEquals(srcdata, res.stdout) def socat_download(self, datafile, cs, ss, connect, listen): server = ss.bg('socat -u OPEN:{} {}'.format(datafile, listen)) time.sleep(SERVER_READY_DELAY) dstdata = cs.output('socat -u {} STDOUT'.format(connect), verbose=False) res = server.run() self.assertEquals(res.exit_status, 0) srcdata = ss.output('cat {}'.format(datafile), verbose=False) self.assertEquals(srcdata, dstdata) def _tcp_socat(self, datafile, connectip, connectport, listenip, listenport, fromip): v6 = isinstance(connectip, ipaddress.IPv6Address) if listenport is None: listenport = connectport if v6: connect = 'TCP6:[{}]:{},ipv6only'.format(connectip, connectport) listen = 'TCP6-LISTEN:{},ipv6only'.format(listenport) else: connect = 'TCP4:{}:{}'.format(connectip, connectport) listen = 'TCP4-LISTEN:{}'.format(listenport) if listenip is not None: listen += ',bind=' + socat_fmt(listenip) if fromip is not None: connect += ',bind=' + socat_fmt(fromip) return (connect, listen) def tcp_upload(self, datafile, cs, ss, connectip, connectport, listenip=None, listenport=None, fromip=None): connect, listen = self._tcp_socat(datafile, connectip, connectport, listenip, listenport, fromip) self.socat_upload(datafile, cs, ss, connect, listen) def tcp_download(self, datafile, cs, ss, connectip, connectport, listenip=None, listenport=None, fromip=None): connect, listen = self._tcp_socat(datafile, connectip, connectport, listenip, listenport, fromip) self.socat_download(datafile, cs, ss, connect, listen) def udp_transfer(self, datafile, cs, ss, connectip, connectport, listenip=None, listenport=None, fromip=None): v6 = isinstance(connectip, ipaddress.IPv6Address) if listenport is None: listenport = connectport if v6: connect = 'UDP6:[{}]:{},ipv6only,shut-null'.format(connectip, connectport) listen = 'UDP6-LISTEN:{},ipv6only,null-eof'.format(listenport) if listenip is not None: assert isinstance(listenip, ipaddress.IPv6Address) listen += ',bind=[{}]'.format(listenip) else: connect = 'UDP4:{}:{},shut-null'.format(connectip, connectport) listen = 'UDP4-LISTEN:{},null-eof'.format(listenport) if listenip is not None: assert isinstance(listenip, ipaddress.IPv4Address) listen += ',bind={}'.format(listenip) self.socat_upload(datafile, cs, ss, connect, listen) def subsetup(self, datafile, cs, ss, ip4, ip6, port, listen_ip4=None, listen_ip6=None, listenport=None, from_ip4=None, from_ip6=None): cs.require_cmds('socat', 'cat') ss.require_cmds('socat', 'cat') Tasst.subsetup(self, BaseTransferTasst, TasstSubData(datafile=datafile, cs=cs, ss=ss, ip4=ip4, ip6=ip6, port=port, listen_ip4=listen_ip4, listen_ip6=listen_ip6, listenport=listenport, from_ip4=from_ip4, from_ip6=from_ip6)) class TcpUploadTasst(BaseTransferTasst): """ :avocado: disable """ def test_tcp4_upload(self): sub = self.get_subsetup(BaseTransferTasst) self.tcp_upload(sub.datafile, sub.cs, sub.ss, sub.ip4, sub.port, listenip=sub.listen_ip4, listenport=sub.listenport, fromip=sub.from_ip4) def test_tcp6_upload(self): sub = self.get_subsetup(BaseTransferTasst) self.tcp_upload(sub.datafile, sub.cs, sub.ss, sub.ip6, sub.port, listenip=sub.listen_ip6, listenport=sub.listenport, fromip=sub.from_ip6) class MetaTcpUploadTasst(TcpUploadTasst): """Ugly workaround for https://github.com/avocado-framework/avocado/issues/5680. Explicitly apply the "meta" tag to the tests in TransferTasst. :avocado: disable :avocado: tags=meta """ def test_tcp4_upload(self): super().test_tcp4_upload() def test_tcp6_upload(self): super().test_tcp6_upload() class UdpTransferTasst(BaseTransferTasst): """ :avocado: disable """ def test_udp4_transfer(self): sub = self.get_subsetup(BaseTransferTasst) self.udp_transfer(sub.datafile, sub.cs, sub.ss, sub.ip4, sub.port, listenip=sub.listen_ip4, listenport=sub.listenport, fromip=sub.from_ip4) def test_udp6_transfer(self): sub = self.get_subsetup(BaseTransferTasst) self.udp_transfer(sub.datafile, sub.cs, sub.ss, sub.ip6, sub.port, listenip=sub.listen_ip6, listenport=sub.listenport, fromip=sub.from_ip6) class MetaUdpTransferTasst(UdpTransferTasst): """Ugly workaround for https://github.com/avocado-framework/avocado/issues/5680. Explicitly apply the "meta" tag to the tests in TransferTasst. :avocado: disable :avocado: tags=meta """ def test_udp4_transfer(self): super().test_udp4_transfer() def test_udp6_transfer(self): super().test_udp6_transfer() LOOPBACK4 = ipaddress.ip_address('127.0.0.1') LOOPBACK6 = ipaddress.ip_address('::1') class BaseLocalTransferTasst(MetaTcpUploadTasst, MetaUdpTransferTasst): """Test the transfer helpers :avocado: disable """ PORT = 10000 def setUp(self): super().setUp() self.ns = UnshareSite(type(self).__name__ + '.netns', '-Un') self.ns.ifup('lo') def tearDown(self): self.ns.close() super().tearDown() class LocalTransferTasst(BaseLocalTransferTasst): """Test the transfer helpers """ def setUp(self): super().setUp() BaseTransferTasst.subsetup(self, 'test/small.bin', self.ns, self.ns, LOOPBACK4, LOOPBACK6, self.PORT) class LocalTransferTasstOptions(BaseLocalTransferTasst): def setUp(self): super().setUp() BaseTransferTasst.subsetup(self, 'test/small.bin', self.ns, self.ns, LOOPBACK4, LOOPBACK6, self.PORT, listen_ip4=LOOPBACK4, listen_ip6=LOOPBACK6, listenport=self.PORT, from_ip4=LOOPBACK4, from_ip6=LOOPBACK6)