Best Python code snippet using autotest_python
base_job_unittest.py
Source:base_job_unittest.py
...33 def read_from_file(self, file_path):34 pass35 def write_to_file(self, file_path):36 pass37 def set_backing_file(self, file_path):38 pass39 def _read_from_backing_file(self):40 pass41 def _write_to_backing_file(self):42 pass43 def _lock_backing_file(self):44 pass45 def _unlock_backing_file(self):46 pass47class test_init(unittest.TestCase):48 class generic_tests(object):49 """50 Generic tests for any implementation of __init__.51 Expectations:52 A self.job attribute where self.job is a __new__'ed instance of53 the job class to be tested, but not yet __init__'ed.54 A self.call_init method that will config the appropriate mocks55 and then call job.__init__. It should undo any mocks it created56 afterwards.57 """58 PUBLIC_ATTRIBUTES = set([59 # standard directories60 'autodir', 'clientdir', 'serverdir', 'resultdir', 'pkgdir',61 'tmpdir', 'testdir', 'site_testdir', 'bindir',62 'profdir', 'toolsdir',63 # other special attributes64 'args', 'automatic_test_tag', 'control',65 'default_profile_only', 'drop_caches',66 'drop_caches_between_iterations', 'harness', 'hosts',67 'logging', 'machines', 'num_tests_failed', 'num_tests_run',68 'pkgmgr', 'profilers', 'resultdir', 'run_test_cleanup',69 'sysinfo', 'tag', 'user', 'use_sequence_number',70 'warning_loggers', 'warning_manager', 'label', 'test_retry',71 'parent_job_id', 'in_lab', 'machine_dict_list',72 'max_result_size_KB', 'fast'73 ])74 OPTIONAL_ATTRIBUTES = set([75 'serverdir',76 'automatic_test_tag', 'control', 'harness', 'num_tests_run',77 'num_tests_failed', 'tag', 'warning_manager',78 'warning_loggers', 'label', 'test_retry', 'parent_job_id',79 'max_result_size_KB', 'fast'80 ])81 OPTIONAL_ATTRIBUTES_DEVICE_ERROR = set(['failed_with_device_error'])82 def test_public_attributes_initialized(self):83 # only the known public attributes should be there after __init__84 self.call_init()85 public_attributes = set(attr for attr in dir(self.job)86 if not attr.startswith('_')87 and not callable(getattr(self.job, attr)))88 expected_attributes = self.PUBLIC_ATTRIBUTES89 missing_attributes = expected_attributes - public_attributes90 self.assertEqual(missing_attributes, set([]),91 'Missing attributes: %s' %92 ', '.join(sorted(missing_attributes)))93 extra_attributes = (public_attributes - expected_attributes -94 self.OPTIONAL_ATTRIBUTES_DEVICE_ERROR)95 self.assertEqual(extra_attributes, set([]),96 'Extra public attributes found: %s' %97 ', '.join(sorted(extra_attributes)))98 def test_required_attributes_not_none(self):99 required_attributes = (self.PUBLIC_ATTRIBUTES -100 self.OPTIONAL_ATTRIBUTES)101 self.call_init()102 for attribute in required_attributes:103 self.assertNotEqual(getattr(self.job, attribute, None), None,104 'job.%s is None but is not optional'105 % attribute)106class test_find_base_directories(unittest.TestCase):107 class generic_tests(object):108 """109 Generic tests for any implementation of _find_base_directories.110 Expectations:111 A self.job attribute where self.job is an instance of the job112 class to be tested.113 """114 def test_autodir_is_not_none(self):115 auto, client, server = self.job._find_base_directories()116 self.assertNotEqual(auto, None)117 def test_clientdir_is_not_none(self):118 auto, client, server = self.job._find_base_directories()119 self.assertNotEqual(client, None)120class test_initialize_dir_properties(unittest.TestCase):121 def make_job(self, autodir, server):122 job = base_job.base_job.__new__(base_job.base_job)123 job._job_directory = stub_job_directory124 job._autodir = stub_job_directory(autodir)125 if server:126 job._clientdir = stub_job_directory(127 os.path.join(autodir, 'client'))128 job._serverdir = stub_job_directory(129 os.path.join(autodir, 'server'))130 else:131 job._clientdir = stub_job_directory(job.autodir)132 job._serverdir = None133 return job134 def setUp(self):135 self.cjob = self.make_job('/atest/client', False)136 self.sjob = self.make_job('/atest', True)137 def test_always_client_dirs(self):138 self.cjob._initialize_dir_properties()139 self.sjob._initialize_dir_properties()140 # check all the always-client dir properties141 self.assertEqual(self.cjob.bindir, self.sjob.bindir)142 self.assertEqual(self.cjob.profdir, self.sjob.profdir)143 self.assertEqual(self.cjob.pkgdir, self.sjob.pkgdir)144 def test_dynamic_dirs(self):145 self.cjob._initialize_dir_properties()146 self.sjob._initialize_dir_properties()147 # check all the context-specifc dir properties148 self.assert_(self.cjob.tmpdir.startswith('/atest/client'))149 self.assert_(self.cjob.testdir.startswith('/atest/client'))150 self.assert_(self.cjob.site_testdir.startswith('/atest/client'))151 self.assertEqual(self.sjob.tmpdir, tempfile.gettempdir())152 self.assert_(self.sjob.testdir.startswith('/atest/server'))153 self.assert_(self.sjob.site_testdir.startswith('/atest/server'))154class test_execution_context(unittest.TestCase):155 def setUp(self):156 clientdir = os.path.abspath(os.path.join(__file__, '..', '..'))157 self.resultdir = tempfile.mkdtemp(suffix='unittest')158 self.job = base_job.base_job.__new__(base_job.base_job)159 self.job._find_base_directories = lambda: (clientdir, clientdir, None)160 self.job._find_resultdir = lambda *_: self.resultdir161 self.job.__init__()162 def tearDown(self):163 shutil.rmtree(self.resultdir, ignore_errors=True)164 def test_pop_fails_without_push(self):165 self.assertRaises(IndexError, self.job.pop_execution_context)166 def test_push_changes_to_subdir(self):167 sub1 = os.path.join(self.resultdir, 'sub1')168 os.mkdir(sub1)169 self.job.push_execution_context('sub1')170 self.assertEqual(self.job.resultdir, sub1)171 def test_push_creates_subdir(self):172 sub2 = os.path.join(self.resultdir, 'sub2')173 self.job.push_execution_context('sub2')174 self.assertEqual(self.job.resultdir, sub2)175 self.assert_(os.path.exists(sub2))176 def test_push_handles_absolute_paths(self):177 otherresults = tempfile.mkdtemp(suffix='unittest')178 try:179 self.job.push_execution_context(otherresults)180 self.assertEqual(self.job.resultdir, otherresults)181 finally:182 shutil.rmtree(otherresults, ignore_errors=True)183 def test_pop_restores_context(self):184 sub3 = os.path.join(self.resultdir, 'sub3')185 self.job.push_execution_context('sub3')186 self.assertEqual(self.job.resultdir, sub3)187 self.job.pop_execution_context()188 self.assertEqual(self.job.resultdir, self.resultdir)189 def test_push_and_pop_are_fifo(self):190 sub4 = os.path.join(self.resultdir, 'sub4')191 subsub = os.path.join(sub4, 'subsub')192 self.job.push_execution_context('sub4')193 self.assertEqual(self.job.resultdir, sub4)194 self.job.push_execution_context('subsub')195 self.assertEqual(self.job.resultdir, subsub)196 self.job.pop_execution_context()197 self.assertEqual(self.job.resultdir, sub4)198 self.job.pop_execution_context()199 self.assertEqual(self.job.resultdir, self.resultdir)200class test_job_directory(unittest.TestCase):201 def setUp(self):202 self.testdir = tempfile.mkdtemp(suffix='unittest')203 self.original_wd = os.getcwd()204 os.chdir(self.testdir)205 def tearDown(self):206 os.chdir(self.original_wd)207 shutil.rmtree(self.testdir, ignore_errors=True)208 def test_passes_if_dir_exists(self):209 os.mkdir('testing')210 self.assert_(os.path.isdir('testing'))211 jd = base_job.job_directory('testing')212 self.assert_(os.path.isdir('testing'))213 def test_fails_if_not_writable_and_dir_doesnt_exist(self):214 self.assert_(not os.path.isdir('testing2'))215 self.assertRaises(base_job.job_directory.MissingDirectoryException,216 base_job.job_directory, 'testing2')217 def test_fails_if_file_already_exists(self):218 open('testing3', 'w').close()219 self.assert_(os.path.isfile('testing3'))220 self.assertRaises(base_job.job_directory.MissingDirectoryException,221 base_job.job_directory, 'testing3')222 def test_passes_if_writable_and_dir_exists(self):223 os.mkdir('testing4')224 self.assert_(os.path.isdir('testing4'))225 jd = base_job.job_directory('testing4', True)226 self.assert_(os.path.isdir('testing4'))227 def test_creates_dir_if_writable_and_dir_doesnt_exist(self):228 self.assert_(not os.path.isdir('testing5'))229 jd = base_job.job_directory('testing5', True)230 self.assert_(os.path.isdir('testing5'))231 def test_recursive_creates_dir_if_writable_and_dir_doesnt_exist(self):232 self.assert_(not os.path.isdir('testing6'))233 base_job.job_directory('testing6/subdir', True)234 self.assert_(os.path.isdir('testing6/subdir'))235 def test_fails_if_writable_and_file_exists(self):236 open('testing7', 'w').close()237 self.assert_(os.path.isfile('testing7'))238 self.assert_(not os.path.isdir('testing7'))239 self.assertRaises(base_job.job_directory.UncreatableDirectoryException,240 base_job.job_directory, 'testing7', True)241 def test_fails_if_writable_and_no_permission_to_create(self):242 os.mkdir('testing8', 0555)243 self.assert_(os.path.isdir('testing8'))244 self.assertRaises(base_job.job_directory.UncreatableDirectoryException,245 base_job.job_directory, 'testing8/subdir', True)246 def test_passes_if_not_is_writable_and_dir_not_writable(self):247 os.mkdir('testing9', 0555)248 self.assert_(os.path.isdir('testing9'))249 self.assert_(not os.access('testing9', os.W_OK))250 jd = base_job.job_directory('testing9')251 def test_fails_if_is_writable_but_dir_not_writable(self):252 os.mkdir('testing10', 0555)253 self.assert_(os.path.isdir('testing10'))254 self.assert_(not os.access('testing10', os.W_OK))255 self.assertRaises(base_job.job_directory.UnwritableDirectoryException,256 base_job.job_directory, 'testing10', True)257 def test_fails_if_no_path_and_not_writable(self):258 self.assertRaises(base_job.job_directory.MissingDirectoryException,259 base_job.job_directory, None)260 def test_no_path_and_and_not_writable_creates_tempdir(self):261 jd = base_job.job_directory(None, True)262 self.assert_(os.path.isdir(jd.path))263 self.assert_(os.access(jd.path, os.W_OK))264 temp_path = jd.path265 del jd266 self.assert_(not os.path.isdir(temp_path))267class test_job_state(unittest.TestCase):268 def setUp(self):269 self.state = base_job.job_state()270 def test_undefined_name_returns_key_error(self):271 self.assertRaises(KeyError, self.state.get, 'ns1', 'missing_name')272 def test_undefined_name_returns_default(self):273 self.assertEqual(42, self.state.get('ns2', 'missing_name', default=42))274 def test_none_is_valid_default(self):275 self.assertEqual(None, self.state.get('ns3', 'missing_name',276 default=None))277 def test_get_returns_set_values(self):278 self.state.set('ns4', 'name1', 50)279 self.assertEqual(50, self.state.get('ns4', 'name1'))280 def test_get_ignores_default_when_value_is_defined(self):281 self.state.set('ns5', 'name2', 55)282 self.assertEqual(55, self.state.get('ns5', 'name2', default=45))283 def test_set_only_sets_one_value(self):284 self.state.set('ns6', 'name3', 50)285 self.assertEqual(50, self.state.get('ns6', 'name3'))286 self.assertRaises(KeyError, self.state.get, 'ns6', 'name4')287 def test_set_works_with_multiple_names(self):288 self.state.set('ns7', 'name5', 60)289 self.state.set('ns7', 'name6', 70)290 self.assertEquals(60, self.state.get('ns7', 'name5'))291 self.assertEquals(70, self.state.get('ns7', 'name6'))292 def test_multiple_sets_override_each_other(self):293 self.state.set('ns8', 'name7', 10)294 self.state.set('ns8', 'name7', 25)295 self.assertEquals(25, self.state.get('ns8', 'name7'))296 def test_get_with_default_does_not_set(self):297 self.assertEquals(100, self.state.get('ns9', 'name8', default=100))298 self.assertRaises(KeyError, self.state.get, 'ns9', 'name8')299 def test_set_in_one_namespace_ignores_other(self):300 self.state.set('ns10', 'name9', 200)301 self.assertEquals(200, self.state.get('ns10', 'name9'))302 self.assertRaises(KeyError, self.state.get, 'ns11', 'name9')303 def test_namespaces_do_not_collide(self):304 self.state.set('ns12', 'name10', 250)305 self.state.set('ns13', 'name10', -150)306 self.assertEquals(-150, self.state.get('ns13', 'name10'))307 self.assertEquals(250, self.state.get('ns12', 'name10'))308 def test_discard_does_nothing_on_undefined_namespace(self):309 self.state.discard('missing_ns', 'missing')310 self.assertRaises(KeyError, self.state.get, 'missing_ns', 'missing')311 def test_discard_does_nothing_on_missing_name(self):312 self.state.set('ns14', 'name20', 111)313 self.state.discard('ns14', 'missing')314 self.assertEqual(111, self.state.get('ns14', 'name20'))315 self.assertRaises(KeyError, self.state.get, 'ns14', 'missing')316 def test_discard_deletes_name(self):317 self.state.set('ns15', 'name21', 4567)318 self.assertEqual(4567, self.state.get('ns15', 'name21'))319 self.state.discard('ns15', 'name21')320 self.assertRaises(KeyError, self.state.get, 'ns15', 'name21')321 def test_discard_doesnt_touch_other_values(self):322 self.state.set('ns16_1', 'name22', 'val1')323 self.state.set('ns16_1', 'name23', 'val2')324 self.state.set('ns16_2', 'name23', 'val3')325 self.assertEqual('val1', self.state.get('ns16_1', 'name22'))326 self.assertEqual('val3', self.state.get('ns16_2', 'name23'))327 self.state.discard('ns16_1', 'name23')328 self.assertEqual('val1', self.state.get('ns16_1', 'name22'))329 self.assertEqual('val3', self.state.get('ns16_2', 'name23'))330 def test_has_is_true_for_all_set_values(self):331 self.state.set('ns17_1', 'name24', 1)332 self.state.set('ns17_1', 'name25', 2)333 self.state.set('ns17_2', 'name25', 3)334 self.assert_(self.state.has('ns17_1', 'name24'))335 self.assert_(self.state.has('ns17_1', 'name25'))336 self.assert_(self.state.has('ns17_2', 'name25'))337 def test_has_is_false_for_all_unset_values(self):338 self.state.set('ns18_1', 'name26', 1)339 self.state.set('ns18_1', 'name27', 2)340 self.state.set('ns18_2', 'name27', 3)341 self.assert_(not self.state.has('ns18_2', 'name26'))342 def test_discard_namespace_drops_all_values(self):343 self.state.set('ns19', 'var1', 10)344 self.state.set('ns19', 'var3', 100)345 self.state.discard_namespace('ns19')346 self.assertRaises(KeyError, self.state.get, 'ns19', 'var1')347 self.assertRaises(KeyError, self.state.get, 'ns19', 'var3')348 def test_discard_namespace_works_on_missing_namespace(self):349 self.state.discard_namespace('missing_ns')350 def test_discard_namespace_doesnt_touch_other_values(self):351 self.state.set('ns20', 'var1', 20)352 self.state.set('ns20', 'var2', 200)353 self.state.set('ns21', 'var2', 21)354 self.state.discard_namespace('ns20')355 self.assertEqual(21, self.state.get('ns21', 'var2'))356# run the same tests as test_job_state, but with a backing file turned on357# also adds some tests to check that each method is persistent358class test_job_state_with_backing_file(test_job_state):359 def setUp(self):360 self.backing_file = tempfile.mktemp()361 self.state = base_job.job_state()362 self.state.set_backing_file(self.backing_file)363 def tearDown(self):364 if os.path.exists(self.backing_file):365 os.remove(self.backing_file)366 def test_set_is_persistent(self):367 self.state.set('persist', 'var', 'value')368 written_state = base_job.job_state()369 written_state.read_from_file(self.backing_file)370 self.assertEqual('value', written_state.get('persist', 'var'))371 def test_discard_is_persistent(self):372 self.state.set('persist', 'var', 'value')373 self.state.discard('persist', 'var')374 written_state = base_job.job_state()375 written_state.read_from_file(self.backing_file)376 self.assertRaises(KeyError, written_state.get, 'persist', 'var')377 def test_discard_namespace_is_persistent(self):378 self.state.set('persist', 'var', 'value')379 self.state.discard_namespace('persist')380 written_state = base_job.job_state()381 written_state.read_from_file(self.backing_file)382 self.assertRaises(KeyError, written_state.get, 'persist', 'var')383class test_job_state_read_write_file(unittest.TestCase):384 def setUp(self):385 self.testdir = tempfile.mkdtemp(suffix='unittest')386 self.original_wd = os.getcwd()387 os.chdir(self.testdir)388 def tearDown(self):389 os.chdir(self.original_wd)390 shutil.rmtree(self.testdir, ignore_errors=True)391 def test_write_read_transfers_all_state(self):392 state1 = base_job.job_state()393 state1.set('ns1', 'var0', 50)394 state1.set('ns2', 'var10', 100)395 state1.write_to_file('transfer_file')396 state2 = base_job.job_state()397 self.assertRaises(KeyError, state2.get, 'ns1', 'var0')398 self.assertRaises(KeyError, state2.get, 'ns2', 'var10')399 state2.read_from_file('transfer_file')400 self.assertEqual(50, state2.get('ns1', 'var0'))401 self.assertEqual(100, state2.get('ns2', 'var10'))402 def test_read_overwrites_in_memory(self):403 state = base_job.job_state()404 state.set('ns', 'myvar', 'hello')405 state.write_to_file('backup')406 state.set('ns', 'myvar', 'goodbye')407 self.assertEqual('goodbye', state.get('ns', 'myvar'))408 state.read_from_file('backup')409 self.assertEqual('hello', state.get('ns', 'myvar'))410 def test_read_updates_persistent_file(self):411 state1 = base_job.job_state()412 state1.set('ns', 'var1', 'value1')413 state1.write_to_file('to_be_read')414 state2 = base_job.job_state()415 state2.set_backing_file('backing_file')416 state2.set('ns', 'var2', 'value2')417 state2.read_from_file('to_be_read')418 state2.set_backing_file(None)419 state3 = base_job.job_state()420 state3.read_from_file('backing_file')421 self.assertEqual('value1', state3.get('ns', 'var1'))422 self.assertEqual('value2', state3.get('ns', 'var2'))423 def test_read_without_merge(self):424 state = base_job.job_state()425 state.set('ns', 'myvar1', 'hello')426 state.write_to_file('backup')427 state.discard('ns', 'myvar1')428 state.set('ns', 'myvar2', 'goodbye')429 self.assertFalse(state.has('ns', 'myvar1'))430 self.assertEqual('goodbye', state.get('ns', 'myvar2'))431 state.read_from_file('backup', merge=False)432 self.assertEqual('hello', state.get('ns', 'myvar1'))433 self.assertFalse(state.has('ns', 'myvar2'))434class test_job_state_set_backing_file(unittest.TestCase):435 def setUp(self):436 self.testdir = tempfile.mkdtemp(suffix='unittest')437 self.original_wd = os.getcwd()438 os.chdir(self.testdir)439 def tearDown(self):440 os.chdir(self.original_wd)441 shutil.rmtree(self.testdir, ignore_errors=True)442 def test_writes_to_file(self):443 state = base_job.job_state()444 state.set_backing_file('outfile1')445 self.assert_(os.path.exists('outfile1'))446 def test_set_backing_file_updates_existing_file(self):447 state1 = base_job.job_state()448 state1.set_backing_file('second_file')449 state1.set('ns0', 'var1x', 100)450 state1.set_backing_file(None)451 state2 = base_job.job_state()452 state2.set_backing_file('first_file')453 state2.set('ns0', 'var0x', 0)454 state2.set_backing_file('second_file')455 state2.set_backing_file(None)456 state3 = base_job.job_state()457 state3.read_from_file('second_file')458 self.assertEqual(0, state3.get('ns0', 'var0x'))459 self.assertEqual(100, state3.get('ns0', 'var1x'))460 def test_set_backing_file_does_not_overwrite_previous_backing_file(self):461 state1 = base_job.job_state()462 state1.set_backing_file('second_file')463 state1.set('ns0', 'var1y', 10)464 state1.set_backing_file(None)465 state2 = base_job.job_state()466 state2.set_backing_file('first_file')467 state2.set('ns0', 'var0y', -10)468 state2.set_backing_file('second_file')469 state2.set_backing_file(None)470 state3 = base_job.job_state()471 state3.read_from_file('first_file')472 self.assertEqual(-10, state3.get('ns0', 'var0y'))473 self.assertRaises(KeyError, state3.get, 'ns0', 'var1y')474 def test_writes_stop_after_backing_file_removed(self):475 state = base_job.job_state()476 state.set('ns', 'var1', 'value1')477 state.set_backing_file('outfile2')478 state.set_backing_file(None)479 os.remove('outfile2')480 state.set('n2', 'var2', 'value2')481 self.assert_(not os.path.exists('outfile2'))482 def test_written_files_can_be_reloaded(self):483 state1 = base_job.job_state()484 state1.set_backing_file('outfile3')485 state1.set('n3', 'var1', 67)486 state1.set_backing_file(None)487 state2 = base_job.job_state()488 self.assertRaises(KeyError, state2.get, 'n3', 'var1')489 state2.set_backing_file('outfile3')490 self.assertEqual(67, state2.get('n3', 'var1'))491 def test_backing_file_overrides_in_memory_values(self):492 state1 = base_job.job_state()493 state1.set_backing_file('outfile4')494 state1.set('n4', 'var1', 42)495 state1.set_backing_file(None)496 state2 = base_job.job_state()497 state2.set('n4', 'var1', 430)498 self.assertEqual(430, state2.get('n4', 'var1'))499 state2.set_backing_file('outfile4')500 self.assertEqual(42, state2.get('n4', 'var1'))501 def test_backing_file_only_overrides_values_it_defines(self):502 state1 = base_job.job_state()503 state1.set_backing_file('outfile5')504 state1.set('n5', 'var1', 123)505 state1.set_backing_file(None)506 state2 = base_job.job_state()507 state2.set('n5', 'var2', 456)508 state2.set_backing_file('outfile5')509 self.assertEqual(123, state2.get('n5', 'var1'))510 self.assertEqual(456, state2.get('n5', 'var2'))511 def test_shared_backing_file_propagates_state_to_get(self):512 state1 = base_job.job_state()513 state1.set_backing_file('outfile6')514 state2 = base_job.job_state()515 state2.set_backing_file('outfile6')516 self.assertRaises(KeyError, state1.get, 'n6', 'shared1')517 self.assertRaises(KeyError, state2.get, 'n6', 'shared1')518 state1.set('n6', 'shared1', 345)519 self.assertEqual(345, state1.get('n6', 'shared1'))520 self.assertEqual(345, state2.get('n6', 'shared1'))521 def test_shared_backing_file_propagates_state_to_has(self):522 state1 = base_job.job_state()523 state1.set_backing_file('outfile7')524 state2 = base_job.job_state()525 state2.set_backing_file('outfile7')526 self.assertFalse(state1.has('n6', 'shared2'))527 self.assertFalse(state2.has('n6', 'shared2'))528 state1.set('n6', 'shared2', 'hello')529 self.assertTrue(state1.has('n6', 'shared2'))530 self.assertTrue(state2.has('n6', 'shared2'))531 def test_shared_backing_file_propagates_state_from_discard(self):532 state1 = base_job.job_state()533 state1.set_backing_file('outfile8')534 state1.set('n6', 'shared3', 10000)535 state2 = base_job.job_state()536 state2.set_backing_file('outfile8')537 self.assertEqual(10000, state1.get('n6', 'shared3'))538 self.assertEqual(10000, state2.get('n6', 'shared3'))539 state1.discard('n6', 'shared3')540 self.assertRaises(KeyError, state1.get, 'n6', 'shared3')541 self.assertRaises(KeyError, state2.get, 'n6', 'shared3')542 def test_shared_backing_file_propagates_state_from_discard_namespace(self):543 state1 = base_job.job_state()544 state1.set_backing_file('outfile9')545 state1.set('n7', 'shared4', -1)546 state1.set('n7', 'shared5', -2)547 state2 = base_job.job_state()548 state2.set_backing_file('outfile9')549 self.assertEqual(-1, state1.get('n7', 'shared4'))550 self.assertEqual(-1, state2.get('n7', 'shared4'))551 self.assertEqual(-2, state1.get('n7', 'shared5'))552 self.assertEqual(-2, state2.get('n7', 'shared5'))553 state1.discard_namespace('n7')554 self.assertRaises(KeyError, state1.get, 'n7', 'shared4')555 self.assertRaises(KeyError, state2.get, 'n7', 'shared4')556 self.assertRaises(KeyError, state1.get, 'n7', 'shared5')557 self.assertRaises(KeyError, state2.get, 'n7', 'shared5')558class test_job_state_backing_file_locking(unittest.TestCase):559 def setUp(self):560 self.testdir = tempfile.mkdtemp(suffix='unittest')561 self.original_wd = os.getcwd()562 os.chdir(self.testdir)563 # create a job_state object with stub read_* and write_* methods564 # to check that a lock is always held during a call to them565 ut_self = self566 class mocked_job_state(base_job.job_state):567 def read_from_file(self, file_path, merge=True):568 if self._backing_file and file_path == self._backing_file:569 ut_self.assertNotEqual(None, self._backing_file_lock)570 return super(mocked_job_state, self).read_from_file(571 file_path, merge=True)572 def write_to_file(self, file_path):573 if self._backing_file and file_path == self._backing_file:574 ut_self.assertNotEqual(None, self._backing_file_lock)575 return super(mocked_job_state, self).write_to_file(file_path)576 self.state = mocked_job_state()577 self.state.set_backing_file('backing_file')578 def tearDown(self):579 os.chdir(self.original_wd)580 shutil.rmtree(self.testdir, ignore_errors=True)581 def test_set(self):582 self.state.set('ns1', 'var1', 100)583 def test_get_missing(self):584 self.assertRaises(KeyError, self.state.get, 'ns2', 'var2')585 def test_get_present(self):586 self.state.set('ns3', 'var3', 333)587 self.assertEqual(333, self.state.get('ns3', 'var3'))588 def test_set_backing_file(self):589 self.state.set_backing_file('some_other_file')590 def test_has_missing(self):591 self.assertFalse(self.state.has('ns4', 'var4'))592 def test_has_present(self):593 self.state.set('ns5', 'var5', 55555)594 self.assertTrue(self.state.has('ns5', 'var5'))595 def test_discard_missing(self):596 self.state.discard('ns6', 'var6')597 def test_discard_present(self):598 self.state.set('ns7', 'var7', -777)599 self.state.discard('ns7', 'var7')600 def test_discard_missing_namespace(self):601 self.state.discard_namespace('ns8')602 def test_discard_present_namespace(self):603 self.state.set('ns8', 'var8', 80)604 self.state.set('ns8', 'var8.1', 81)605 self.state.discard_namespace('ns8')606 def test_disable_backing_file(self):607 self.state.set_backing_file(None)608 def test_change_backing_file(self):609 self.state.set_backing_file('another_backing_file')610 def test_read_from_a_non_backing_file(self):611 state = base_job.job_state()612 state.set('ns9', 'var9', 9999)613 state.write_to_file('non_backing_file')614 self.state.read_from_file('non_backing_file')615 def test_write_to_a_non_backing_file(self):616 self.state.write_to_file('non_backing_file')617class test_job_state_property_factory(unittest.TestCase):618 def setUp(self):619 class job_stub(object):620 pass621 self.job_class = job_stub622 self.job = job_stub()623 self.state = base_job.job_state()...
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!!