How to use rebuild method in Behave

Best Python code snippet using behave

Run Behave automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

test_rebuild.py

Source: test_rebuild.py Github

copy
1from mock import patch, MagicMock, mock
2from srv.modules.runners import rebuild
3
4
5class TestRebuild():
6    """
7    """
8
9    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
10    @patch('salt.runner.Runner', autospec=True)
11    @patch('salt.client.LocalClient', autospec=True)
12    def test_minions(self, localclient, runner, mm):
13        rebuild.__opts__ = {}
14
15        local = localclient.return_value
16        local.cmd.return_value = {'data1.ceph': 'data1.ceph'}
17        rr = rebuild.Rebuild(['data1.ceph'])
18        assert rr.minions == ['data1.ceph']
19
20    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
21    @patch('salt.runner.Runner', autospec=True)
22    @patch('salt.client.LocalClient', autospec=True)
23    def test_minions_targets(self, localclient, runner, mm):
24        rebuild.__opts__ = {}
25
26        local = localclient.return_value
27        local.cmd.return_value = {'data1.ceph': 'data1.ceph',
28                                  'data2.ceph':'data2.ceph'}
29        rr = rebuild.Rebuild(['[email protected]:storage'])
30        assert rr.minions == ['data1.ceph', 'data2.ceph']
31
32    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
33    @patch('salt.runner.Runner', autospec=True)
34    @patch('salt.client.LocalClient', autospec=True)
35    def test_minions_multiple(self, localclient, runner, mm):
36        rebuild.__opts__ = {}
37
38        local = localclient.return_value
39        local.cmd.return_value = {'data1.ceph': 'data1.ceph',
40                                  'data2.ceph':'data2.ceph'}
41        rr = rebuild.Rebuild(['data*.ceph', 'data*.ceph'])
42        assert rr.minions == ['data1.ceph', 'data2.ceph']
43
44    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
45    @patch('salt.runner.Runner', autospec=True)
46    @patch('salt.client.LocalClient', autospec=True)
47    def test_osd_list(self, localclient, runner, mm):
48        rebuild.__opts__ = {}
49
50        local = localclient.return_value
51        local.cmd.return_value = {'data1.ceph': [0, 1, 2]}
52        rr = rebuild.Rebuild(['data*.ceph'])
53        result = rr._osd_list('data1.ceph')
54        assert result == [0, 1, 2]
55
56    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
57    @patch('salt.runner.Runner', autospec=True)
58    @patch('salt.client.LocalClient', autospec=True)
59    def test_osd_list_no_match(self, localclient, runner, mm):
60        rebuild.__opts__ = {}
61
62        local = localclient.return_value
63        local.cmd.return_value = {'data1.ceph': [0, 1, 2]}
64        rr = rebuild.Rebuild(['data*.ceph'])
65        result = rr._osd_list('data2.ceph')
66        assert result == None
67
68    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
69    @patch('salt.runner.Runner', autospec=True)
70    @patch('salt.client.LocalClient', autospec=True)
71    def test_validate_osd_df(self, localclient, runner, mm):
72        rebuild.__opts__ = {}
73
74        local = localclient.return_value
75        local.cmd.return_value = {}
76        mm.return_value = 'admin'
77        rr = rebuild.Rebuild(['data*.ceph'])
78
79        osd_df = {'admin': {'nodes': '', 'summary':{'total_kb_avail': '0'}}}
80        result = rr._validate_osd_df(osd_df)
81        assert result == True
82
83    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
84    @patch('salt.runner.Runner', autospec=True)
85    @patch('salt.client.LocalClient', autospec=True)
86    def test_validate_osd_df_missing_mm(self, localclient, runner, mm):
87        rebuild.__opts__ = {}
88
89        local = localclient.return_value
90        local.cmd.return_value = {}
91        mm.return_value = 'admin'
92        rr = rebuild.Rebuild(['data*.ceph'])
93
94        osd_df = {'': {'nodes': '', 'summary':{'total_kb_avail': '0'}}}
95        result = rr._validate_osd_df(osd_df)
96        assert result == False
97
98    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
99    @patch('salt.runner.Runner', autospec=True)
100    @patch('salt.client.LocalClient', autospec=True)
101    def test_validate_osd_df_missing_nodes(self, localclient, runner, mm):
102        rebuild.__opts__ = {}
103
104        local = localclient.return_value
105        local.cmd.return_value = {}
106        mm.return_value = 'admin'
107        rr = rebuild.Rebuild(['data*.ceph'])
108
109        osd_df = {'admin': {'': '', 'summary':{'total_kb_avail': '0'}}}
110        result = rr._validate_osd_df(osd_df)
111        assert result == False
112
113    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
114    @patch('salt.runner.Runner', autospec=True)
115    @patch('salt.client.LocalClient', autospec=True)
116    def test_validate_osd_df_missing_summary(self, localclient, runner, mm):
117        rebuild.__opts__ = {}
118
119        local = localclient.return_value
120        local.cmd.return_value = {}
121        mm.return_value = 'admin'
122        rr = rebuild.Rebuild(['data*.ceph'])
123
124        osd_df = {'admin': {'nodes': '', '':{'total_kb_avail': '0'}}}
125        result = rr._validate_osd_df(osd_df)
126        assert result == False
127
128    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
129    @patch('salt.runner.Runner', autospec=True)
130    @patch('salt.client.LocalClient', autospec=True)
131    def test_validate_osd_df_missing_total(self, localclient, runner, mm):
132        rebuild.__opts__ = {}
133
134        local = localclient.return_value
135        local.cmd.return_value = {}
136        mm.return_value = 'admin'
137        rr = rebuild.Rebuild(['data*.ceph'])
138
139        osd_df = {'admin': {'nodes': '', 'summary':{'': '0'}}}
140        result = rr._validate_osd_df(osd_df)
141        assert result == False
142
143    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
144    @patch('salt.runner.Runner', autospec=True)
145    @patch('salt.client.LocalClient', autospec=True)
146    def test_safe(self, localclient, runner, mm):
147        rebuild.__opts__ = {}
148
149        local = localclient.return_value
150        local.cmd.return_value = {'admin': {'nodes': [
151                                             {'id': 0, 'kb_used': 10},
152                                             {'id': 1, 'kb_used': 10}
153                                             ],
154                                            'summary':{'total_kb_avail': 30}}}
155        mm.return_value = 'admin'
156        rr = rebuild.Rebuild(['data*.ceph'])
157
158        result = rr.safe(['0', '1'])
159        assert result == True
160
161    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
162    @patch('salt.runner.Runner', autospec=True)
163    @patch('salt.client.LocalClient', autospec=True)
164    def test_safe_fails(self, localclient, runner, mm):
165        rebuild.__opts__ = {}
166
167        local = localclient.return_value
168        local.cmd.return_value = {'admin': {'nodes': [
169                                             {'id': 0, 'kb_used': 10},
170                                             {'id': 1, 'kb_used': 10}
171                                             ],
172                                            'summary':{'total_kb_avail': 10}}}
173        mm.return_value = 'admin'
174        rr = rebuild.Rebuild(['data*.ceph'])
175
176        result = rr.safe(['0', '1'])
177        assert result == False
178
179    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
180    @patch('salt.runner.Runner', autospec=True)
181    @patch('salt.client.LocalClient', autospec=True)
182    def test_safe_fails_validation(self, localclient, runner, mm):
183        rebuild.__opts__ = {}
184
185        local = localclient.return_value
186        local.cmd.return_value = {'': {'nodes': [
187                                             {'id': 0, 'kb_used': 10},
188                                             {'id': 1, 'kb_used': 10}
189                                             ],
190                                            'summary':{'total_kb_avail': 10}}}
191        mm.return_value = 'admin'
192        rr = rebuild.Rebuild(['data*.ceph'])
193
194        result = rr.safe(['0', '1'])
195        assert result == False
196
197    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
198    @patch('salt.runner.Runner', autospec=True)
199    @patch('salt.client.LocalClient', autospec=True)
200    def test_check_failed(self, localclient, runner, mm):
201        rebuild.__opts__ = {}
202
203        rr = rebuild.Rebuild(['data*.ceph'])
204
205        ret = {0: True, 1:True}
206        result = rr._check_failed(ret, 'data1.ceph')
207        assert result == False
208        assert rr.skipped == []
209
210    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
211    @patch('salt.runner.Runner', autospec=True)
212    @patch('salt.client.LocalClient', autospec=True)
213    def test_check_failed_finds_failure(self, localclient, runner, mm):
214        rebuild.__opts__ = {}
215
216        rr = rebuild.Rebuild(['data*.ceph'])
217
218        ret = {0: False, 1:True}
219        result = rr._check_failed(ret, 'data1.ceph')
220        assert result == True
221        assert rr.skipped == ['data1.ceph']
222
223    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
224    @patch('salt.runner.Runner', autospec=True)
225    @patch('salt.client.LocalClient', autospec=True)
226    def test_check_failed_finds_error(self, localclient, runner, mm):
227        rebuild.__opts__ = {}
228
229        rr = rebuild.Rebuild(['data*.ceph'])
230
231        ret = "An error message"
232        result = rr._check_failed(ret, 'data1.ceph')
233        assert result == True
234        assert rr.skipped == ['data1.ceph']
235
236    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
237    @patch('salt.runner.Runner', autospec=True)
238    @patch('salt.client.LocalClient', autospec=True)
239    def test_run(self, localclient, runner, mm):
240        rebuild.__opts__ = {}
241
242        rr = rebuild.Rebuild(['data*.ceph'])
243        rr._disengaged = mock.Mock()
244        rr._disengaged.return_value = True
245        rr.minions = ['data1.ceph']
246        rr._osd_list = mock.Mock()
247        rr._osd_list.return_value = [0, 1]
248        rr.safe = mock.Mock()
249        rr.safe.return_value = True
250        rr._busy_wait = mock.Mock()
251        rr.runner.cmd = mock.Mock()
252        rr.runner.cmd.return_value = {}
253        rr._check_failed = mock.Mock()
254        rr._check_failed.return_value = False
255        rr._skipped_summary = mock.Mock()
256
257        result = rr.run()
258        assert result == ""
259        assert rr._osd_list.called
260        assert rr.safe.called
261        assert rr.runner.cmd.called
262        assert rr._check_failed.called
263        assert rr._skipped_summary.called
264
265    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
266    @patch('salt.runner.Runner', autospec=True)
267    @patch('salt.client.LocalClient', autospec=True)
268    def test_run_multiple(self, localclient, runner, mm):
269        rebuild.__opts__ = {}
270
271        rr = rebuild.Rebuild(['data*.ceph'])
272        rr._disengaged = mock.Mock()
273        rr._disengaged.return_value = True
274        rr.minions = ['data1.ceph', 'data2.ceph']
275        rr._osd_list = mock.Mock()
276        rr._osd_list.return_value = [0, 1]
277        rr.safe = mock.Mock()
278        rr.safe.return_value = True
279        rr._busy_wait = mock.Mock()
280        rr.runner.cmd = mock.Mock()
281        rr.runner.cmd.return_value = {}
282        rr._check_failed = mock.Mock()
283        rr._check_failed.return_value = False
284        rr._skipped_summary = mock.Mock()
285
286        result = rr.run()
287        assert result == ""
288        assert rr._osd_list.call_count == 2
289        assert rr.safe.call_count == 2
290        assert rr.runner.cmd.call_count == 4
291        assert rr._check_failed.call_count == 2
292        assert rr._skipped_summary.called
293
294    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
295    @patch('salt.runner.Runner', autospec=True)
296    @patch('salt.client.LocalClient', autospec=True)
297    def test_run_check_only(self, localclient, runner, mm):
298        rebuild.__opts__ = {}
299
300        rr = rebuild.Rebuild(['data*.ceph'])
301        rr._disengaged = mock.Mock()
302        rr._disengaged.return_value = True
303        rr.minions = ['data1.ceph']
304        rr._osd_list = mock.Mock()
305        rr._osd_list.return_value = [0, 1]
306        rr.safe = mock.Mock()
307        rr.safe.return_value = True
308        rr.runner.cmd = mock.Mock()
309        rr.runner.cmd.return_value = {}
310        rr._check_failed = mock.Mock()
311        rr._check_failed.return_value = False
312        rr._skipped_summary = mock.Mock()
313
314        result = rr.run(checkonly=True)
315        assert result == ""
316        assert rr._osd_list.called
317        assert rr.safe.called
318        assert rr.runner.cmd.called is False
319        assert rr._check_failed.called is False
320        assert rr._skipped_summary.called
321
322    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
323    @patch('salt.runner.Runner', autospec=True)
324    @patch('salt.client.LocalClient', autospec=True)
325    def test_run_safety_engaged(self, localclient, runner, mm):
326        rebuild.__opts__ = {}
327
328        rr = rebuild.Rebuild(['data*.ceph'])
329        rr._disengaged = mock.Mock()
330        rr._disengaged.return_value = False
331        rr.minions = ['data1.ceph']
332        rr._osd_list = mock.Mock()
333        rr._osd_list.return_value = [0, 1]
334        rr.safe = mock.Mock()
335        rr.safe.return_value = True
336        rr.runner.cmd = mock.Mock()
337        rr.runner.cmd.return_value = {}
338        rr._check_failed = mock.Mock()
339        rr._check_failed.return_value = False
340        rr._skipped_summary = mock.Mock()
341
342        result = rr.run()
343        assert result == ""
344        assert rr._osd_list.called is False
345        assert rr.safe.called is False
346        assert rr.runner.cmd.called is False
347        assert rr._check_failed.called is False
348        assert rr._skipped_summary.called is False
349
350    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
351    @patch('salt.runner.Runner', autospec=True)
352    @patch('salt.client.LocalClient', autospec=True)
353    def test_run_no_remove(self, localclient, runner, mm):
354        rebuild.__opts__ = {}
355
356        rr = rebuild.Rebuild(['data*.ceph'])
357        rr._disengaged = mock.Mock()
358        rr._disengaged.return_value = True
359        rr.minions = ['data1.ceph']
360        rr._osd_list = mock.Mock()
361        rr._osd_list.return_value = []
362        rr.safe = mock.Mock()
363        rr.safe.return_value = True
364        rr.runner.cmd = mock.Mock()
365        rr.runner.cmd.return_value = {}
366        rr._check_failed = mock.Mock()
367        rr._check_failed.return_value = False
368        rr._skipped_summary = mock.Mock()
369
370        result = rr.run()
371        assert result == ""
372        assert rr._osd_list.called
373        assert rr.safe.called is False
374        assert rr.runner.cmd.called
375        assert rr._check_failed.called is False
376        assert rr._skipped_summary.called
377
378    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
379    @patch('salt.runner.Runner', autospec=True)
380    @patch('salt.client.LocalClient', autospec=True)
381    def test_run_not_safe(self, localclient, runner, mm):
382        rebuild.__opts__ = {}
383
384        rr = rebuild.Rebuild(['data*.ceph'])
385        rr._disengaged = mock.Mock()
386        rr._disengaged.return_value = True
387        rr.minions = ['data1.ceph']
388        rr._osd_list = mock.Mock()
389        rr._osd_list.return_value = [0, 1]
390        rr.safe = mock.Mock()
391        rr.safe.return_value = False
392        rr.runner.cmd = mock.Mock()
393        rr.runner.cmd.return_value = {}
394        rr._check_failed = mock.Mock()
395        rr._check_failed.return_value = False
396        rr._skipped_summary = mock.Mock()
397
398        result = rr.run()
399        assert result == ""
400        assert rr._osd_list.called
401        assert rr.safe.called
402        assert rr.runner.cmd.called is False
403        assert rr._check_failed.called is False
404        assert rr._skipped_summary.called is False
405
406    @patch('srv.modules.runners.rebuild.master_minion', autospec=True)
407    @patch('salt.runner.Runner', autospec=True)
408    @patch('salt.client.LocalClient', autospec=True)
409    def test_run_remove_failed(self, localclient, runner, mm):
410        rebuild.__opts__ = {}
411
412        rr = rebuild.Rebuild(['data*.ceph'])
413        rr._disengaged = mock.Mock()
414        rr._disengaged.return_value = True
415        rr.minions = ['data1.ceph']
416        rr._osd_list = mock.Mock()
417        rr._osd_list.return_value = [0, 1]
418        rr.safe = mock.Mock()
419        rr.safe.return_value = True
420        rr._busy_wait = mock.Mock()
421        rr.runner.cmd = mock.Mock()
422        rr.runner.cmd.return_value = {}
423        rr._check_failed = mock.Mock()
424        rr._check_failed.return_value = True
425        rr._skipped_summary = mock.Mock()
426
427        result = rr.run()
428        assert result == ""
429        assert rr._osd_list.called
430        assert rr.safe.called
431        assert rr.runner.cmd.called
432        assert rr._check_failed.called
433        assert rr._skipped_summary.called
434
435
436
Full Screen

test_rebuild_server.py

Source: test_rebuild_server.py Github

copy
1# -*- coding: utf-8 -*-
2
3# Licensed under the Apache License, Version 2.0 (the "License"); you may
4# not use this file except in compliance with the License. You may obtain
5# a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations
13# under the License.
14
15"""
16test_rebuild_server
17----------------------------------
18
19Tests for the `rebuild_server` command.
20"""
21
22from mock import patch, Mock
23import os_client_config
24from shade import _utils
25from shade import meta
26from shade import OpenStackCloud
27from shade.exc import (OpenStackCloudException, OpenStackCloudTimeout)
28from shade.tests import base, fakes
29
30
31class TestRebuildServer(base.TestCase):
32
33    def setUp(self):
34        super(TestRebuildServer, self).setUp()
35        config = os_client_config.OpenStackConfig()
36        self.client = OpenStackCloud(
37            cloud_config=config.get_one_cloud(validate=False))
38
39    def test_rebuild_server_rebuild_exception(self):
40        """
41        Test that an exception in the novaclient rebuild raises an exception in
42        rebuild_server.
43        """
44        with patch("shade.OpenStackCloud"):
45            config = {
46                "servers.rebuild.side_effect": Exception("exception"),
47            }
48            OpenStackCloud.nova_client = Mock(**config)
49            self.assertRaises(
50                OpenStackCloudException, self.client.rebuild_server, "a", "b")
51
52    def test_rebuild_server_server_error(self):
53        """
54        Test that a server error while waiting for the server to rebuild
55        raises an exception in rebuild_server.
56        """
57        rebuild_server = fakes.FakeServer('1234', '', 'REBUILD')
58        error_server = fakes.FakeServer('1234', '', 'ERROR')
59        with patch("shade.OpenStackCloud"):
60            config = {
61                "servers.rebuild.return_value": rebuild_server,
62                "servers.get.return_value": error_server,
63            }
64            OpenStackCloud.nova_client = Mock(**config)
65            self.assertRaises(
66                OpenStackCloudException,
67                self.client.rebuild_server, "a", "b", wait=True)
68
69    def test_rebuild_server_timeout(self):
70        """
71        Test that a timeout while waiting for the server to rebuild raises an
72        exception in rebuild_server.
73        """
74        rebuild_server = fakes.FakeServer('1234', '', 'REBUILD')
75        with patch("shade.OpenStackCloud"):
76            config = {
77                "servers.rebuild.return_value": rebuild_server,
78                "servers.get.return_value": rebuild_server,
79            }
80            OpenStackCloud.nova_client = Mock(**config)
81            self.assertRaises(
82                OpenStackCloudTimeout,
83                self.client.rebuild_server, "a", "b", wait=True, timeout=0.001)
84
85    def test_rebuild_server_no_wait(self):
86        """
87        Test that rebuild_server with no wait and no exception in the
88        novaclient rebuild call returns the server instance.
89        """
90        with patch("shade.OpenStackCloud"):
91            rebuild_server = fakes.FakeServer('1234', '', 'REBUILD')
92            config = {
93                "servers.rebuild.return_value": rebuild_server
94            }
95            OpenStackCloud.nova_client = Mock(**config)
96            self.assertEqual(meta.obj_to_dict(rebuild_server),
97                             self.client.rebuild_server("a", "b"))
98
99    def test_rebuild_server_with_admin_pass_no_wait(self):
100        """
101        Test that a server with an admin_pass passed returns the password
102        """
103        with patch("shade.OpenStackCloud"):
104            rebuild_server = fakes.FakeServer('1234', '', 'REBUILD',
105                                              adminPass='ooBootheiX0edoh')
106            config = {
107                "servers.rebuild.return_value": rebuild_server,
108            }
109            OpenStackCloud.nova_client = Mock(**config)
110            self.assertEqual(
111                meta.obj_to_dict(rebuild_server),
112                self.client.rebuild_server('a', 'b',
113                                           admin_pass='ooBootheiX0edoh'))
114
115    def test_rebuild_server_with_admin_pass_wait(self):
116        """
117        Test that a server with an admin_pass passed returns the password
118        """
119        with patch("shade.OpenStackCloud"):
120            rebuild_server = fakes.FakeServer('1234', '', 'REBUILD',
121                                              adminPass='ooBootheiX0edoh')
122            active_server = fakes.FakeServer('1234', '', 'ACTIVE')
123            ret_active_server = fakes.FakeServer('1234', '', 'ACTIVE',
124                                                 adminPass='ooBootheiX0edoh')
125            config = {
126                "servers.rebuild.return_value": rebuild_server,
127                "servers.get.return_value": active_server,
128            }
129            OpenStackCloud.nova_client = Mock(**config)
130            self.client.name = 'cloud-name'
131            self.assertEqual(
132                _utils.normalize_server(
133                    meta.obj_to_dict(ret_active_server),
134                    cloud_name='cloud-name', region_name=''),
135                self.client.rebuild_server("a", "b", wait=True,
136                                           admin_pass='ooBootheiX0edoh'))
137
138    def test_rebuild_server_wait(self):
139        """
140        Test that rebuild_server with a wait returns the server instance when
141        its status changes to "ACTIVE".
142        """
143        with patch("shade.OpenStackCloud"):
144            rebuild_server = fakes.FakeServer('1234', '', 'REBUILD')
145            active_server = fakes.FakeServer('1234', '', 'ACTIVE')
146            config = {
147                "servers.rebuild.return_value": rebuild_server,
148                "servers.get.return_value": active_server
149            }
150            OpenStackCloud.nova_client = Mock(**config)
151            self.client.name = 'cloud-name'
152            self.assertEqual(
153                _utils.normalize_server(
154                    meta.obj_to_dict(active_server),
155                    cloud_name='cloud-name', region_name=''),
156                self.client.rebuild_server("a", "b", wait=True))
157
Full Screen

_sstatetests_noauto.py

Source: _sstatetests_noauto.py Github

copy
1import datetime
2import unittest
3import os
4import re
5import shutil
6
7import oeqa.utils.ftools as ftools
8from oeqa.selftest.base import oeSelfTest
9from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_test_layer
10from oeqa.selftest.sstate import SStateBase
11
12
13class RebuildFromSState(SStateBase):
14
15    @classmethod
16    def setUpClass(self):
17        self.builddir = os.path.join(os.environ.get('BUILDDIR'))
18
19    def get_dep_targets(self, primary_targets):
20        found_targets = []
21        bitbake("-g " + ' '.join(map(str, primary_targets)))
22        with open(os.path.join(self.builddir, 'pn-buildlist'), 'r') as pnfile:
23            found_targets = pnfile.read().splitlines()
24        return found_targets
25
26    def configure_builddir(self, builddir):
27        os.mkdir(builddir)
28        self.track_for_cleanup(builddir)
29        os.mkdir(os.path.join(builddir, 'conf'))
30        shutil.copyfile(os.path.join(os.environ.get('BUILDDIR'), 'conf/local.conf'), os.path.join(builddir, 'conf/local.conf'))
31        config = {}
32        config['default_sstate_dir'] = "SSTATE_DIR ?= \"${TOPDIR}/sstate-cache\""
33        config['null_sstate_mirrors'] = "SSTATE_MIRRORS = \"\""
34        config['default_tmp_dir'] = "TMPDIR = \"${TOPDIR}/tmp\""
35        for key in config:
36            ftools.append_file(os.path.join(builddir, 'conf/selftest.inc'), config[key])
37        shutil.copyfile(os.path.join(os.environ.get('BUILDDIR'), 'conf/bblayers.conf'), os.path.join(builddir, 'conf/bblayers.conf'))
38        try:
39            shutil.copyfile(os.path.join(os.environ.get('BUILDDIR'), 'conf/auto.conf'), os.path.join(builddir, 'conf/auto.conf'))
40        except:
41            pass
42
43    def hardlink_tree(self, src, dst):
44        os.mkdir(dst)
45        self.track_for_cleanup(dst)
46        for root, dirs, files in os.walk(src):
47            if root == src:
48                continue
49            os.mkdir(os.path.join(dst, root.split(src)[1][1:]))
50            for sstate_file in files:
51                os.link(os.path.join(root, sstate_file), os.path.join(dst, root.split(src)[1][1:], sstate_file))
52
53    def run_test_sstate_rebuild(self, primary_targets, relocate=False, rebuild_dependencies=False):
54        buildA = os.path.join(self.builddir, 'buildA')
55        if relocate:
56            buildB = os.path.join(self.builddir, 'buildB')
57        else:
58            buildB = buildA
59
60        if rebuild_dependencies:
61            rebuild_targets = self.get_dep_targets(primary_targets)
62        else:
63            rebuild_targets = primary_targets
64
65        self.configure_builddir(buildA)
66        runCmd((". %s/oe-init-build-env %s && " % (get_bb_var('COREBASE'), buildA)) + 'bitbake  ' + ' '.join(map(str, primary_targets)), shell=True, executable='/bin/bash')
67        self.hardlink_tree(os.path.join(buildA, 'sstate-cache'), os.path.join(self.builddir, 'sstate-cache-buildA'))
68        shutil.rmtree(buildA)
69
70        failed_rebuild = []
71        failed_cleansstate = []
72        for target in rebuild_targets:
73            self.configure_builddir(buildB)
74            self.hardlink_tree(os.path.join(self.builddir, 'sstate-cache-buildA'), os.path.join(buildB, 'sstate-cache'))
75
76            result_cleansstate = runCmd((". %s/oe-init-build-env %s && " % (get_bb_var('COREBASE'), buildB)) + 'bitbake -ccleansstate ' + target, ignore_status=True, shell=True, executable='/bin/bash')
77            if not result_cleansstate.status == 0:
78                failed_cleansstate.append(target)
79                shutil.rmtree(buildB)
80                continue
81
82            result_build = runCmd((". %s/oe-init-build-env %s && " % (get_bb_var('COREBASE'), buildB)) + 'bitbake ' + target, ignore_status=True, shell=True, executable='/bin/bash')
83            if not result_build.status == 0:
84                failed_rebuild.append(target)
85
86            shutil.rmtree(buildB)
87
88        self.assertFalse(failed_rebuild, msg="The following recipes have failed to rebuild: %s" % ' '.join(map(str, failed_rebuild)))
89        self.assertFalse(failed_cleansstate, msg="The following recipes have failed cleansstate(all others have passed both cleansstate and rebuild from sstate tests): %s" % ' '.join(map(str, failed_cleansstate)))
90
91    def test_sstate_relocation(self):
92        self.run_test_sstate_rebuild(['core-image-sato-sdk'], relocate=True, rebuild_dependencies=True)
93
94    def test_sstate_rebuild(self):
95        self.run_test_sstate_rebuild(['core-image-sato-sdk'], relocate=False, rebuild_dependencies=True)
96
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Behave on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)