Best Python code snippet using autotest_python
net_utils_unittest.py
Source:net_utils_unittest.py  
...27    #28    # test network_util29    #30    def test_network_util_reset(self):31        utils.system.expect_call('service network restart', ignore_status=False)32        net_utils.network_utils().reset()33        self.god.check_playback()34    def test_network_util_start(self):35        utils.system.expect_call('service network start', ignore_status=False)36        net_utils.network_utils().start()37        self.god.check_playback()38    def test_network_util_stop(self):39        utils.system.expect_call('service network stop', ignore_status=False)40        net_utils.network_utils().stop()41        self.god.check_playback()42    def test_network_util_disable_ip_local_loopback(self):43        msg = "echo '1' > /proc/sys/net/ipv4/route/no_local_loopback"44        utils.system.expect_call(msg, ignore_status=False)45        msg = 'echo 1 > /proc/sys/net/ipv4/route/flush'46        utils.system.expect_call(msg, ignore_status=False)47        net_utils.network_utils().disable_ip_local_loopback()48        self.god.check_playback()49    def test_network_util_enable_ip_local_loopback(self):50        msg = "echo '0' > /proc/sys/net/ipv4/route/no_local_loopback"51        utils.system.expect_call(msg, ignore_status=False)52        msg = 'echo 1 > /proc/sys/net/ipv4/route/flush'53        utils.system.expect_call(msg, ignore_status=False)54        net_utils.network_utils().enable_ip_local_loopback()55        self.god.check_playback()56    #57    # test network_interface58    #59    def test_network_interface_init(self):60        self.god.stub_function(socket, 'socket')61        s = net_utils_mock.socket_stub('eth0', socket, socket)62        socket.socket.expect_call(socket.PF_PACKET,63                                  socket.SOCK_RAW).and_return(s)64        self.god.stub_function(s, 'bind')65        self.god.stub_function(s, 'settimeout')66        s.settimeout.expect_call(net_utils.TIMEOUT)67        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))68        mock_netif = self.network_interface_mock(iface='eth0', test_init=True)69        self.god.check_playback()70        self.assertEquals(mock_netif.ethtool, 'ethtool')71        self.assertEquals(mock_netif._name, 'eth0')72        self.assertEquals(mock_netif.was_down, 'is_down')73        self.assertEquals(mock_netif.orig_ipaddr, 'get_ipaddr')74        self.assertEquals(mock_netif.was_loopback_enabled,75                          'is_loopback_enabled')76        self.assertEquals(mock_netif._socket, s)77    def test_network_interface_restore(self):78        mock_netif = self.network_interface_mock('eth0')79        mock_netif.was_loopback_enabled = False80        mock_netif.loopback_enabled = True81        mock_netif.was_down = False82        # restore using phyint83        cmd = 'ifconfig %s %s' % (mock_netif._name, mock_netif.orig_ipaddr)84        utils.system.expect_call(cmd)85        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,86                                  'phyint', 'disable')87        utils.system.expect_call(cmd, ignore_status=True).and_return(0)88        mock_netif.restore()89        self.god.check_playback()90        # restore using mac91        cmd = 'ifconfig %s %s' % (mock_netif._name, mock_netif.orig_ipaddr)92        utils.system.expect_call(cmd)93        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,94                                  'phyint', 'disable')95        utils.system.expect_call(cmd, ignore_status=True).and_return(1)96        cmd = '%s -L %s %s %s' % (mock_netif.ethtool,97                                  mock_netif._name, 'mac', 'disable')98        utils.system.expect_call(cmd, ignore_status=True).and_return(0)99        mock_netif.restore()100        self.god.check_playback()101        # check that down is restored102        mock_netif.was_loopback_enabled = False103        mock_netif.loopback_enabled = True104        mock_netif.was_down = True105        cmd = 'ifconfig %s %s' % (mock_netif._name, mock_netif.orig_ipaddr)106        utils.system.expect_call(cmd)107        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,108                                  'phyint', 'disable')109        utils.system.expect_call(cmd, ignore_status=True).and_return(0)110        cmd = 'ifconfig %s down' % mock_netif._name111        utils.system.expect_call(cmd)112        mock_netif.restore()113        self.god.check_playback()114        # check that loopback, down are done in sequence115        mock_netif.was_loopback_enabled = True116        mock_netif.loopback_enabled = True117        mock_netif.was_down = True118        cmd = 'ifconfig %s %s' % (mock_netif._name,119                                  mock_netif.orig_ipaddr)120        utils.system.expect_call(cmd)121        cmd = 'ifconfig %s down' % mock_netif._name122        utils.system.expect_call(cmd)123        mock_netif.restore()124        self.god.check_playback()125        # prior loopback matches current loopback126        mock_netif.was_loopback_enabled = False127        mock_netif.loopback_enabled = False128        mock_netif.was_down = True129        cmd = 'ifconfig %s %s' % (mock_netif._name,130                                  mock_netif.orig_ipaddr)131        utils.system.expect_call(cmd)132        cmd = 'ifconfig %s down' % mock_netif._name133        utils.system.expect_call(cmd)134        mock_netif.restore()135        self.god.check_playback()136    def test_network_interface_get_name(self):137        mock_netif = self.network_interface_mock(iface='eth0')138        self.assertEquals(mock_netif.get_name(), 'eth0')139    def test_network_interface_parse_ethtool(self):140        mock_netif = self.network_interface_mock()141        cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name)142        utils.system_output.expect_call(cmd).and_return('\n field: match')143        self.assertEquals(mock_netif.parse_ethtool('field', 'some|match'),144                          'match')145        self.god.check_playback()146        utils.system_output.expect_call(cmd).and_return(None)147        self.assertEquals(mock_netif.parse_ethtool('field',148                                                   'some|match'), '')149        utils.system_output.expect_call(cmd).and_return(' field: match')150        self.assertEquals(mock_netif.parse_ethtool('field',151                                                   'some|match'), '')152        self.god.check_playback()153    def test_network_interface_get_stats(self):154        mock_netif = self.network_interface_mock()155        self.god.stub_function(os, 'listdir')156        stat_path = '/sys/class/net/%s/statistics/' % mock_netif._name157        # no stat found158        os.listdir.expect_call(stat_path).and_return(())159        self.assertEquals(mock_netif.get_stats(), {})160        self.god.check_playback()161        # can not open stat file162        os.listdir.expect_call(stat_path).and_return(('some_stat',))163        f = self.god.create_mock_class(file, 'file')164        net_utils.open.expect_call(stat_path + 'some_stat', 'r').and_return(None)165        self.assertEquals(mock_netif.get_stats(), {})166        self.god.check_playback()167        # found a single stat168        os.listdir.expect_call(stat_path).and_return(('some_stat',))169        f = self.god.create_mock_class(file, 'file')170        net_utils.open.expect_call(stat_path + 'some_stat', 'r').and_return(f)171        f.read.expect_call().and_return(1234)172        f.close.expect_call()173        self.assertEquals(mock_netif.get_stats(), {'some_stat':1234})174        self.god.check_playback()175        # found multiple stats176        os.listdir.expect_call(stat_path).and_return(('stat1','stat2'))177        f = self.god.create_mock_class(file, 'file')178        net_utils.open.expect_call(stat_path + 'stat1', 'r').and_return(f)179        f.read.expect_call().and_return(1234)180        f.close.expect_call()181        net_utils.open.expect_call(stat_path + 'stat2', 'r').and_return(f)182        f.read.expect_call().and_return(5678)183        f.close.expect_call()184        self.assertEquals(mock_netif.get_stats(), {'stat1':1234, 'stat2':5678})185        self.god.check_playback()186    def test_network_interface_get_stats_diff(self):187        mock_netif = self.network_interface_mock()188        self.god.stub_function(os, 'listdir')189        stat_path = '/sys/class/net/%s/statistics/' % mock_netif._name190        os.listdir.expect_call(stat_path).and_return(('stat1','stat2', 'stat4'))191        f = self.god.create_mock_class(file, 'file')192        net_utils.open.expect_call(stat_path + 'stat1', 'r').and_return(f)193        f.read.expect_call().and_return(1234)194        f.close.expect_call()195        net_utils.open.expect_call(stat_path + 'stat2', 'r').and_return(f)196        f.read.expect_call().and_return(0)197        f.close.expect_call()198        net_utils.open.expect_call(stat_path + 'stat4', 'r').and_return(f)199        f.read.expect_call().and_return(10)200        f.close.expect_call()201        self.assertEquals(mock_netif.get_stats_diff({'stat1':1, 'stat2':2,202                                                     'stat3':0}),203                          {'stat1':1233, 'stat2':-2, 'stat4':10})204        self.god.check_playback()205    def test_network_interface_get_driver(self):206        mock_netif = self.network_interface_mock()207        mock_netif.get_driver = net_utils.network_interface.get_driver208        self.god.stub_function(os, 'readlink')209        stat_path = '/sys/class/net/%s/device/driver' % mock_netif._name210        os.readlink.expect_call(stat_path).and_return((211                                                  stat_path+'/driver_name'))212        self.assertEquals(mock_netif.get_driver(mock_netif), 'driver_name')213        self.god.check_playback()214    def test_network_interface_get_carrier(self):215        mock_netif = self.network_interface_mock()216        self.god.stub_function(os, 'readlink')217        stat_path = '/sys/class/net/%s/carrier' % mock_netif._name218        f = self.god.create_mock_class(file, 'file')219        net_utils.open.expect_call(stat_path).and_return(f)220        f.read.expect_call().and_return(' 1 ')221        f.close.expect_call()222        self.assertEquals(mock_netif.get_carrier(), '1')223        self.god.check_playback()224        net_utils.open.expect_call(stat_path).and_return(f)225        f.read.expect_call().and_return(' 0 ')226        f.close.expect_call()227        self.assertEquals(mock_netif.get_carrier(), '0')228        self.god.check_playback()229        net_utils.open.expect_call(stat_path).and_return(f)230        f.read.expect_call().and_return('')231        f.close.expect_call()232        self.assertEquals(mock_netif.get_carrier(), '')233        self.god.check_playback()234        net_utils.open.expect_call(stat_path).and_return(None)235        self.assertEquals(mock_netif.get_carrier(), '')236        self.god.check_playback()237    def test_network_interface_is_autoneg_advertised(self):238        mock_netif = self.network_interface_mock()239        cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name)240        utils.system_output.expect_call(cmd).and_return(241            '\n Advertised auto-negotiation: Yes')242        self.assertEquals(mock_netif.is_autoneg_advertised(), True)243        self.god.check_playback()244        utils.system_output.expect_call(cmd).and_return(245            '\n Advertised auto-negotiation: No')246        self.assertEquals(mock_netif.is_autoneg_advertised(), False)247        self.god.check_playback()248        utils.system_output.expect_call(cmd).and_return('')249        self.assertEquals(mock_netif.is_autoneg_advertised(), False)250        self.god.check_playback()251    def test_network_interface_get_speed(self):252        mock_netif = self.network_interface_mock()253        cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name)254        utils.system_output.expect_call(cmd).and_return(255            '\n Speed: 1000')256        self.assertEquals(mock_netif.get_speed(), 1000)257        self.god.check_playback()258        utils.system_output.expect_call(cmd).and_return(259            '\n Speed: 10000')260        self.assertEquals(mock_netif.get_speed(), 10000)261        self.god.check_playback()262        utils.system_output.expect_call(cmd).and_return('')263        try:264            mock_netif.get_speed()265        except ValueError:266            pass267        else:268            self.assertEquals(0,1)269        self.god.check_playback()270    def test_network_interface_is_full_duplex(self):271        mock_netif = self.network_interface_mock()272        cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name)273        utils.system_output.expect_call(cmd).and_return(274            '\n Duplex: Full')275        self.assertEquals(mock_netif.is_full_duplex(), True)276        self.god.check_playback()277        utils.system_output.expect_call(cmd).and_return(278            '\n Duplex: Half')279        self.assertEquals(mock_netif.is_full_duplex(), False)280        self.god.check_playback()281        utils.system_output.expect_call(cmd).and_return('')282        self.assertEquals(mock_netif.is_full_duplex(), False)283        self.god.check_playback()284    def test_network_interface_is_autoneg_on(self):285        mock_netif = self.network_interface_mock()286        cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name)287        utils.system_output.expect_call(cmd).and_return(288            '\n Auto-negotiation: on')289        self.assertEquals(mock_netif.is_autoneg_on(), True)290        self.god.check_playback()291        utils.system_output.expect_call(cmd).and_return(292            '\n Auto-negotiation: off')293        self.assertEquals(mock_netif.is_autoneg_on(), False)294        self.god.check_playback()295        utils.system_output.expect_call(cmd).and_return('')296        self.assertEquals(mock_netif.is_autoneg_on(), False)297        self.god.check_playback()298    def test_network_interface_get_wakeon(self):299        mock_netif = self.network_interface_mock()300        cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name)301        utils.system_output.expect_call(cmd).and_return(302            '\n Wake-on: g')303        self.assertEquals(mock_netif.get_wakeon(), 'g')304        self.god.check_playback()305    def test_network_interface_is_rx_summing_on(self):306        mock_netif = self.network_interface_mock()307        cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name)308        utils.system_output.expect_call(cmd).and_return(309            '\n rx-checksumming: on')310        self.assertEquals(mock_netif.is_rx_summing_on(), True)311        self.god.check_playback()312        utils.system_output.expect_call(cmd).and_return(313            '\n rx-checksumming: off')314        self.assertEquals(mock_netif.is_rx_summing_on(), False)315        self.god.check_playback()316        utils.system_output.expect_call(cmd).and_return('')317        self.assertEquals(mock_netif.is_rx_summing_on(), False)318        self.god.check_playback()319    def test_network_interface_is_tx_summing_on(self):320        mock_netif = self.network_interface_mock()321        cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name)322        utils.system_output.expect_call(cmd).and_return(323            '\n tx-checksumming: on')324        self.assertEquals(mock_netif.is_tx_summing_on(), True)325        self.god.check_playback()326        utils.system_output.expect_call(cmd).and_return(327            '\n tx-checksumming: off')328        self.assertEquals(mock_netif.is_tx_summing_on(), False)329        self.god.check_playback()330        utils.system_output.expect_call(cmd).and_return('')331        self.assertEquals(mock_netif.is_tx_summing_on(), False)332        self.god.check_playback()333    def test_network_interface_is_scatter_gather_on(self):334        mock_netif = self.network_interface_mock()335        cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name)336        utils.system_output.expect_call(cmd).and_return(337            '\n scatter-gather: on')338        self.assertEquals(mock_netif.is_scatter_gather_on(), True)339        self.god.check_playback()340        utils.system_output.expect_call(cmd).and_return(341            '\n scatter-gather: off')342        self.assertEquals(mock_netif.is_scatter_gather_on(), False)343        self.god.check_playback()344        utils.system_output.expect_call(cmd).and_return('')345        self.assertEquals(mock_netif.is_scatter_gather_on(), False)346        self.god.check_playback()347    def test_network_interface_is_tso_on(self):348        mock_netif = self.network_interface_mock()349        cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name)350        utils.system_output.expect_call(cmd).and_return(351            '\n tcp segmentation offload: on')352        self.assertEquals(mock_netif.is_tso_on(), True)353        self.god.check_playback()354        utils.system_output.expect_call(cmd).and_return(355            '\n tcp segmentation offload: off')356        self.assertEquals(mock_netif.is_tso_on(), False)357        self.god.check_playback()358        utils.system_output.expect_call(cmd).and_return('')359        self.assertEquals(mock_netif.is_tso_on(), False)360        self.god.check_playback()361    def test_network_interface_is_pause_autoneg_on(self):362        mock_netif = self.network_interface_mock()363        cmd = '%s %s %s' % (mock_netif.ethtool, '-a', mock_netif._name)364        utils.system_output.expect_call(cmd).and_return(365            '\n Autonegotiate: on')366        self.assertEquals(mock_netif.is_pause_autoneg_on(), True)367        self.god.check_playback()368        utils.system_output.expect_call(cmd).and_return(369            '\n Autonegotiate: off')370        self.assertEquals(mock_netif.is_pause_autoneg_on(), False)371        self.god.check_playback()372        utils.system_output.expect_call(cmd).and_return('')373        self.assertEquals(mock_netif.is_pause_autoneg_on(), False)374        self.god.check_playback()375    def test_network_interface_is_tx_pause_on(self):376        mock_netif = self.network_interface_mock()377        cmd = '%s %s %s' % (mock_netif.ethtool, '-a', mock_netif._name)378        utils.system_output.expect_call(cmd).and_return(379            '\n TX: on')380        self.assertEquals(mock_netif.is_tx_pause_on(), True)381        self.god.check_playback()382        utils.system_output.expect_call(cmd).and_return(383            '\n TX: off')384        self.assertEquals(mock_netif.is_tx_pause_on(), False)385        self.god.check_playback()386        utils.system_output.expect_call(cmd).and_return('')387        self.assertEquals(mock_netif.is_tx_pause_on(), False)388        self.god.check_playback()389    def test_network_interface_is_rx_pause_on(self):390        mock_netif = self.network_interface_mock()391        cmd = '%s %s %s' % (mock_netif.ethtool, '-a', mock_netif._name)392        utils.system_output.expect_call(cmd).and_return(393            '\n RX: on')394        self.assertEquals(mock_netif.is_rx_pause_on(), True)395        self.god.check_playback()396        utils.system_output.expect_call(cmd).and_return(397            '\n RX: off')398        self.assertEquals(mock_netif.is_rx_pause_on(), False)399        self.god.check_playback()400        utils.system_output.expect_call(cmd).and_return('')401        self.assertEquals(mock_netif.is_rx_pause_on(), False)402        self.god.check_playback()403    def test_network_interface_enable_loopback(self):404        mock_netif = self.network_interface_mock('eth0')405        mock_netif.was_loopback_enabled = False406        mock_netif.loopback_enabled = False407        mock_netif.was_down = False408        self.god.stub_function(net_utils.bonding, 'is_enabled')409        # restore using phyint410        net_utils.bonding.is_enabled.expect_call().and_return(False)411        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,412                                  'phyint', 'enable')413        utils.system.expect_call(cmd, ignore_status=True).and_return(0)414        mock_netif.enable_loopback()415        self.god.check_playback()416        # restore using mac417        net_utils.bonding.is_enabled.expect_call().and_return(False)418        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,419                                  'phyint', 'enable')420        utils.system.expect_call(cmd, ignore_status=True).and_return(1)421        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,422                                  'mac', 'enable')423        utils.system.expect_call(cmd, ignore_status=True).and_return(0)424        mock_netif.enable_loopback()425        self.god.check_playback()426        # catch exception on phyint and mac failures427        net_utils.bonding.is_enabled.expect_call().and_return(False)428        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,429                                  'phyint', 'enable')430        utils.system.expect_call(cmd, ignore_status=True).and_return(1)431        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,432                                  'mac', 'enable')433        utils.system.expect_call(cmd, ignore_status=True).and_return(1)434        try:435            mock_netif.enable_loopback()436        except error.TestError:437            pass438        else:439            self.assertEquals(0,1)440        self.god.check_playback()441        # catch exception on bond enabled442        net_utils.bonding.is_enabled.expect_call().and_return(True)443        try:444            mock_netif.enable_loopback()445        except error.TestError:446            pass447        else:448            self.assertEquals(0,1)449        self.god.check_playback()450        # check that setting tg3 and bnx2x driver have a sleep call451        mock_netif.driver = 'tg3'452        net_utils.bonding.is_enabled.expect_call().and_return(False)453        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,454                                  'phyint', 'enable')455        utils.system.expect_call(cmd, ignore_status=True).and_return(0)456        time.sleep.expect_call(1)457        mock_netif.enable_loopback()458        self.god.check_playback()459        mock_netif.driver = 'bnx2x'460        net_utils.bonding.is_enabled.expect_call().and_return(False)461        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,462                                  'phyint', 'enable')463        utils.system.expect_call(cmd, ignore_status=True).and_return(0)464        time.sleep.expect_call(1)465        mock_netif.enable_loopback()466        self.god.check_playback()467    def test_network_interface_disable_loopback(self):468        mock_netif = self.network_interface_mock('eth0')469        mock_netif.was_loopback_enabled = False470        mock_netif.loopback_enabled = True471        mock_netif.was_down = False472        # restore using phyint473        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,474                                  'phyint', 'disable')475        utils.system.expect_call(cmd, ignore_status=True).and_return(0)476        mock_netif.disable_loopback()477        self.god.check_playback()478        # restore using mac479        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,480                                  'phyint', 'disable')481        utils.system.expect_call(cmd, ignore_status=True).and_return(1)482        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,483                                  'mac', 'disable')484        utils.system.expect_call(cmd, ignore_status=True).and_return(0)485        mock_netif.disable_loopback()486        self.god.check_playback()487        # catch exception on phyint and mac failures488        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,489                                  'phyint', 'disable')490        utils.system.expect_call(cmd, ignore_status=True).and_return(1)491        cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name,492                                  'mac', 'disable')493        utils.system.expect_call(cmd, ignore_status=True).and_return(1)494        try:495            mock_netif.disable_loopback()496        except error.TestError:497            pass498        else:499            self.assertEquals(0,1)500        self.god.check_playback()501    def test_network_interface_is_loopback_enabled(self):502        mock_netif = self.network_interface_mock('eth0')503        mock_netif.is_loopback_enabled = \504            net_utils.network_interface.is_loopback_enabled505        try:506            mock_netif.is_loopback_enabled(mock_netif)507        except error.TestError:508            pass509        else:510            self.assertEquals(0,1)511        self.god.check_playback()512        self.god.stub_function(net_utils.bonding, 'is_enabled')513        mock_netif._name = 'eth0'514        net_utils.bonding.is_enabled.expect_call().and_return(False)515        cmd = '%s -l %s' % (mock_netif.ethtool, mock_netif._name)516        utils.system_output.expect_call(cmd).and_return('')517        self.assertEquals(mock_netif.is_loopback_enabled(mock_netif), False)518        self.god.check_playback()519        for ifname in ('eth0', 'eth1', 'eth2', 'eth3', 'eth4'):520            mock_netif._name = ifname521            for bond_enable in (True, False):522                for state in (('disabled', 'disabled', 'enabled'),523                              ('disabled', 'enabled', 'disabled'),524                              ('enabled', 'disabled', 'disabled'),525                              ('disabled', 'disabled', 'disabled')):526                    net_utils.bonding.is_enabled.expect_call().and_return(527                        bond_enable)528                    if bond_enable:529                        self.assertEquals(mock_netif.is_loopback_enabled(530                            mock_netif), False)531                    else:532                        cmd = '%s -l %s' % (mock_netif.ethtool, mock_netif._name)533                        out = 'MAC loopback is %s\n'\534                              'PHY internal loopback is %s\n'\535                              'PHY external loopback is %s' % (536                            state[0], state[1], state[2])537                        utils.system_output.expect_call(cmd).and_return(out)538                        self.assertEquals(mock_netif.is_loopback_enabled(539                            mock_netif), 'enabled' in state)540                    self.god.check_playback()541    def test_network_interface_enable_promisc(self):542        mock_netif = self.network_interface_mock('eth0')543        cmd = 'ifconfig %s promisc' % mock_netif._name544        utils.system.expect_call(cmd)545        mock_netif.enable_promisc()546        self.god.check_playback()547    def test_network_interface_disable_promisc(self):548        mock_netif = self.network_interface_mock()549        cmd = 'ifconfig %s -promisc' % mock_netif._name550        utils.system.expect_call(cmd)551        mock_netif.disable_promisc()552        self.god.check_playback()553    def test_network_interface_get_hwaddr(self):554        mock_netif = self.network_interface_mock()555        f = self.god.create_mock_class(file, 'file')556        net_utils.open.expect_call('/sys/class/net/%s/address'557                                       % mock_netif._name).and_return(f)558        hw_addr = '00:0e:0c:c3:7d:a8'559        f.read.expect_call().and_return(' ' + hw_addr + ' ')560        f.close.expect_call()561        self.assertEquals(mock_netif.get_hwaddr(), hw_addr)562        self.god.check_playback()563    def test_network_interface_set_hwaddr(self):564        mock_netif = self.network_interface_mock()565        hw_addr = '00:0e:0c:c3:7d:a8'566        cmd = 'ifconfig %s hw ether %s' % (mock_netif._name,567                                           hw_addr)568        utils.system.expect_call(cmd)569        mock_netif.set_hwaddr(hw_addr)570        self.god.check_playback()571    def test_network_interface_add_maddr(self):572        mock_netif = self.network_interface_mock()573        maddr = '01:00:5e:00:00:01'574        cmd = 'ip maddr add %s dev %s' % (maddr, mock_netif._name)575        utils.system.expect_call(cmd)576        mock_netif.add_maddr(maddr)577        self.god.check_playback()578    def test_network_interface_del_maddr(self):579        mock_netif = self.network_interface_mock()580        maddr = '01:00:5e:00:00:01'581        cmd = 'ip maddr del %s dev %s' % (maddr, mock_netif._name)582        utils.system.expect_call(cmd)583        mock_netif.del_maddr(maddr)584        self.god.check_playback()585    def test_network_interface_get_ipaddr(self):586        mock_netif = self.network_interface_mock()587        ip_addr = '110.211.112.213'588        out_format = \589          'eth0      Link encap:Ethernet  HWaddr 00:0E:0C:C3:7D:A8\n'\590          '          inet addr:%s  Bcast:10.246.90.255'\591          ' Mask:255.255.255.0\n'\592          '          UP BROADCAST RUNNING MASTER MULTICAST  MTU:1500'\593          ' Metric:1\n'\594          '          RX packets:463070 errors:0 dropped:0 overruns:0'\595          ' frame:0\n'\596          '          TX packets:32548 errors:0 dropped:0 overruns:0'\597          ' carrier:0\n'\598          '          collisions:0 txqueuelen:0'599        out = out_format % ip_addr600        cmd = 'ifconfig %s' % mock_netif._name601        utils.system_output.expect_call(cmd).and_return(out)602        self.assertEquals(mock_netif.get_ipaddr(), ip_addr)603        self.god.check_playback()604        cmd = 'ifconfig %s' % mock_netif._name605        utils.system_output.expect_call(cmd).and_return('some output')606        self.assertEquals(mock_netif.get_ipaddr(), '0.0.0.0')607        self.god.check_playback()608        cmd = 'ifconfig %s' % mock_netif._name609        utils.system_output.expect_call(cmd).and_return(None)610        self.assertEquals(mock_netif.get_ipaddr(), '0.0.0.0')611        self.god.check_playback()612        ip_addr = '1.2.3.4'613        out = out_format % ip_addr614        cmd = 'ifconfig %s' % mock_netif._name615        utils.system_output.expect_call(cmd).and_return(out)616        self.assertEquals(mock_netif.get_ipaddr(), ip_addr)617        self.god.check_playback()618    def test_network_interface_set_ipaddr(self):619        mock_netif = self.network_interface_mock()620        ip_addr = '1.2.3.4'621        cmd = 'ifconfig %s %s' % (mock_netif._name, ip_addr)622        utils.system.expect_call(cmd)623        mock_netif.set_ipaddr(ip_addr)624        self.god.check_playback()625    def test_network_interface_is_down(self):626        mock_netif = self.network_interface_mock()627        out_format = \628          'eth0      Link encap:Ethernet  HWaddr 00:0E:0C:C3:7D:A8\n'\629          '          inet addr:1.2.3.4  Bcast:10.246.90.255'\630          ' Mask:255.255.255.0\n'\631          '          %s BROADCAST RUNNING MASTER MULTICAST  MTU:1500'\632          ' Metric:1\n'\633          '          RX packets:463070 errors:0 dropped:0 overruns:0'\634          ' frame:0\n'\635          '          TX packets:32548 errors:0 dropped:0 overruns:0'\636          ' carrier:0\n'\637          '          collisions:0 txqueuelen:0'638        for state in ('UP', 'DOWN', 'NONE', ''):639            out = out_format % state640            cmd = 'ifconfig %s' % mock_netif._name641            utils.system_output.expect_call(cmd).and_return(out)642            self.assertEquals(mock_netif.is_down(), state != 'UP')643            self.god.check_playback()644        cmd = 'ifconfig %s' % mock_netif._name645        utils.system_output.expect_call(cmd).and_return(None)646        self.assertEquals(mock_netif.is_down(), False)647        self.god.check_playback()648    def test_network_interface_up(self):649        mock_netif = self.network_interface_mock()650        cmd = 'ifconfig %s up' % mock_netif._name651        utils.system.expect_call(cmd)652        mock_netif.up()653        self.god.check_playback()654    def test_network_interface_down(self):655        mock_netif = self.network_interface_mock()656        cmd = 'ifconfig %s down' % mock_netif._name657        utils.system.expect_call(cmd)658        mock_netif.down()659        self.god.check_playback()660    def test_network_interface_wait_for_carrier(self):661        mock_netif = self.network_interface_mock()662        mock_netif.wait_for_carrier = \663                             net_utils.network_interface.wait_for_carrier664        f = self.god.create_mock_class(file, 'file')665        spath = '/sys/class/net/%s/carrier' % mock_netif._name666        # y = 0 - test that an exception is thrown667        # y = 1, 100 - check that carrier is checked until timeout668        for y in (0, 1, 100):669            max_timeout = y670            if y:671                for x in xrange(max_timeout - 1):672                    net_utils.open.expect_call(spath).and_return(f)673                    f.read.expect_call().and_return(' ' + '0' + ' ')674                    f.close.expect_call()675                    time.sleep.expect_call(1)676                net_utils.open.expect_call(spath).and_return(f)677                f.read.expect_call().and_return(' ' + '1' + ' ')678                f.close.expect_call()679            try:680                mock_netif.wait_for_carrier(mock_netif, max_timeout)681            except:682                pass683            else:684                if not y:685                    self.assertEquals(0, 1)686            self.god.check_playback()687    def test_network_interface_send(self):688        mock_netif = self.network_interface_mock()689        mock_netif.send('test buffer')690        self.assertEquals(mock_netif._socket.send_val, 'test buffer')691    def test_network_interface_recv(self):692        mock_netif = self.network_interface_mock()693        test_str = 'test string'694        mock_netif._socket.recv_val = test_str695        rcv_str = mock_netif.recv(len(test_str))696        self.assertEquals(rcv_str, test_str)697    def test_network_interface_flush(self):698        mock_netif = self.network_interface_mock()699        self.god.stub_function(mock_netif._socket, 'close')700        mock_netif._socket.close.expect_call()701        s = self.god.create_mock_class(socket.socket, "socket")702        self.god.stub_function(socket, 'socket')703        socket.socket.expect_call(socket.PF_PACKET,704                                  socket.SOCK_RAW).and_return(s)705        s.settimeout.expect_call(net_utils.TIMEOUT)706        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))707    #708    # bonding tests709    #710    def test_bonding_is_enabled(self):711        try:712            net_utils.bond().is_enabled()713        except error.TestError:714            pass715        else:716            self.assertEquals(1, 0)717    def test_bonding_is_bondable(self):718        try:719            net_utils.bond().is_enabled()720        except error.TestError:721            pass722        else:723            self.assertEquals(1, 0)724    def test_bonding_enable(self):725        try:726            net_utils.bond().is_enabled()727        except error.TestError:728            pass729        else:730            self.assertEquals(1, 0)731    def test_bonding_disable(self):732        try:733            net_utils.bond().is_enabled()734        except error.TestError:735            pass736        else:737            self.assertEquals(1, 0)738    def test_bonding_get_mii_status(self):739        self.assertEquals(net_utils.bond().get_mii_status(), {})740    def test_get_mode_bonding(self):741        self.assertEquals(net_utils.bond().get_mode(), net_utils.bonding.NO_MODE)742    def test_bonding_wait_for_state_change(self):743        self.god.stub_function(utils, "ping_default_gateway")744        time.sleep.expect_call(10)745        utils.ping_default_gateway.expect_call().and_return(False)746        self.assertEquals(net_utils.bond().wait_for_state_change(), True)747        for x in xrange(9):748            time.sleep.expect_call(10)749            utils.ping_default_gateway.expect_call().and_return(True)750        time.sleep.expect_call(10)751        utils.ping_default_gateway.expect_call().and_return(False)752        self.assertEquals(net_utils.bond().wait_for_state_change(), True)753        for x in xrange(10):754            time.sleep.expect_call(10)755            utils.ping_default_gateway.expect_call().and_return(True)756        self.assertEquals(net_utils.bond().wait_for_state_change(), False)757        self.god.check_playback()758    def test_bonding_get_active_interfaces(self):759        self.assertEquals(net_utils.bond().get_active_interfaces(), [])760        self.god.check_playback()761    def test_bonding_get_slave_interfaces(self):762        self.assertEquals(net_utils.bond().get_slave_interfaces(), [])763        self.god.check_playback()764    #765    # ethernet tests766    #767    def test_ethernet_mac_string_to_binary(self):768        mac_bin = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05')769        self.assertEqual(mac_bin, '\x00\x01\x02\x03\x04\x05')770    def test_ethernet_mac_binary_to_string(self):771        mac_str = net_utils.ethernet.mac_binary_to_string(772            '\x00\x01\x02\x03\x04\x05')773        self.assertEqual(mac_str, '00:01:02:03:04:05')774    def test_ethernet_pack(self):775        dst = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05')776        src = net_utils.ethernet.mac_string_to_binary('16:17:18:19:1A:1B')777        protocol = 2030778        payload = 'some payload'779        frame = struct.pack("!6s6sH", dst, src, protocol) + payload780        self.assertEquals(net_utils.ethernet.pack(dst, src,protocol, payload),781                          frame)782    def test_ethernet_unpack(self):783        dst = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05')784        src = net_utils.ethernet.mac_string_to_binary('16:17:18:19:1A:1B')785        protocol = 2030786        payload = 'some payload'787        frame = net_utils.ethernet.pack(dst, src, protocol, payload)788        uframe = net_utils.ethernet.unpack(frame)789        self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_DST_MAC], dst)790        self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_SRC_MAC], src)791        self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_PROTO], protocol)792        self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_PAYLOAD], payload)793    # raw_socket tests794    #795    def test_raw_socket_open(self):796        self.god.stub_function(socket, 'setdefaulttimeout')797        s = self.god.create_mock_class(socket.socket, "socket")798        self.god.stub_function(socket, 'socket')799        # open without a protocol800        socket.setdefaulttimeout.expect_call(1)801        socket.socket.expect_call(socket.PF_PACKET,802                                  socket.SOCK_RAW).and_return(s)803        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))804        s.settimeout.expect_call(1)805        sock = net_utils.raw_socket('eth0')806        sock.open(protocol=None)807        self.god.check_playback()808        # double open should throw an exception809        try:810            sock.open()811        except error.TestError:812            pass813        else:814            self.assertEquals(1, 0)815        self.god.check_playback()816        # open a protocol817        socket.setdefaulttimeout.expect_call(1)818        socket.socket.expect_call(socket.PF_PACKET,819                                  socket.SOCK_RAW,820                                  socket.htons(1234)).and_return(s)821        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))822        s.settimeout.expect_call(1)823        sock = net_utils.raw_socket('eth0')824        sock.open(protocol=1234)825        self.god.check_playback()826    def test_raw_socket_close(self):827        self.god.stub_function(socket, 'setdefaulttimeout')828        s = self.god.create_mock_class(socket.socket, "socket")829        self.god.stub_function(socket, 'socket')830        # close without open831        socket.setdefaulttimeout.expect_call(1)832        sock = net_utils.raw_socket('eth0')833        try:834            sock.close()835        except error.TestError:836            pass837        else:838            self.assertEquals(1, 0)839        # close after open840        socket.setdefaulttimeout.expect_call(1)841        socket.socket.expect_call(socket.PF_PACKET,842                                  socket.SOCK_RAW).and_return(s)843        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))844        s.settimeout.expect_call(1)845        sock = net_utils.raw_socket('eth0')846        sock.open(protocol=None)847        s.close.expect_call()848        sock.close()849        self.god.check_playback()850    def test_raw_socket_recv(self):851        self.god.stub_function(socket, 'setdefaulttimeout')852        self.god.create_mock_class(socket.socket, "socket")853        self.god.stub_function(socket, 'socket')854        # rcv without open855        socket.setdefaulttimeout.expect_call(1)856        sock = net_utils.raw_socket('eth0')857        try:858            sock.recv(10)859        except error.TestError:860            pass861        else:862            self.assertEquals(1, 0)863        self.god.check_playback()864        # open a protocol and try to get packets of varying sizes865        # I could not get socket.recv to get a mock expect_call. To keep866        # on going, added a socket stub867        s = net_utils_mock.socket_stub('eth0', socket, socket)868        socket.socket.expect_call(socket.PF_PACKET,869                                  socket.SOCK_RAW,870                                  socket.htons(1234)).and_return(s)871        self.god.stub_function(s, 'bind')872        self.god.stub_function(s, 'settimeout')873        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))874        s.settimeout.expect_call(1)875        sock.open(protocol=1234)876        s.recv_val = ''877        self.assertEquals(sock.recv(1), (None, 0))878        s.recv_val = '\xFF' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE-5)879        self.assertEquals(sock.recv(1), (None, 0))880        # when receiving a packet, make sure the timeout is not change881        s.recv_val = '\xEE' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE-4)882        self.assertEquals(sock.recv(1), (s.recv_val, 1))883        s.recv_val = '\xDD' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE)884        self.assertEquals(sock.recv(1), (s.recv_val, 1))885        s.recv_val = '\xCC' * (net_utils.ethernet.ETH_PACKET_MAX_SIZE)886        self.assertEquals(sock.recv(1), (s.recv_val, 1))887        s.recv_val = '\xBB' * (net_utils.ethernet.ETH_PACKET_MAX_SIZE+1)888        packet, time_left = sock.recv(1)889        self.assertEquals(len(packet), net_utils.ethernet.ETH_PACKET_MAX_SIZE)890        self.assertEquals(packet,891                          s.recv_val[:net_utils.ethernet.ETH_PACKET_MAX_SIZE])892        # test timeout893        s.recv_val = ''894        s.throw_timeout = False895        sock.recv(5)896        self.assertEquals(sock.recv(1), (None, 0))897        s.throw_timeout = True898        sock.recv(5)899        self.assertEquals(sock.recv(1), (None, 0))900        self.god.check_playback()901    def test_raw_socket_send(self):902        self.god.stub_function(socket, 'setdefaulttimeout')903        self.god.create_mock_class(socket.socket, "socket")904        self.god.stub_function(socket, 'socket')905        self.god.stub_function(socket, 'send')906        # send without open907        socket.setdefaulttimeout.expect_call(1)908        sock = net_utils.raw_socket('eth0')909        try:910            sock.send('test this packet')911        except error.TestError:912            pass913        else:914            self.assertEquals(1, 0)915        self.god.check_playback()916        # open a protocol and try to send a packet917        s = net_utils_mock.socket_stub('eth0', socket, socket)918        self.god.stub_function(s, 'bind')919        self.god.stub_function(s, 'settimeout')920        socket.socket.expect_call(socket.PF_PACKET,921                                  socket.SOCK_RAW,922                                  socket.htons(1234)).and_return(s)923        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))924        s.settimeout.expect_call(1)925        packet = '\xFF\xAA\xBB\xCC\xDD\x11packet data\x00\x00'926        s.send.expect_call(packet)927        sock.open(protocol=1234)928        sock.send(packet)929        self.god.check_playback()930    def test_raw_socket_send_to(self):931        self.god.stub_function(socket, 'setdefaulttimeout')932        self.god.create_mock_class(socket.socket, "socket")933        self.god.stub_function(socket, 'socket')934        self.god.stub_function(socket, 'send')935        # send without open936        socket.setdefaulttimeout.expect_call(1)937        sock = net_utils.raw_socket('eth0')938        try:939            sock.send_to('0', '1', 1, 'test this packet')940        except error.TestError:941            pass942        else:943            self.assertEquals(1, 0)944        self.god.check_playback()945        # open a protocol and try to send a packet946        s = net_utils_mock.socket_stub('eth0', socket, socket)947        self.god.stub_function(s, 'bind')948        self.god.stub_function(s, 'settimeout')949        socket.socket.expect_call(socket.PF_PACKET,950                                  socket.SOCK_RAW,951                                  socket.htons(1234)).and_return(s)952        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))953        s.settimeout.expect_call(1)954        packet = '\x00\x00packet data\x00\x00'955        s.send.expect_call(packet)956        sock.open(protocol=1234)957        try:958            sock.send_to(None, None, 1, packet)959        except error.TestError:960            pass961        else:962            self.assertEquals(1, 0)963        self.god.check_playback()964        dst_mac = '\x00\x01\x02\x03\x04\x05'965        src_mac = '\xFF\xEE\xDD\xCC\xBB\xAA'966        protocol = 1234967        s.send.expect_call(dst_mac+src_mac+'%d'%protocol+packet)968        sock.send_to(dst_mac, src_mac, protocol, packet)969        self.god.check_playback()970    def test_raw_socket_recv_from(self):971        def __set_clock(sock):972            time.clock.expect_call().and_return(0.0)973            time.clock.expect_call().and_return(0.0)974            time.clock.expect_call().and_return(float(sock.socket_timeout()) + 0.5)975        self.god.stub_function(socket, 'setdefaulttimeout')976        self.god.create_mock_class(socket.socket, "socket")977        self.god.stub_function(socket, 'socket')978        # rcv without open979        socket.setdefaulttimeout.expect_call(1)980        sock = net_utils.raw_socket('eth0')981        try:982            sock.recv_from(None, None, None)983        except error.TestError:984            pass985        else:986            self.assertEquals(1, 0)987        self.god.check_playback()988        # open a protocol and try to get packets of varying sizes989        # I could not get socket.recv to get a mock expect_call. To keep990        # on going, added a socket stub991        s = net_utils_mock.socket_stub('eth0', socket, socket)992        socket.socket.expect_call(socket.PF_PACKET,993                                  socket.SOCK_RAW,994                                  socket.htons(1234)).and_return(s)995        self.god.stub_function(s, 'bind')996        self.god.stub_function(s, 'settimeout')997        s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL))998        s.settimeout.expect_call(1)999        sock.open(protocol=1234)1000        s.recv_val = ''1001        dst_mac = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05')1002        src_mac = net_utils.ethernet.mac_string_to_binary('16:17:18:19:1A:1B')1003        t_mac = net_utils.ethernet.mac_string_to_binary('E6:E7:E8:E9:EA:EB')1004        protocol = 20301005        t_protocol = 12341006        data = '\xEE' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE)1007        # no data to receive at socket1008        self.assertEquals(sock.recv_from(None, None, None), None)1009        self.assertEquals(sock.recv_from(dst_mac, None, None), None)1010        self.assertEquals(sock.recv_from(None, src_mac, None), None)1011        self.assertEquals(sock.recv_from(None, None, protocol), None)1012        # receive packet < min size...kernel_unittest.py
Source:kernel_unittest.py  
...10                          initrd='initrd'):11        god = mock.mock_god()12        bootloader = god.create_mock_class(boottool.boottool, "boottool")13        # record14        bootloader.remove_kernel.expect_call(tag)15        bootloader.add_kernel.expect_call(image, tag, initrd=initrd,16                                          args='_dummy_', root=bootloader_root)17        for a in bootloader_args.split():18            bootloader.add_args.expect_call(kernel=tag, args=a)19        bootloader.remove_args.expect_call(kernel=tag, args='_dummy_')20        # run and check21        kernel._add_kernel_to_bootloader(bootloader, base_args, tag, args,22                                         image, initrd)23        god.check_playback()24    def test_add_kernel_to_bootloader(self):25        self.add_to_bootloader(base_args='baseargs', args='',26                               bootloader_args='baseargs', bootloader_root=None)27        self.add_to_bootloader(base_args='arg1 root=/dev/oldroot arg2',28                               args='root=/dev/newroot arg3',29                               bootloader_args='arg1 arg2 arg3',30                               bootloader_root='/dev/newroot')31class TestBootableKernel(unittest.TestCase):32    def setUp(self):33        self.god = mock.mock_god()34        self.god.stub_function(time, "time")35        self.god.stub_function(utils, "system")36        self.god.stub_function(kernel, "_add_kernel_to_bootloader")37        job_ = self.god.create_mock_class(job.job, "job")38        self.kernel = kernel.BootableKernel(job_)39        self.kernel.job.bootloader = self.god.create_mock_class(40                              boottool.boottool, "boottool")41    def tearDown(self):42        # note: time.time() can only be unstubbed via tearDown()43        self.god.unstub_all()44    def boot_kernel(self, ident_check):45        notes = "applied_patches"46        when = 147        args = ''48        base_args = 'base_args'49        tag = 'ident'50        subdir = 'subdir'51        self.kernel.image = 'image'52        self.kernel.initrd = 'initrd'53        self.kernel.installed_as = tag54        # record55        args_ = args56        if ident_check:57            time.time.expect_call().and_return(when)58            args_ += " IDENT=%d" % when59            status = ["job.end_reboot_and_verify", when, tag, subdir, notes]60        else:61            status = ["job.end_reboot", subdir, tag, notes]62        self.kernel.job.next_step_prepend.expect_call(status)63        self.kernel.job.config_get.expect_call(64                'boot.default_args').and_return(base_args)65        kernel._add_kernel_to_bootloader.expect_call(66                self.kernel.job.bootloader, base_args, tag,67                args_, self.kernel.image, self.kernel.initrd)68        utils.system.expect_call('touch /fastboot')69        self.kernel.job.start_reboot.expect_call()70        self.kernel.job.reboot.expect_call(tag=tag)71        # run and check72        self.kernel._boot_kernel(args=args, ident_check=ident_check,73                                 expected_ident=tag, subdir=subdir, notes=notes)74        self.god.check_playback()75    def test_boot_kernel(self):76        self.boot_kernel(ident_check=False)77        self.boot_kernel(ident_check=True)78class TestKernel(unittest.TestCase):79    def setUp(self):80        self.god = mock.mock_god()81        logging.disable(logging.CRITICAL)82        self.god.stub_function(time, "time")83        self.god.stub_function(os, "mkdir")84        self.god.stub_function(os, "chdir")85        self.god.stub_function(os, "symlink")86        self.god.stub_function(os, "remove")87        self.god.stub_function(os.path, "isdir")88        self.god.stub_function(os.path, "exists")89        self.god.stub_function(os.path, "isfile")90        self.god.stub_function(os_dep, "commands")91        self.god.stub_function(kernel, "open")92        self.god.stub_function(utils, "system")93        self.god.stub_function(utils, "system_output")94        self.god.stub_function(utils, "get_file")95        self.god.stub_function(utils, "get_current_kernel_arch")96        self.god.stub_function(utils, "cat_file_to_cmd")97        self.god.stub_function(utils, "force_copy")98        self.god.stub_function(utils, "extract_tarball_to_dir")99        self.god.stub_function(utils, "count_cpus")100        self.god.stub_function(utils, "get_os_vendor")101        self.god.stub_function(kernelexpand, "expand_classic")102        self.god.stub_function(kernel_config, "modules_needed")103        self.god.stub_function(glob, "glob")104        def dummy_mark(filename, msg):105            pass106        self.god.stub_with(kernel, '_mark', dummy_mark)107        self.job = self.god.create_mock_class(job.job, "job")108        self.job.bootloader = self.god.create_mock_class(boottool.boottool,109                                                         "boottool")110        class DummyLoggingManager(object):111            def tee_redirect_debug_dir(self, *args, **kwargs):112                pass113            def restore(self, *args, **kwargs):114                pass115        self.job.logging = DummyLoggingManager()116        self.job.autodir = "autodir"117        self.base_tree = "2.6.24"118        self.tmp_dir = "tmpdir"119        self.subdir = "subdir"120    def tearDown(self):121        self.god.unstub_all()122    def construct_kernel(self):123        self.kernel = kernel.kernel.__new__(kernel.kernel)124        self.god.stub_function(self.kernel, "extract")125        # setup126        self.src_dir    = os.path.join(self.tmp_dir, 'src')127        self.build_dir  = os.path.join(self.tmp_dir, "build_dir")128        self.config_dir = os.path.join(self.subdir, 'config')129        self.log_dir    = os.path.join(self.subdir, 'debug')130        self.results_dir = os.path.join(self.subdir, 'results')131        # record132        os.path.isdir.expect_call(self.src_dir).and_return(True)133        utils.system.expect_call('rm -rf ' + self.src_dir)134        os.path.isdir.expect_call(self.build_dir).and_return(True)135        utils.system.expect_call('rm -rf ' + self.build_dir)136        os.path.exists.expect_call(self.src_dir).and_return(False)137        os.mkdir.expect_call(self.src_dir)138        for path in [self.config_dir, self.log_dir, self.results_dir]:139            os.path.exists.expect_call(path).and_return(True)140            utils.system.expect_call('rm -rf ' + path)141            os.mkdir.expect_call(path)142        logpath = os.path.join(self.log_dir, 'build_log')143        self.logfile = self.god.create_mock_class(file, "file")144        kernel.open.expect_call(logpath, 'w+').and_return(self.logfile)145        utils.get_current_kernel_arch.expect_call().and_return('ia64')146        self.logfile.write.expect_call('BASE: %s\n' % self.base_tree)147        self.kernel.extract.expect_call(self.base_tree)148        # finish creation of kernel object and test (and unstub extract)149        self.kernel.__init__(self.job, self.base_tree, self.subdir,150                             self.tmp_dir, "build_dir")151        self.god.check_playback()152        self.god.unstub(self.kernel, "extract")153    def test_constructor(self):154        self.construct_kernel()155    def test_kernelexpand1(self):156        self.construct_kernel()157        ret_val = self.kernel.kernelexpand("/path/to/kernel")158        self.assertEquals(ret_val, ["/path/to/kernel"])159        self.god.check_playback()160    def test_kernel_expand2(self):161        self.construct_kernel()162        kernel = "kernel.tar.gz"163        # record164        self.job.config_get.expect_call('mirror.mirrors').and_return('mirror')165        kernelexpand.expand_classic.expect_call(kernel,166            'mirror').and_return('patches')167        # run168        self.assertEquals(self.kernel.kernelexpand(kernel), 'patches')169        self.god.check_playback()170    def test_kernel_expand3(self):171        self.construct_kernel()172        kernel = "kernel.tar.gz"173        # record174        self.job.config_get.expect_call('mirror.mirrors')175        self.job.config_get.expect_call(176            'mirror.ftp_kernel_org').and_return('mirror')177        korg = 'http://www.kernel.org/pub/linux/kernel'178        mirrors = [179                   [ korg + '/v2.6', 'mirror' + '/v2.6' ],180                   [ korg + '/people/akpm/patches/2.6', 'mirror' + '/akpm' ],181                   [ korg + '/people/mbligh', 'mirror' + '/mbligh' ],182                  ]183        kernelexpand.expand_classic.expect_call(kernel,184            mirrors).and_return('patches')185        # run186        self.assertEquals(self.kernel.kernelexpand(kernel), 'patches')187        self.god.check_playback()188    def test_extract1(self):189        self.construct_kernel()190        # setup191        self.god.stub_function(self.kernel, "get_kernel_tree")192        # record193        os.path.exists.expect_call(self.base_tree).and_return(True)194        self.kernel.get_kernel_tree.expect_call(self.base_tree)195        self.job.record.expect_call('GOOD', self.subdir, 'kernel.extract')196        # run197        self.kernel.extract(self.base_tree)198        self.god.check_playback()199        self.god.unstub(self.kernel, "get_kernel_tree")200    def test_extract2(self):201        self.construct_kernel()202        # setup203        self.god.stub_function(self.kernel, "kernelexpand")204        self.god.stub_function(self.kernel, "get_kernel_tree")205        self.god.stub_function(self.kernel, "patch")206        # record207        os.path.exists.expect_call(self.base_tree).and_return(False)208        components = ["component0", "component1"]209        self.kernel.kernelexpand.expect_call(self.base_tree).and_return(210            components)211        self.kernel.get_kernel_tree.expect_call(components[0])212        self.kernel.patch.expect_call(components[1])213        self.job.record.expect_call('GOOD', self.subdir, 'kernel.extract')214        # run215        self.kernel.extract(self.base_tree)216        self.god.check_playback()217        self.god.unstub(self.kernel, "kernelexpand")218        self.god.unstub(self.kernel, "get_kernel_tree")219        self.god.unstub(self.kernel, "patch")220    def test_patch1(self):221        self.construct_kernel()222        patches = ('patch1', 'patch2')223        self.god.stub_function(self.kernel, "apply_patches")224        self.god.stub_function(self.kernel, "get_patches")225        #record226        self.kernel.get_patches.expect_call(patches).and_return(patches)227        self.kernel.apply_patches.expect_call(patches)228        self.job.record.expect_call('GOOD', self.subdir, 'kernel.patch')229        #run230        self.kernel.patch(*patches)231        self.god.check_playback()232        self.god.unstub(self.kernel, "apply_patches")233        self.god.unstub(self.kernel, "get_patches")234    def test_patch2(self):235        self.construct_kernel()236        patches = []237        # record238        self.job.record.expect_call('GOOD', self.subdir, 'kernel.patch')239        # run240        self.kernel.patch(*patches)241        self.god.check_playback()242    def test_config(self):243        self.construct_kernel()244        # setup245        self.god.stub_function(self.kernel, "set_cross_cc")246        self.god.stub_class(kernel_config, "kernel_config")247        # record248        self.kernel.set_cross_cc.expect_call()249        kernel_config.kernel_config.expect_new(self.job, self.build_dir,250                                               self.config_dir, '', None,251                                               False, self.base_tree, None)252        self.job.record.expect_call('GOOD', self.subdir, 'kernel.config')253        # run254        self.kernel.config()255        self.god.check_playback()256        self.god.unstub(self.kernel, "set_cross_cc")257    def test_get_patches(self):258        self.construct_kernel()259        # setup260        patches = ['patch1', 'patch2', 'patch3']261        local_patches = []262        # record263        for patch in patches:264            dest = os.path.join(self.src_dir, os.path.basename(patch))265            utils.get_file.expect_call(patch, dest)266            utils.system_output.expect_call(267                'md5sum ' + dest).and_return('md5sum')268            local_patches.append((patch, dest, 'md5sum'))269        # run and check270        self.assertEquals(self.kernel.get_patches(patches), local_patches)271        self.god.check_playback()272    def test_apply_patches(self):273        self.construct_kernel()274        # setup275        patches = []276        patches.append(('patch1', 'patch1.gz', 'md5sum1'))277        patches.append(('patch2', 'patch2.bz2', 'md5sum2'))278        patches.append(('patch3', 'patch3', 'md5sum3'))279        applied_patches = []280        # record281        os.chdir.expect_call(self.build_dir)282        patch_id = "%s %s %s" % ('patch1', 'patch1', 'md5sum1')283        log = "PATCH: " + patch_id + "\n"284        utils.cat_file_to_cmd.expect_call('patch1.gz',285            'patch -p1 > /dev/null')286        self.logfile.write.expect_call(log)287        applied_patches.append(patch_id)288        patch_id = "%s %s %s" % ('patch2', 'patch2', 'md5sum2')289        log = "PATCH: " + patch_id + "\n"290        utils.cat_file_to_cmd.expect_call('patch2.bz2',291            'patch -p1 > /dev/null')292        self.logfile.write.expect_call(log)293        applied_patches.append(patch_id)294        utils.force_copy.expect_call('patch3',295            self.results_dir).and_return('local_patch3')296        self.job.relative_path.expect_call('local_patch3').and_return(297            'rel_local_patch3')298        patch_id = "%s %s %s" % ('patch3', 'rel_local_patch3', 'md5sum3')299        log = "PATCH: " + patch_id + "\n"300        utils.cat_file_to_cmd.expect_call('patch3',301            'patch -p1 > /dev/null')302        self.logfile.write.expect_call(log)303        applied_patches.append(patch_id)304        # run and test305        self.kernel.apply_patches(patches)306        self.assertEquals(self.kernel.applied_patches, applied_patches)307        self.god.check_playback()308    def test_get_kernel_tree1(self):309        self.construct_kernel()310        # record311        os.path.isdir.expect_call(self.base_tree).and_return(True)312        os.symlink.expect_call(self.base_tree, self.build_dir)313        # run and check314        self.kernel.get_kernel_tree(self.base_tree)315        self.god.check_playback()316    def test_get_kernel_tree2(self):317        self.construct_kernel()318        # record319        os.path.isdir.expect_call(self.base_tree).and_return(False)320        os.chdir.expect_call(os.path.dirname(self.src_dir))321        tarball = os.path.join(self.src_dir, os.path.basename(self.base_tree))322        utils.get_file.expect_call(self.base_tree, tarball)323        utils.extract_tarball_to_dir.expect_call(tarball,324                                                          self.build_dir)325        # run and check326        self.kernel.get_kernel_tree(self.base_tree)327        self.god.check_playback()328    def test_extraversion(self):329        self.construct_kernel()330        tag = "tag"331        # setup332        self.god.stub_function(self.kernel, "config")333        # record334        os.chdir.expect_call(self.build_dir)335        extraversion_sub = r's/^CONFIG_LOCALVERSION=\s*"\(.*\)"/CONFIG_LOCALVERSION='336        cfg = self.build_dir + '/.config'337        p = extraversion_sub + '"\\1-%s"/' % tag338        utils.system.expect_call('mv %s %s.old' % (cfg, cfg))339        utils.system.expect_call("sed '%s' < %s.old > %s" % (p, cfg, cfg))340        self.kernel.config.expect_call(make='oldconfig')341        # run and check342        self.kernel.extraversion(tag)343        self.god.check_playback()344    def test_build(self):345        self.construct_kernel()346        self.god.stub_function(self.kernel, "extraversion")347        self.god.stub_function(self.kernel, "set_cross_cc")348        self.god.stub_function(self.kernel, "get_kernel_build_ver")349        self.kernel.build_target = 'build_target'350        # record351        os_dep.commands.expect_call('gcc', 'make')352        logfile = os.path.join(self.log_dir, 'kernel_build')353        os.chdir.expect_call(self.build_dir)354        self.kernel.extraversion.expect_call('autotest')355        self.kernel.set_cross_cc.expect_call()356        utils.system.expect_call('make dep', ignore_status=True)357        utils.count_cpus.expect_call().and_return(4)358        threads = 2 * 4359        build_string = 'make -j %d %s %s' % (threads, '', 'build_target')360        utils.system.expect_call(build_string)361        kernel_config.modules_needed.expect_call('.config').and_return(True)362        utils.system.expect_call('make -j %d modules' % (threads))363        self.kernel.get_kernel_build_ver.expect_call().and_return('2.6.24')364        kernel_version = re.sub('-autotest', '', '2.6.24')365        self.logfile.write.expect_call('BUILD VERSION: %s\n' % kernel_version)366        utils.force_copy.expect_call(self.build_dir+'/System.map',367                                              self.results_dir)368        self.job.record.expect_call('GOOD', self.subdir, 'kernel.build')369        # run and check370        self.kernel.build()371        self.god.check_playback()372    def test_build_timed(self):373        self.construct_kernel()374        self.god.stub_function(self.kernel, "set_cross_cc")375        self.god.stub_function(self.kernel, "clean")376        # record377        os.chdir.expect_call(self.build_dir)378        self.kernel.set_cross_cc.expect_call()379        self.kernel.clean.expect_call()380        build_string = "/usr/bin/time -o /dev/null make  -j 8 vmlinux"381        build_string += ' > /dev/null 2>&1'382        utils.system.expect_call(build_string)383        os.path.isfile.expect_call('vmlinux').and_return(True)384        # run and check385        self.kernel.build_timed(threads=8)386        self.god.check_playback()387    def test_clean(self):388        self.construct_kernel()389        # record390        os.chdir.expect_call(self.build_dir)391        utils.system.expect_call('make clean > /dev/null 2> /dev/null')392        self.job.record.expect_call('GOOD', self.subdir, 'kernel.clean')393        # run and check394        self.kernel.clean()395        self.god.check_playback()396    def test_mkinitrd(self):397        self.construct_kernel()398        # record399        utils.get_os_vendor.expect_call().and_return('Ubuntu')400        os.path.isfile.expect_call('initrd').and_return(True)401        os.remove.expect_call('initrd')402        self.job.config_get.expect_call(403            'kernel.mkinitrd_extra_args').and_return(None)404        args = ''405        glob.glob.expect_call('/lib/modules/2.6.24*').and_return(['2.6.24'])406        os.path.isfile.expect_call('/usr/sbin/mkinitrd').and_return(True)407        cmd = '/usr/sbin/mkinitrd'408        utils.system.expect_call('%s %s -o initrd 2.6.24' % (cmd, args))409        self.job.record.expect_call('GOOD', self.subdir, 'kernel.mkinitrd')410        # run and check411        self.kernel.mkinitrd(version="2.6.24", image="image",412                             system_map="system_map", initrd="initrd")413        self.god.check_playback()414    def test_install(self):415        self.construct_kernel()416        tag = 'autotest'417        prefix = '/'418        self.kernel.build_image = None419        self.kernel.build_target = 'build_target'420        self.god.stub_function(self.kernel, "get_kernel_build_ver")421        self.god.stub_function(self.kernel, "mkinitrd")422        # record423        os.chdir.expect_call(self.build_dir)424        os.path.isdir.expect_call(prefix).and_return(False)425        os.mkdir.expect_call(prefix)426        boot_dir = os.path.join(prefix, 'boot')427        os.path.isdir.expect_call(boot_dir).and_return(False)428        os.mkdir.expect_call(boot_dir)429        glob.glob.expect_call(430            'arch/*/boot/' + 'build_target').and_return('')431        build_image = self.kernel.build_target432        utils.force_copy.expect_call('vmlinux',433            '/boot/vmlinux-autotest')434        utils.force_copy.expect_call('build_target',435            '/boot/vmlinuz-autotest')436        utils.force_copy.expect_call('System.map',437            '/boot/System.map-autotest')438        utils.force_copy.expect_call('.config',439            '/boot/config-autotest')440        kernel_config.modules_needed.expect_call('.config').and_return(True)441        utils.system.expect_call('make modules_install INSTALL_MOD_PATH=%s'442                                 % prefix)443        initrd = boot_dir + '/initrd-' + tag444        self.kernel.get_kernel_build_ver.expect_call().and_return('2.6.24')445        self.kernel.mkinitrd.expect_call('2.6.24', '/boot/vmlinuz-autotest',446            '/boot/System.map-autotest', '/boot/initrd-autotest')447        self.job.record.expect_call('GOOD', self.subdir, 'kernel.install')448        # run and check449        self.kernel.install()450        self.god.check_playback()451    def test_get_kernel_build_arch1(self):452        self.construct_kernel()453        # record454        utils.get_current_kernel_arch.expect_call().and_return("i386")455        # run and check456        self.assertEquals(self.kernel.get_kernel_build_arch(), "i386")457        self.god.check_playback()458    def test_get_kernel_build_arch2(self):459        self.construct_kernel()460        # run and check461        self.assertEquals(self.kernel.get_kernel_build_arch('i586'), "i386")462        self.god.check_playback()463    def test_get_kernel_build_release(self):464        self.construct_kernel()465        mock_file = self.god.create_mock_class(file, "file")466        # record467        for f in [self.build_dir + "/include/linux/version.h",468                  self.build_dir + "/include/linux/utsrelease.h"]:469            os.path.exists.expect_call(f).and_return(True)470            kernel.open.expect_call(f, 'r').and_return(mock_file)471            mock_file.readlines.expect_call().and_return("Some lines")472            mock_file.close.expect_call()473        for f in [self.build_dir + "/include/linux/compile.h",474                  self.build_dir + "/include/generated/utsrelease.h",475                  self.build_dir + "/include/generated/compile.h"]:476            os.path.exists.expect_call(f).and_return(False)477        # run and test478        self.kernel.get_kernel_build_release()479        self.god.check_playback()480    def test_get_kernel_build_ident(self):481        self.construct_kernel()482        self.god.stub_function(self.kernel, "get_kernel_build_release")483        # record484        self.kernel.get_kernel_build_release.expect_call().and_return(485            ("AwesomeRelease", "1.0"))486        # run and check487        self.assertEquals(self.kernel.get_kernel_build_ident(),488            "AwesomeRelease::1.0")489        self.god.check_playback()490    def test_boot(self):491        self.construct_kernel()492        self.god.stub_function(self.kernel, "get_kernel_build_ident")493        self.god.stub_function(self.kernel, "install")494        self.god.stub_function(self.kernel, "_boot_kernel")495        self.kernel.applied_patches = "applied_patches"496        self.kernel.installed_as = None497        args = ''498        expected_ident = 'ident'499        ident = True500        # record501        self.kernel.install.expect_call()502        self.kernel.get_kernel_build_ident.expect_call(503                ).and_return(expected_ident)504        self.kernel._boot_kernel.expect_call(505                args, ident, expected_ident,506                self.subdir, self.kernel.applied_patches)507        # run and check508        self.kernel.boot(args=args, ident=ident)509        self.god.check_playback()510if __name__ == "__main__":...autotest_unittest.py
Source:autotest_unittest.py  
...55    def construct(self):56        # setup57        self.serverdir = "serverdir"58        # record59        utils.get_server_dir.expect_call().and_return(self.serverdir)60        # create the autotest object61        self.base_autotest = autotest.BaseAutotest(self.host)62        self.base_autotest.job = self.host.job63        self.god.stub_function(self.base_autotest, "_install_using_send_file")64        # stub out abspath65        self.god.stub_function(os.path, "abspath")66        # check67        self.god.check_playback()68    def record_install_prologue(self):69        self.construct()70        # setup71        self.god.stub_class(packages, "PackageManager")72        self.base_autotest.got = False73        location = os.path.join(self.serverdir, '../client')74        location = os.path.abspath.expect_call(location).and_return(location)75        # record76        os.getcwd.expect_call().and_return('cwd')77        os.chdir.expect_call(os.path.join(self.serverdir, '../client'))78        utils.system.expect_call('tools/make_clean', ignore_status=True)79        os.chdir.expect_call('cwd')80        utils.get.expect_call(os.path.join(self.serverdir,81            '../client')).and_return('source_material')82        self.host.wait_up.expect_call(timeout=30)83        self.host.setup.expect_call()84        self.host.get_autodir.expect_call().and_return("autodir")85        self.host.set_autodir.expect_call("autodir")86        self.host.run.expect_call('mkdir -p autodir')87        self.host.run.expect_call('rm -rf autodir/results/*',88                                  ignore_status=True)89    def test_constructor(self):90        self.construct()91        # we should check the calls92        self.god.check_playback()93    def test_full_client_install(self):94        self.record_install_prologue()95        self.host.run.expect_call('rm -f "autodir/packages.checksum"')96        c = autotest.global_config.global_config97        c.get_config_value.expect_call('PACKAGES',98                                       'serve_packages_from_autoserv',99                                       type=bool).and_return(False)100        self.host.send_file.expect_call('source_material', 'autodir',101                                        delete_dest=True)102        # run and check103        self.base_autotest.install_full_client()104        self.god.check_playback()105    def test_autoserv_install(self):106        self.record_install_prologue()107        c = autotest.global_config.global_config108        c.get_config_value.expect_call('PACKAGES',109            'fetch_location', type=list, default=[]).and_return([])110        c.get_config_value.expect_call('PACKAGES',111                                       'serve_packages_from_autoserv',112                                       type=bool).and_return(True)113        self.base_autotest._install_using_send_file.expect_call(self.host,114                                                                'autodir')115        # run and check116        self.base_autotest.install()117        self.god.check_playback()118    def test_packaging_install(self):119        self.record_install_prologue()120        c = autotest.global_config.global_config121        c.get_config_value.expect_call('PACKAGES',122            'fetch_location', type=list, default=[]).and_return(['repo'])123        pkgmgr = packages.PackageManager.expect_new('autodir',124            repo_urls=['repo'], hostname='hostname', do_locking=False,125            run_function=self.host.run, run_function_dargs=dict(timeout=600))126        pkg_dir = os.path.join('autodir', 'packages')127        cmd = ('cd autodir && ls | grep -v "^packages$"'128               ' | xargs rm -rf && rm -rf .[!.]*')129        self.host.run.expect_call(cmd)130        pkgmgr.install_pkg.expect_call('autotest', 'client', pkg_dir,131                                       'autodir', preserve_install_dir=True)132        # run and check133        self.base_autotest.install()134        self.god.check_playback()135    def test_run(self):136        self.construct()137        # setup138        control = "control"139        # stub out install140        self.god.stub_function(self.base_autotest, "install")141        # record142        self.base_autotest.install.expect_call(self.host, use_packaging=True)143        self.host.wait_up.expect_call(timeout=30)144        os.path.abspath.expect_call('.').and_return('.')145        run_obj = autotest._Run.expect_new(self.host, '.', None, False, False)146        tag = None147        run_obj.manual_control_file = os.path.join('autodir',148                                                   'control.%s' % tag)149        run_obj.remote_control_file = os.path.join('autodir',150                                                   'control.%s.autoserv' % tag)151        run_obj.tag = tag152        run_obj.autodir = 'autodir'153        run_obj.verify_machine.expect_call()154        run_obj.background = False155        debug = os.path.join('.', 'debug')156        os.makedirs.expect_call(debug)157        delete_file_list = [run_obj.remote_control_file,158                            run_obj.remote_control_file + '.state',159                            run_obj.manual_control_file,160                            run_obj.manual_control_file + '.state']161        cmd = ';'.join('rm -f ' + control for control in delete_file_list)162        self.host.run.expect_call(cmd, ignore_status=True)163        utils.get.expect_call(control, local_copy=True).and_return("temp")164        c = autotest.global_config.global_config165        c.get_config_value.expect_call("PACKAGES",166            'fetch_location', type=list, default=[]).and_return(['repo'])167        cfile = self.god.create_mock_class(file, "file")168        cfile_orig = "original control file"169        cfile_new = "args = []\njob.add_repository(['repo'])\n"170        cfile_new += cfile_orig171        autotest.open.expect_call("temp").and_return(cfile)172        cfile.read.expect_call().and_return(cfile_orig)173        autotest.open.expect_call("temp", 'w').and_return(cfile)174        cfile.write.expect_call(cfile_new)175        self.host.job.preprocess_client_state.expect_call().and_return(176            '/job/tmp/file1')177        self.host.send_file.expect_call(178            "/job/tmp/file1", "autodir/control.None.autoserv.init.state")179        os.remove.expect_call("/job/tmp/file1")180        self.host.send_file.expect_call("temp", run_obj.remote_control_file)181        os.path.abspath.expect_call('temp').and_return('control_file')182        os.path.abspath.expect_call('control').and_return('control')183        os.remove.expect_call("temp")184        run_obj.execute_control.expect_call(timeout=30,185                                            client_disconnect_timeout=240)186        # run and check output187        self.base_autotest.run(control, timeout=30)188        self.god.check_playback()189    def _stub_get_client_autodir_paths(self):190        def mock_get_client_autodir_paths(cls, host):191            return ['/some/path', '/another/path']192        self.god.stub_with(autotest.Autotest, 'get_client_autodir_paths',193                           classmethod(mock_get_client_autodir_paths))194    def _expect_failed_run(self, command):195        (self.host.run.expect_call(command)196         .and_raises(error.AutoservRunError('dummy', object())))197    def test_get_installed_autodir(self):198        self._stub_get_client_autodir_paths()199        self.host.get_autodir.expect_call().and_return(None)200        self._expect_failed_run('test -x /some/path/bin/autotest')201        self.host.run.expect_call('test -x /another/path/bin/autotest')202        self.host.run.expect_call('test -w /another/path')203        autodir = autotest.Autotest.get_installed_autodir(self.host)204        self.assertEquals(autodir, '/another/path')205    def test_get_install_dir(self):206        self._stub_get_client_autodir_paths()207        self.host.get_autodir.expect_call().and_return(None)208        self._expect_failed_run('test -x /some/path/bin/autotest')209        self._expect_failed_run('test -x /another/path/bin/autotest')210        self._expect_failed_run('mkdir -p /some/path')211        self.host.run.expect_call('mkdir -p /another/path')212        self.host.run.expect_call('test -w /another/path')213        install_dir = autotest.Autotest.get_install_dir(self.host)214        self.assertEquals(install_dir, '/another/path')215    def test_client_logger_process_line_log_copy_collection_failure(self):216        collector = autotest.log_collector.expect_new(self.host, '', '')217        logger = autotest.client_logger(self.host, '', '')218        collector.collect_client_job_results.expect_call().and_raises(219                Exception('log copy failure'))220        logging.exception.expect_call(mock.is_string_comparator())221        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo1')222    def test_client_logger_process_line_log_copy_fifo_failure(self):223        collector = autotest.log_collector.expect_new(self.host, '', '')224        logger = autotest.client_logger(self.host, '', '')225        collector.collect_client_job_results.expect_call()226        self.host.run.expect_call('echo A > /autotest/fifo2').and_raises(227                Exception('fifo failure'))228        logging.exception.expect_call(mock.is_string_comparator())229        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo2')230    def test_client_logger_process_line_package_install_fifo_failure(self):231        collector = autotest.log_collector.expect_new(self.host, '', '')232        logger = autotest.client_logger(self.host, '', '')233        self.god.stub_function(logger, '_send_tarball')234        c = autotest.global_config.global_config235        c.get_config_value.expect_call('PACKAGES',236                                       'serve_packages_from_autoserv',237                                       type=bool).and_return(True)238        c.get_config_value.expect_call('PACKAGES',239                                       'serve_packages_from_autoserv',240                                       type=bool).and_return(True)241        logger._send_tarball.expect_call('pkgname.tar.bz2', '/autotest/dest/')242        self.host.run.expect_call('echo B > /autotest/fifo3').and_raises(243                Exception('fifo failure'))244        logging.exception.expect_call(mock.is_string_comparator())245        logger._process_line('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'246                             '/autotest/dest/:/autotest/fifo3')247if __name__ == "__main__":...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
