KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use build_manager method in prospector
Best Python code snippet using prospector_python
execution_tests.py
Source:execution_tests.py 
1import mock2import numbers3import unittest4import copy5import json6import builder.build7import builder.execution8from builder.tests.tests_jobs import *9from builder.build import BuildManager10from builder.execution import Executor, ExecutionManager, ExecutionResult, _submit_from_json11from builder.expanders import TimestampExpander12import builder.util as util13arrow = util.arrow_factory14class ExtendedMockExecutor(Executor):15 """"Executes" by running the jobs effect. An effect is a dictionary of16 things to do. Here is an example effect17 { }18 This effect updates non of the targets19 Here is another effect20 {21 "A-target": 50022 }23 This effect set's A's target's do_get_mtime to a mock thar returns 50024 """25 should_update_build_graph = True26 def do_execute(self, job):27 build_graph = self.get_build_graph()28 command = job.get_command()29 effect = job.get_effect()30 if isinstance(effect, numbers.Number):31 success = True32 else:33 success = effect.get('success') or True34 target_ids = build_graph.get_target_ids(job.get_id())35 for target_id in target_ids:36 target = build_graph.get_target(target_id)37 if isinstance(effect, numbers.Number):38 target.do_get_mtime = mock.Mock(return_value=effect)39 elif target_id not in effect:40 continue41 else:42 target.do_get_mtime = mock.Mock(return_value=effect[target_id])43 print target_id, target.do_get_mtime()44 result = ExecutionResult(False, success, command, command)45 return result46class ExecutionManagerTests1(unittest.TestCase):47 def _get_execution_manager(self, jobs, executor=None):48 build_manager = builder.build.BuildManager(jobs=jobs, metas=[])49 if executor is None:50 executor = mock.Mock(return_value=mock.Mock(status=True, stdout='', stderr=''))51 execution_manager = builder.execution.ExecutionManager(build_manager, executor)52 return execution_manager53 def _get_buildable_job(self):54 return EffectJobDefinition("buildable_job", expander_type=TimestampExpander,55 depends=[{"unexpanded_id": "buildable_15_minute_target_01-%Y-%m-%d-%H-%M", "file_step": "15min"},56 {"unexpanded_id": "buildable_5_minute_target_01-%Y-%m-%d-%H-%M", "file_step": "5min"},57 {"unexpanded_id": "buildable_15_minute_target_02-%Y-%m-%d-%H-%M", "file_step": "15min",58 "type": "depends_one_or_more"},59 {"unexpanded_id": "buildable_5_minute_target_02-%Y-%m-%d-%H-%M", "file_step": "5min",60 "type": "depends_one_or_more"}])61 def test_submit(self):62 # Given63 execution_manager = self._get_execution_manager([self._get_buildable_job()])64 build_context = {65 'start_time': arrow.get('2015-01-01')66 }67 # When68 execution_manager.running = True69 execution_manager.submit('buildable_job', build_context)70 # Then71 self.assertIn('buildable_job', execution_manager.get_build())72 def test_start_excution_run_to_completion(self):73 # Given74 execution_manager = self._get_execution_manager([self._get_buildable_job()], executor=ExtendedMockExecutor)75 execution_manager.executor.execute = mock.Mock(wraps=execution_manager.executor.execute)76 build_context = {77 'start_time': arrow.get('2015-01-01')78 }79 # When80 execution_manager.build.add_job('buildable_job', build_context, force=True)81 execution_manager.start_execution(inline=True)82 # Then83 self.assertTrue(execution_manager.executor.execute.called)84 def test_inline_execution_simple_plan(self):85 # Given86 jobs = [87 EffectJobDefinition('A', targets=['target-A']),88 EffectJobDefinition('B', depends=['target-A'], targets=['target-B1', 'target-B2'])89 ]90 execution_manager = self._get_execution_manager(jobs, executor=ExtendedMockExecutor)91 execution_manager.executor.execute = mock.Mock(wraps=execution_manager.executor.execute)92 build_context = {93 'start_time': arrow.get('2015-01-01')94 }95 # When96 execution_manager.build.add_job('B', build_context)97 execution_manager.start_execution(inline=True)98 # Then99 self.assertEquals(execution_manager.executor.execute.call_count, 2)100 def test_inline_execution_retries(self):101 # Given102 jobs = [103 EffectJobDefinition('A', targets=['target-A'], effect={"target-A": None}),104 ]105 execution_manager = self._get_execution_manager(jobs, executor=ExtendedMockExecutor)106 execution_manager.executor.execute = mock.Mock(wraps=execution_manager.executor.execute)107 build_context = {108 'start_time': arrow.get('2015-01-01')109 }110 # When111 execution_manager.build.add_job('A', build_context)112 execution_manager.start_execution(inline=True)113 # Then114 self.assertEquals(execution_manager.executor.execute.call_count, 5)115class ExecutionManagerTests2(unittest.TestCase):116 def _get_execution_manager(self, jobs):117 build_manager = BuildManager(jobs, metas=[])118 execution_manager = ExecutionManager(build_manager, lambda execution_manager, config=None: ExtendedMockExecutor(execution_manager, config=config))119 return execution_manager120 def _get_execution_manager_with_effects(self, jobs):121 build_manager = BuildManager(jobs, metas=[])122 execution_manager = ExecutionManager(build_manager, ExtendedMockExecutor)123 return execution_manager124 def test_get_starting_jobs(self):125 # given126 jobs = [SimpleTestJobDefinition('get_starting_jobs_01'),127 SimpleTestJobDefinition('get_starting_jobs_02'),128 SimpleTestJobDefinition('get_starting_jobs_03'),129 SimpleTestJobDefinition('get_starting_jobs_04')]130 execution_manager = self._get_execution_manager(jobs)131 build1 = execution_manager.get_build()132 build_context = {}133 for job in jobs:134 build1.add_job(job.unexpanded_id, copy.deepcopy(build_context))135 expected_starting_job_ids = [136 "get_starting_jobs_01",137 "get_starting_jobs_03"138 ]139 (build1.node140 ["get_starting_jobs_01"]141 ["object"].should_run) = True142 (build1.node143 ["get_starting_jobs_01"]144 ["object"].parents_should_run) = False145 (build1.node146 ["get_starting_jobs_02"]147 ["object"].should_run) = True148 (build1.node149 ["get_starting_jobs_02"]150 ["object"].parents_should_run) = True151 (build1.node152 ["get_starting_jobs_03"]153 ["object"].should_run) = True154 (build1.node155 ["get_starting_jobs_03"]156 ["object"].parents_should_run) = False157 (build1.node158 ["get_starting_jobs_04"]159 ["object"].should_run) = False160 (build1.node161 ["get_starting_jobs_04"]162 ["object"].parents_should_run) = False163 # when164 starting_job_ids = execution_manager.get_jobs_to_run()165 # then166 self.assertItemsEqual(starting_job_ids, expected_starting_job_ids)167 def test_no_depends_next_jobs(self):168 """tests_no_depends_next_jobs169 tests a situation where nothing depends on the job. When the job170 finishes, nothing should be returned as the next job to run171 """172 # Given173 jobs = [EffectJobDefinition("A",174 depends=None,175 targets=["A-target"])]176 execution_manager = self._get_execution_manager(jobs)177 # When178 execution_manager.running = True179 execution_manager.submit("A", {})180 execution_manager.execute("A")181 # Then182 self.assertEquals(set([]), execution_manager.get_next_jobs_to_run("A"))183 def test_simple_get_next_jobs(self):184 """test_simple_get_next_jobs185 test a situation where a job depends on a target of another job. When186 the depended on job finishes, the other job should be the next job to187 run188 """189 # Given190 jobs = [191 EffectJobDefinition("A",192 depends=None,targets=["A-target"]),193 EffectJobDefinition("B",194 depends=['A-target'], targets=["B-target"])195 ]196 execution_manager = self._get_execution_manager(jobs)197 # When198 execution_manager.running = True199 execution_manager.submit("B", {})200 execution_manager.execute("A")201 # Then202 self.assertEquals(set(["B"]), execution_manager.get_next_jobs_to_run("A"))203 def test_simple_get_next_jobs_lower(self):204 # Given205 jobs = [206 EffectJobDefinition("A", depends=None, targets=["A-target"]),207 EffectJobDefinition(208 "B", depends=["A-target"], targets=[209 {210 "unexpanded_id": "B-target",211 "start_mtime": 200212 }213 ]214 ),215 EffectJobDefinition(216 "C", depends=[217 {218 "unexpanded_id": "B-target",219 "start_mtime": 200220 }221 ], targets=["C-target"]222 )223 ]224 execution_manager = self._get_execution_manager(jobs)225 # When226 execution_manager.running = True227 execution_manager.submit("C", {})228 execution_manager.execute("A")229 # Then230 for node_id, node in execution_manager.build.node.iteritems():231 if execution_manager.build.is_target_object(node["object"]):232 print node_id, node["object"].get_mtime()233 self.assertEquals(set(["C"]), execution_manager.get_next_jobs_to_run("A"))234 def test_simple_get_next_jobs_failed_but_creates_targets(self):235 """test_simple_get_next_jobs_failed236 test a situation where a job depends on a target of another job. When237 the depended on job finishes, but fails, does not reach it's max238 fail count, and creates targets, the dependent should be next job to run239 """240 # Given241 jobs = [242 EffectJobDefinition("A",243 depends=None,targets=["A-target"], effect={"A-target": 1, "success": False}),244 EffectJobDefinition("B",245 depends=['A-target'], targets=["B-target"])246 ]247 execution_manager = self._get_execution_manager_with_effects(jobs)248 # When249 execution_manager.running = True250 execution_manager.submit("B", {})251 execution_manager.execute("A")252 # The253 self.assertEquals(set(["B"]), execution_manager.get_next_jobs_to_run("A"))254 def test_simple_get_next_jobs_failed_but_no_targets(self):255 """test_simple_get_next_jobs_failed256 test a situation where a job depends on a target of another job. When257 the depended on job finishes, but fails and does not reach it's max258 fail count, and does not create targets, the job should run again259 """260 # Given261 jobs = [262 EffectJobDefinition("A",263 depends=None,targets=["A-target"]),264 EffectJobDefinition("B",265 depends=['A-target'], targets=["B-target"])266 ]267 execution_manager = self._get_execution_manager(jobs)268 execution_manager.executor.execute = mock.Mock(return_value=mock.Mock(status=True, stdout='', stderr=''))269 # When270 execution_manager.running = True271 execution_manager.submit("B", {})272 execution_manager.execute("A")273 # The274 self.assertEquals(set(["A"]), execution_manager.get_next_jobs_to_run("A"))275 def test_simple_get_next_jobs_failed_max(self):276 """test_simple_get_next_jobs_failed_max277 test a situation where a job depends on a target of another job.278 When the depended on job finishes, but fails and reaches it's max fail279 count, return nothing as the next job to run.280 """281 # Given282 jobs = [283 EffectJobDefinition("A",284 depends=None,targets=["A-target"], effect={"A-target": None}),285 EffectJobDefinition("B",286 depends=['A-target'], targets=["B-target"])287 ]288 execution_manager = self._get_execution_manager(jobs)289 # When290 execution_manager.running = True291 execution_manager.submit("B", {})292 for i in xrange(6):293 execution_manager.execute("A")294 # The295 self.assertEquals(set([]), execution_manager.get_next_jobs_to_run("A"))296 def test_multiple_get_next_jobs(self):297 """test_multiple_get_next_jobs298 test a situation where a job creates multiple targets where individual299 jobs depend on individual targets. When the depended on job finishes, all300 of the lower jobs should be the next job to run.301 """302 # Given303 jobs = [304 EffectJobDefinition("A",305 depends=None,targets=["A-target"]),306 EffectJobDefinition("B",307 depends=['A-target'], targets=["B-target"]),308 EffectJobDefinition("C",309 depends=['A-target'], targets=["C-target"]),310 EffectJobDefinition("D",311 depends=['A-target'], targets=["D-target"]),312 ]313 execution_manager = self._get_execution_manager(jobs)314 # When315 execution_manager.running = True316 execution_manager.submit("A", {}, direction={"up", "down"})317 execution_manager.execute("A")318 # Then319 self.assertEquals({"B", "C", "D"}, set(execution_manager.get_next_jobs_to_run("A")))320 def test_multiple_get_next_jobs_failed(self):321 """test_multiple_get_next_jobs_failed322 test a situation where a job creates multiple targets where individual323 jobs depend on individual targets. When the depended on job finishes,324 but fails and does not reach it's max fail count, either the failed job325 should be the next job to run or nothing should be the next job to run.326 When the depended on job finishes it should make some of it's targets.327 This tests to make sure that when the job fails, the job's that now have328 their dependencies don't run. This is not covered by should run as there329 is a possibility that the lower nodes are check for should run before330 the parent job is invalidated.331 """332 # Given333 jobs = [334 SimpleTestJobDefinition("A",335 depends=None,targets=["A1-target", "A2-target", "A3-target"]),336 SimpleTestJobDefinition("B",337 depends=['A1-target'], targets=["B-target"]),338 SimpleTestJobDefinition("C",339 depends=['A2-target'], targets=["C-target"]),340 SimpleTestJobDefinition("D",341 depends=['A3-target'], targets=["D-target"]),342 ]343 execution_manager = self._get_execution_manager(jobs)344 execution_manager.executor.execute = mock.Mock(return_value=mock.Mock(status=True, stdout='', stderr=''))345 execution_manager.running = True346 execution_manager.submit("A", {}, direction={"up", "down"})347 execution_manager.build.get_target('A1-target').do_get_mtime = mock.Mock(return_value=None)348 # When349 execution_manager.execute("A")350 # Then351 self.assertEquals({"A"}, set(execution_manager.get_next_jobs_to_run("A")))352 def test_multiple_get_next_jobs_failed_max(self):353 """test_multiple_get_next_jobs_failed_max354 test a situation where a job creates multiple targets where individual355 jobs depend on individual targets. When the depended on job finishes,356 but fails, reaches it's max fail count, and some targets are created,357 all the jobs below with dependencies that exist should be the next jobs358 to run.359 """360 # Given361 jobs = [362 EffectJobDefinition("A",363 depends=None,targets=["A1-target", "A2-target", "A3-target"], effect={"A1-target": None, "A2-target":1, "A3-target":1}),364 EffectJobDefinition("B",365 depends=['A1-target'], targets=["B-target"]),366 EffectJobDefinition("C",367 depends=['A2-target'], targets=["C-target"]),368 EffectJobDefinition("D",369 depends=['A3-target'], targets=["D-target"]),370 ]371 execution_manager = self._get_execution_manager(jobs)372 execution_manager.running = True373 execution_manager.submit("A", {}, direction={"up", "down"})374 # When375 for i in xrange(6):376 execution_manager.execute("A")377 execution_manager.build.get_target('A1-target').do_get_mtime = mock.Mock(return_value=None)378 # Then379 self.assertEquals({"C", "D"}, set(execution_manager.get_next_jobs_to_run("A")))380 self.assertEquals(execution_manager.get_build().get_job("C").get_should_run(), True)381 self.assertEquals(execution_manager.get_build().get_job("D").get_should_run(), True)382 def test_depends_one_or_more_next_jobs(self):383 """test_depends_one_or_more_next_jobs384 test a situation where a job has a depends one or more dependency. It is385 not past it's curfew so it needs all of the dependencies to run.386 Complete each of it's dependencies individually. Each one should return387 nothing until the last one.388 """389 # Given390 jobs = [391 EffectJobDefinition("A1",392 depends=None, targets=["A1-target"],393 effect={"A1-target": 1}),394 EffectJobDefinition("A2",395 depends=None, targets=["A2-target"],396 effect=[{"A2-target": None}, {"A2-target": 1}]),397 EffectJobDefinition("B",398 depends=[399 {"unexpanded_id": "A1-target", "type": "depends_one_or_more"},400 {"unexpanded_id": "A2-target", "type": "depends_one_or_more"}],401 targets=["B-target"])402 ]403 execution_manager = self._get_execution_manager_with_effects(jobs)404 build_context = {"start_time": arrow.get("2015-01-01-00-00"), "end_time": arrow.get("2015-01-01-00-10")}405 execution_manager.running = True406 execution_manager.submit("B", build_context)407 # When408 execution_manager.execute("A1")409 execution_manager.execute("A2")410 # Then411 self.assertEquals(set(), set(execution_manager.get_next_jobs_to_run("A1")))412 self.assertEquals({"A2"}, set(execution_manager.get_next_jobs_to_run("A2")))413 self.assertEquals(execution_manager.get_build().get_job("B").get_should_run(), False)414 # On rerun, A2 complete successfully and therefore B should run415 execution_manager.execute("A2")416 self.assertEquals({"B"}, set(execution_manager.get_next_jobs_to_run("A1")))417 self.assertEquals({"B"}, set(execution_manager.get_next_jobs_to_run("A2")))418 self.assertEquals(execution_manager.get_build().get_job("B").get_should_run(), True)419 def test_depends_one_or_more_next_jobs_failed_max_lower(self):420 """test_depends_one_or_more_next_jobs_failed421 test a situation where a job has a depends one or more dependency. It422 is not past it's curfew so it needs all of the dependencies to run.423 Each of the dependencies should also depend on a single job so there are424 a total of three layers of jobs. Complete each of the jobs in the first425 two rows except the last job. The last job in the first row should fail426 and reach it's max fail count. It's next job should be the job in the427 bottom row as all of it's buildable dependencies are built and all of428 the non buildable dependencies are due to a failure.429 """430 jobs = [431 EffectTimestampExpandedJobDefinition("A", file_step="5min",432 depends=None,433 targets=[{"unexpanded_id": "A-target-%Y-%m-%d-%H-%M", "file_step": "5min"}]),434 EffectTimestampExpandedJobDefinition("B", file_step="5min",435 depends=None,436 targets=[{"unexpanded_id": "B-target-%Y-%m-%d-%H-%M", "file_step": "5min"}],437 effect=[{"B-target-2015-01-01-00-00": 1, "B-target-2015-01-01-00-05": None, "success": False}]),438 EffectJobDefinition("C", expander_type=TimestampExpander,439 depends=[440 {"unexpanded_id": "A-target-%Y-%m-%d-%H-%M", "file_step": "5min", "type": "depends_one_or_more"},441 {"unexpanded_id": "B-target-%Y-%m-%d-%H-%M", "file_step": "5min", "type": "depends_one_or_more"}],442 targets=[{"unexpanded_id": "C-target-%Y-%m-%d-%H-%M", "file_step": "5min"}])443 ]444 execution_manager = self._get_execution_manager_with_effects(jobs)445 build_context = {"start_time": arrow.get("2015-01-01-00-00"), "end_time": arrow.get("2015-01-01-00-10")}446 execution_manager.running = True447 execution_manager.submit("C", build_context)448 # When449 executions = ["A_2015-01-01-00-05-00", "A_2015-01-01-00-00-00", "B_2015-01-01-00-00-00"] + ["B_2015-01-01-00-05-00"]*6450 for execution in executions:451 execution_manager.execute(execution)452 # Then453 self.assertEquals(execution_manager.get_build().get_job("B_2015-01-01-00-05-00").get_should_run(), False)454 self.assertEquals(execution_manager.get_build().get_job("B_2015-01-01-00-05-00").get_stale(), True)455 for job_id in ("A_2015-01-01-00-05-00", "A_2015-01-01-00-00-00", "B_2015-01-01-00-00-00", "B_2015-01-01-00-05-00"):456 self.assertEquals({"C"}, set(execution_manager.get_next_jobs_to_run(job_id)))457 self.assertEquals(execution_manager.get_build().get_job("C").get_should_run(), True)458 def test_upper_update(self):459 """tests situations where a job starts running and while it is running a460 job above it should run again, possibly due to a target being deleted461 or a force. When the running job finishes, none of it's lower jobs462 should run.463 """464 # Given465 jobs = [466 EffectJobDefinition("A",467 depends=None, targets=["A-target"],468 effect=[1,100]),469 EffectJobDefinition("B",470 depends=["A-target"], targets=["B-target"], effect=1),471 EffectJobDefinition("C",472 depends=["B-target"], targets=["C-target"], effect=1),473 ]474 execution_manager = self._get_execution_manager_with_effects(jobs)475 build_context = {}476 execution_manager.running = True477 execution_manager.submit("C", build_context)478 # When479 execution_manager.execute("A")480 execution_manager.execute("B")481 execution_manager.submit("A", {}, force=True)482 for node_id, node in execution_manager.build.node.iteritems():483 if execution_manager.build.is_target(node_id):484 print node_id, node["object"].get_mtime()485 # Then486 self.assertEquals(set(), set(execution_manager.get_next_jobs_to_run("B")))487 def test_multiple_targets_one_exists(self):488 # Given489 jobs = [490 EffectJobDefinition("A",491 depends=None, targets=["A1-target", "A2-target"],492 effect=[{"A1-target": 1, "A2-target": None}, {"A1-target": 1, "A2-target": None}, {"A1-target": 1, "A2-target": 4}]),493 EffectJobDefinition("B",494 depends=["A1-target"], targets=["B-target"], effect=2),495 EffectJobDefinition("C",496 depends=["A2-target"], targets=["C-target"], effect=5),497 EffectJobDefinition("D",498 depends=["B-target"], targets=["D-target"], effect=3),499 EffectJobDefinition("E",500 depends=["C-target"], targets=["E-target"], effect=6),501 ]502 execution_manager = self._get_execution_manager_with_effects(jobs)503 build_context = {}504 execution_manager.build.add_job("A", build_context, direction={"down", "up"})505 # When506 for execution in ("A", "B", "C", "E", "A", "A"):507 job = execution_manager.build.get_job(execution)508 if job.get_should_run_immediate():509 print execution510 execution_manager.execute(execution)511 # Then512 self.assertEquals({"C", "D"}, set(execution_manager.get_next_jobs_to_run("A")))513 self.assertEquals(execution_manager.get_build().get_job("C").get_should_run(), True)514 execution_manager.execute("C")515 self.assertEquals({"E"}, set(execution_manager.get_next_jobs_to_run("C")))516 self.assertEquals(execution_manager.get_build().get_job("E").get_should_run(), True)517 self.assertEquals(execution_manager.get_build().get_job("D").get_should_run(), True)518 def test_effect_job(self):519 # Given520 jobs = [521 EffectJobDefinition("A", targets=["A-target"]),522 EffectJobDefinition("B", depends=["A-target"], targets=["B-target"]),523 ]524 execution_manager = self._get_execution_manager_with_effects(jobs)525 # When526 execution_manager.running = True527 execution_manager.submit("B", {})528 execution_manager.start_execution(inline=True)529 # Then530 job_A = execution_manager.build.get_job("A")531 job_B = execution_manager.build.get_job("B")532 self.assertEqual(job_A.count, 1)533 self.assertEqual(job_B.count, 1)534class ExecutionDaemonTests(unittest.TestCase):535 def _get_execution_manager_with_effects(self):536 build_manager = BuildManager([EffectTimestampExpandedJobDefinition("A", file_step="5min",537 targets=[{"unexpanded_id": "A-target", "file_step": "5min"}])], metas=[])538 execution_manager = ExecutionManager(build_manager, ExtendedMockExecutor)539 return execution_manager540 def test_submission(self):541 # Given542 execution_manager = self._get_execution_manager_with_effects()543 json_body = json.dumps({544 "job_definition_id": "A",545 "build_context": {546 "start_time": "2015-04-01", "end_time": "2015-04-01"547 }548 })549 # When550 execution_manager.running = True551 _submit_from_json(execution_manager, json_body)552 # Then553 self.assertTrue(execution_manager.get_build().is_job("A_2015-04-01-00-00-00"))554 def test_update_lower_nodes(self):555 """test_update_lower_nodes556 Add a node that should run that is above the nodes in the graph.557 All the other nodes originally didn't have parent's that should run.558 Now they do.559 """560 # Given561 jobs = [562 SimpleTestJobDefinition('job1', targets=['target1']),563 SimpleTestJobDefinition('job2', targets=['target2'],564 depends=['target1']),565 SimpleTestJobDefinition('job3', targets=['target3'],566 depends=['target2']),567 ]568 build_manager = BuildManager(jobs, [])569 execution_manager = ExecutionManager(build_manager,570 ExtendedMockExecutor)571 execution_manager.build.add_job('job1', {}, depth=1)572 execution_manager.build.add_job('job2', {}, depth=1)573 execution_manager.build.add_job('job3', {}, depth=1)574 build_graph = execution_manager.build575 job1 = build_graph.get_job('job1')576 job2 = build_graph.get_job('job2')577 job3 = build_graph.get_job('job3')578 job1.get_should_run_immediate = mock.Mock(return_value=True)579 job2.parents_should_run = False580 job3.parents_should_run = False581 # When582 execution_manager.update_parents_should_run('job1')583 # Then584 self.assertFalse(job1.parents_should_run)585 self.assertIsNone(job2.parents_should_run)586 self.assertIsNone(job3.parents_should_run)587 def test_update_lower_nodes_connection(self):588 """test_update_lower_nodes_connection589 Add a node that shouldn't run but has parent's that should run.590 This node connects the parent nodes to lower nodes.591 All the lower nodes originally didn't have parent's that should run.592 Now they do.593 """594 # Given595 jobs = [596 SimpleTestJobDefinition('job1', targets=['target1']),597 SimpleTestJobDefinition('job2', targets=['target2'],598 depends=['target1']),599 SimpleTestJobDefinition('job3', targets=['target3'],600 depends=['target2']),601 ]602 build_manager = BuildManager(jobs, [])603 execution_manager = ExecutionManager(build_manager,604 ExtendedMockExecutor)605 execution_manager.build.add_job('job1', {}, depth=1)606 execution_manager.build.add_job('job2', {}, depth=1)607 execution_manager.build.add_job('job3', {}, depth=1)608 build_graph = execution_manager.build609 job1 = build_graph.get_job('job1')610 job2 = build_graph.get_job('job2')611 job3 = build_graph.get_job('job3')612 job1.get_should_run_immediate = mock.Mock(return_value=True)613 job2.get_should_run_immediate = mock.Mock(return_value=False)614 job3.parents_should_run = False615 # When616 execution_manager.update_parents_should_run('job2')617 # Then618 self.assertTrue(job2.parents_should_run)619 self.assertIsNone(job3.parents_should_run)620 def test_update_lower_nodes_cached(self):621 """test_update_lower_nodes_cached622 Add a node that should run. Update all lower nodes until you get to a623 node that already had parent's that should have ran.624 """625 # Given626 jobs = [627 SimpleTestJobDefinition('job1', targets=['target1']),628 SimpleTestJobDefinition('job2', targets=['target2'],629 depends=['target1']),630 SimpleTestJobDefinition('job3', targets=['target3'],631 depends=['target2']),632 SimpleTestJobDefinition('job4', targets=['target4'],633 depends=['target3']),634 ]635 build_manager = BuildManager(jobs, [])636 execution_manager = ExecutionManager(build_manager,637 ExtendedMockExecutor)638 execution_manager.build.add_job('job1', {}, depth=1)639 execution_manager.build.add_job('job2', {}, depth=1)640 execution_manager.build.add_job('job3', {}, depth=1)641 execution_manager.build.add_job('job4', {}, depth=1)642 build_graph = execution_manager.build643 job1 = build_graph.get_job('job1')644 job2 = build_graph.get_job('job2')645 job3 = build_graph.get_job('job3')646 job4 = build_graph.get_job('job4')647 job1.get_should_run_immediate = mock.Mock(return_value=True)648 job2.parents_should_run = False649 job3.parents_should_run = True650 job4.parents_should_run = False # can't really happen651 # just being used to check stopage652 # When653 execution_manager.update_parents_should_run('job1')654 # Then655 self.assertIsNone(job2.parents_should_run)656 self.assertTrue(job3.parents_should_run)657 self.assertFalse(job4.parents_should_run)658 def test_update_lower_exit_early(self):659 """test_update_lower_exit_early660 Add a node that should not run and it's parent's should not run.661 Don't invalidate the lower nodes662 """663 # Given664 jobs = [665 SimpleTestJobDefinition('job1', targets=['target1']),666 SimpleTestJobDefinition('job2', targets=['target2'],667 depends=['target1']),668 ]669 build_manager = BuildManager(jobs, [])670 execution_manager = ExecutionManager(build_manager,671 ExtendedMockExecutor)672 execution_manager.build.add_job('job1', {}, depth=1)673 execution_manager.build.add_job('job2', {}, depth=1)674 build_graph = execution_manager.build675 job1 = build_graph.get_job('job1')676 job2 = build_graph.get_job('job2')677 job1.get_should_run_immediate = mock.Mock(return_value=False)678 job2.parents_should_run = False679 # When680 execution_manager.update_parents_should_run('job1')681 # Then682 self.assertFalse(job2.parents_should_run)683 def test_update_lower_nodes_ignore_parents(self):684 """test_update_lower_nodes_ignore_parents685 Add a node that should run. Update all lower nodes until you get to a686 node that has a node that ignores it's parents687 """688 # Given689 jobs = [690 SimpleTestJobDefinition('job1', targets=['target1']),691 SimpleTestJobDefinition('job2', targets=['target2'],692 depends=['target1']),693 SimpleTestJobDefinition('job3', targets=['target3'],694 depends=['target2']),695 SimpleTestJobDefinition('job4', targets=['target4'],696 depends=['target3']),697 ]698 build_manager = BuildManager(jobs, [])699 execution_manager = ExecutionManager(build_manager,700 ExtendedMockExecutor)701 execution_manager.build.add_job('job1', {}, depth=1)702 execution_manager.build.add_job('job2', {}, depth=1)703 execution_manager.build.add_job('job3', {}, depth=1)704 execution_manager.build.add_job('job4', {}, depth=1)705 build_graph = execution_manager.build706 job1 = build_graph.get_job('job1')707 job2 = build_graph.get_job('job2')708 job3 = build_graph.get_job('job3')709 job4 = build_graph.get_job('job4')710 job1.get_should_run_immediate = mock.Mock(return_value=True)711 job2.parents_should_run = False712 job3.ignore_parents = mock.Mock(return_value=True)713 job3.parents_should_run = False714 job4.parents_should_run = False # can't really happen715 # just being used to check stopage716 # When717 execution_manager.update_parents_should_run('job1')718 # Then719 self.assertIsNone(job2.parents_should_run)720 self.assertFalse(job3.parents_should_run)721 self.assertFalse(job4.parents_should_run)722 def test_double_update_lower_nodes(self):723 """test_double_update_lower_nodes724 Update two nodes, make sure that all the things get iterated725 through726 """727 # Given728 jobs = [729 SimpleTestJobDefinition("job1", targets=["target1"]),730 SimpleTestJobDefinition("job2", targets=["target2"],731 depends=["target1"]),732 SimpleTestJobDefinition("job3", targets=["target3"],733 depends=["target2"]),734 SimpleTestJobDefinition("job1'", targets=["target1'"]),735 SimpleTestJobDefinition("job2'", targets=["target2'"],736 depends=["target1'"]),737 SimpleTestJobDefinition("job3'", targets=["target3'"],738 depends=["target2'"]),739 ]740 build_manager = BuildManager(jobs, [])741 execution_manager = ExecutionManager(build_manager,742 ExtendedMockExecutor)743 execution_manager.build.add_job("job1", {}, depth=1)744 execution_manager.build.add_job("job2", {}, depth=1)745 execution_manager.build.add_job("job3", {}, depth=1)746 execution_manager.build.add_job("job1'", {}, depth=1)747 execution_manager.build.add_job("job2'", {}, depth=1)748 execution_manager.build.add_job("job3'", {}, depth=1)749 build_graph = execution_manager.build750 job1 = build_graph.get_job("job1")751 job2 = build_graph.get_job("job2")752 job3 = build_graph.get_job("job3")753 job1_ = build_graph.get_job("job1'")754 job2_ = build_graph.get_job("job2'")755 job3_ = build_graph.get_job("job3'")756 job1.get_should_run_immediate = mock.Mock(return_value=True)757 job2.parents_should_run = False758 job3.parents_should_run = False759 job1_.get_should_run_immediate = mock.Mock(return_value=True)760 job2_.parents_should_run = False761 job3_.parents_should_run = False762 # When763 for job_id in ["job1", "job1'"]:764 execution_manager.update_parents_should_run(job_id)765 # Then766 self.assertFalse(job1.parents_should_run)767 self.assertIsNone(job2.parents_should_run)768 self.assertIsNone(job3.parents_should_run)769 self.assertFalse(job1_.parents_should_run)770 self.assertIsNone(job2_.parents_should_run)771 self.assertIsNone(job3_.parents_should_run)772 def test_update_lower_first_ignores_parents(self):773 """test_update_lower_first_ignores_parents774 A test where the first job ignores it's parents. Nothing should be775 updated as everything ignores things that ignores it's parents776 """777 # Given778 jobs = [779 SimpleTestJobDefinition('job1', targets=['target1']),780 SimpleTestJobDefinition('job2', targets=['target2'],781 depends=['target1']),782 ]783 build_manager = BuildManager(jobs, [])784 execution_manager = ExecutionManager(build_manager,785 ExtendedMockExecutor)786 execution_manager.build.add_job('job1', {}, depth=1)787 execution_manager.build.add_job('job2', {}, depth=1)788 build_graph = execution_manager.build789 job1 = build_graph.get_job('job1')790 job2 = build_graph.get_job('job2')791 job1.get_should_run_immediate = mock.Mock(return_value=True)792 job2.parents_should_run = False793 # When794 execution_manager.update_parents_should_run('job1')795 # Then796 self.assertFalse(job2.parents_should_run)797 def test_update_lower_same_twice(self):798 """test_update_lower_same_twice799 A test that has two jobs updated that have the same dependant job800 The first job should invalidate the job, the second should find that801 it's parents should run and stop there802 """803 # Given804 jobs = [805 SimpleTestJobDefinition('job1', targets=['target1']),806 SimpleTestJobDefinition('job2', targets=['target2']),807 SimpleTestJobDefinition('job3', targets=['target3'],808 depends=[809 'target1',810 'target2',811 ]),812 SimpleTestJobDefinition('job4', targets=['target4'],813 depends=['target3']),814 ]815 build_manager = BuildManager(jobs, [])816 execution_manager = ExecutionManager(build_manager,817 ExtendedMockExecutor)818 execution_manager.build.add_job('job1', {}, depth=1)819 execution_manager.build.add_job('job2', {}, depth=1)820 execution_manager.build.add_job('job3', {}, depth=1)821 execution_manager.build.add_job('job4', {}, depth=1)822 build_graph = execution_manager.build823 job1 = build_graph.get_job('job1')824 job2 = build_graph.get_job('job2')825 job3 = build_graph.get_job('job3')826 job4 = build_graph.get_job('job4')827 job1.get_should_run_immediate = mock.Mock(return_value=True)828 job2.get_should_run_immediate = mock.Mock(return_value=True)829 job3.parents_should_run = False830 job4.parents_should_run = False831 # When832 for job_id in ["job1", "job2"]:833 execution_manager.update_parents_should_run(job_id)834 # Then835 self.assertTrue(job3.parents_should_run)836 self.assertIsNone(job4.parents_should_run)837 def test_addition_updates(self):838 """test_addition_updates839 This tests to make sure that when a job is added to the graph that the840 execution system updates the graph to be consisitent841 """842 jobs = [843 SimpleTestJobDefinition(844 'job1', targets=[845 {846 'unexpanded_id': 'target1',847 'start_mtime': 100,848 }849 ], depends=[850 {851 'unexpanded_id': 'super_target1',852 'start_mtime': 200,853 }854 ]855 ),856 SimpleTestJobDefinition('job2', targets=['target2'],857 depends=['target1']),858 SimpleTestJobDefinition('job3', targets=['target3'],859 depends=['target2']),860 ]861 build_manager = BuildManager(jobs, [])862 execution_manager = ExecutionManager(build_manager,863 ExtendedMockExecutor)864 execution_manager.build.add_job('job2', {}, depth=1)865 execution_manager.build.add_job('job3', {}, depth=1)866 build_graph = execution_manager.build867 job2 = build_graph.get_job('job2')868 job3 = build_graph.get_job('job3')869 job2.parents_should_run = False870 job3.parents_should_run = False871 # When872 execution_manager.running = True873 execution_manager.submit('job1', {})874 job1 = build_graph.get_job('job1')875 # Then876 self.assertFalse(job1.parents_should_run)877 self.assertIsNone(job2.parents_should_run)878 self.assertIsNone(job3.parents_should_run)879 def test_update_target_no_creator_should_not_run(self):880 # Given881 jobs = [882 SimpleTestJobDefinition('job1', targets=['target1'],883 depends=['super_target1']),884 SimpleTestJobDefinition('job2', targets=['target2'],885 depends=['target1'])886 ]887 build_manager = BuildManager(jobs, [])888 execution_manager = ExecutionManager(build_manager,889 ExtendedMockExecutor)890 build_graph = execution_manager.build891 build_graph.add_job("job2", {})892 job1 = build_graph.get_job("job1")893 job2 = build_graph.get_job("job2")894 job1.should_run_immediate = True895 job2.should_run_immediate = True896 job1.parents_should_run = False897 job2.parents_should_run = True898 super_target1 = build_graph.get_target("super_target1")899 target1 = build_graph.get_target("target1")900 target2 = build_graph.get_target("target2")901 super_target1.do_get_mtime = mock.Mock(return_value=100)902 target1.do_get_mtime = mock.Mock(return_value=200)903 target2.do_get_mtime = mock.Mock(return_value=None)904 # When905 execution_manager.external_update_targets(['super_target1'])906 # Then907 self.assertFalse(execution_manager._work_queue.empty())908 self.assertEqual(execution_manager._work_queue.get(False), 'job2')909 self.assertTrue(execution_manager._work_queue.empty())910 def test_update_target_no_creator_should_run(self):911 # Given912 jobs = [913 SimpleTestJobDefinition('job1', targets=['target1'],914 depends=['super_target1']),915 SimpleTestJobDefinition('job2', targets=['target2'],916 depends=['target1'])917 ]918 build_manager = BuildManager(jobs, [])919 execution_manager = ExecutionManager(build_manager,920 ExtendedMockExecutor)921 build_graph = execution_manager.build922 build_graph.add_job("job2", {})923 job1 = build_graph.get_job("job1")924 job2 = build_graph.get_job("job2")925 job1.should_run_immediate = True926 job2.should_run_immediate = True927 job1.parents_should_run = False928 job2.parents_should_run = True929 super_target1 = build_graph.get_target("super_target1")930 target1 = build_graph.get_target("target1")931 target2 = build_graph.get_target("target2")932 super_target1.do_get_mtime = mock.Mock(return_value=100)933 target1.do_get_mtime = mock.Mock(return_value=50)934 target2.do_get_mtime = mock.Mock(return_value=None)935 # When936 execution_manager.external_update_targets(['super_target1'])937 # Then938 self.assertFalse(execution_manager._work_queue.empty())939 self.assertEqual(execution_manager._work_queue.get(False), 'job1')940 self.assertTrue(execution_manager._work_queue.empty())941 def test_update_target_no_creator_should_not_run_deep(self):942 # Given943 jobs = [944 SimpleTestJobDefinition('job1', targets=['target1'],945 depends=['super_target1']),946 SimpleTestJobDefinition('job2', targets=['target2'],947 depends=['target1']),948 SimpleTestJobDefinition('job3', targets=['target3'],949 depends=['target2']),950 ]951 build_manager = BuildManager(jobs, [])952 execution_manager = ExecutionManager(build_manager,953 ExtendedMockExecutor)954 build_graph = execution_manager.build955 build_graph.add_job("job3", {})956 job1 = build_graph.get_job("job1")957 job2 = build_graph.get_job("job2")958 job3 = build_graph.get_job("job3")959 job1.should_run_immediate = True960 job2.should_run_immediate = False961 job3.should_run_immediate = True962 job1.parents_should_run = False963 job2.parents_should_run = True964 job3.parents_should_run = True965 super_target1 = build_graph.get_target("super_target1")966 target1 = build_graph.get_target("target1")967 target2 = build_graph.get_target("target2")968 target3 = build_graph.get_target("target3")969 super_target1.do_get_mtime = mock.Mock(return_value=100)970 target1.do_get_mtime = mock.Mock(return_value=150)971 target2.do_get_mtime = mock.Mock(return_value=200)972 target3.do_get_mtime = mock.Mock(return_value=None)973 # When974 execution_manager.external_update_targets(['super_target1'])975 # Then976 self.assertFalse(execution_manager._work_queue.empty())977 self.assertEqual(execution_manager._work_queue.get(False), 'job3')978 self.assertTrue(execution_manager._work_queue.empty())979 def test_update_target_should_run(self):980 # Given981 jobs = [982 SimpleTestJobDefinition('job1', targets=['target1']),983 SimpleTestJobDefinition('job2', targets=['target2'],984 depends=['target1']),985 SimpleTestJobDefinition('job3', targets=['target3'],986 depends=['target2']),987 ]988 build_manager = BuildManager(jobs, [])989 execution_manager = ExecutionManager(build_manager,990 ExtendedMockExecutor)991 build_graph = execution_manager.build992 build_graph.add_job("job3", {})993 job1 = build_graph.get_job("job1")994 job2 = build_graph.get_job("job2")995 job3 = build_graph.get_job("job3")996 job1.should_run_immediate = False997 job2.should_run_immediate = False998 job3.should_run_immediate = True999 job1.parents_should_run = False1000 job2.parents_should_run = False1001 job3.parents_should_run = False1002 target1 = build_graph.get_target("target1")1003 target2 = build_graph.get_target("target2")1004 target3 = build_graph.get_target("target3")1005 target1.do_get_mtime = mock.Mock(return_value=None)1006 target2.do_get_mtime = mock.Mock(return_value=100)1007 target3.do_get_mtime = mock.Mock(return_value=50)1008 # When1009 execution_manager.external_update_targets(['target1'])1010 # Then1011 self.assertFalse(execution_manager._work_queue.empty())1012 self.assertEqual(execution_manager._work_queue.get(False), 'job1')1013 self.assertTrue(job3.get_parents_should_run())1014 self.assertTrue(execution_manager._work_queue.empty())1015 def test_update_target_multiple(self):1016 # Given1017 jobs = [1018 SimpleTestJobDefinition('job1', targets=['target1']),1019 SimpleTestJobDefinition('job2', targets=['target2'],1020 depends=['target1']),1021 SimpleTestJobDefinition('job3', targets=['target3'],1022 depends=['target2']),1023 SimpleTestJobDefinition("job1'", targets=["target1'"],1024 depends=["super_target1'"]),1025 SimpleTestJobDefinition("job2'", targets=["target2'"],1026 depends=["target1'"]),1027 SimpleTestJobDefinition("job3'", targets=["target3'"],1028 depends=["target2'"]),1029 ]1030 build_manager = BuildManager(jobs, [])1031 execution_manager = ExecutionManager(build_manager,1032 ExtendedMockExecutor)1033 build_graph = execution_manager.build1034 build_graph.add_job("job3", {})1035 build_graph.add_job("job3'", {})1036 job1 = build_graph.get_job("job1")1037 job2 = build_graph.get_job("job2")1038 job3 = build_graph.get_job("job3")1039 job1_ = build_graph.get_job("job1'")1040 job2_ = build_graph.get_job("job2'")1041 job3_ = build_graph.get_job("job3'")1042 job1.should_run_immediate = False1043 job2.should_run_immediate = False1044 job3.should_run_immediate = True1045 job1.parents_should_run = False1046 job2.parents_should_run = False1047 job3.parents_should_run = False1048 job1_.should_run_immediate = True1049 job2_.should_run_immediate = False1050 job3_.should_run_immediate = True1051 job1_.parents_should_run = False1052 job2_.parents_should_run = True1053 job3_.parents_should_run = True1054 target1 = build_graph.get_target("target1")1055 target2 = build_graph.get_target("target2")1056 target3 = build_graph.get_target("target3")1057 target1.do_get_mtime = mock.Mock(return_value=None)1058 target2.do_get_mtime = mock.Mock(return_value=100)1059 target3.do_get_mtime = mock.Mock(return_value=50)1060 super_target1_ = build_graph.get_target("super_target1'")1061 target1_ = build_graph.get_target("target1'")1062 target2_ = build_graph.get_target("target2'")1063 target3_ = build_graph.get_target("target3'")1064 super_target1_.do_get_mtime = mock.Mock(return_value=100)1065 target1_.do_get_mtime = mock.Mock(return_value=150)1066 target2_.do_get_mtime = mock.Mock(return_value=200)1067 target3_.do_get_mtime = mock.Mock(return_value=None)1068 # When1069 execution_manager.external_update_targets(["target1", "super_target1'"])1070 # Then1071 self.assertFalse(execution_manager._work_queue.empty())1072 work_job1 = execution_manager._work_queue.get(False)1073 self.assertFalse(execution_manager._work_queue.empty())1074 work_job2 = execution_manager._work_queue.get(False)1075 self.assertIn(work_job1, ["job3'", "job1"])1076 self.assertIn(work_job2, ["job3'", "job1"])1077 self.assertNotEqual(work_job1, work_job2)1078 self.assertTrue(job3.get_parents_should_run())1079 self.assertTrue(execution_manager._work_queue.empty())1080 def test_force_existing(self):1081 # Given1082 jobs = [1083 EffectJobDefinition(1084 "job1", depends=[1085 {1086 "unexpanded_id": "super_target1",1087 "start_mtime": 100,1088 }1089 ], targets=["target1"], effect=2001090 )1091 ]1092 build_manager = BuildManager(jobs, [])1093 execution_manager = ExecutionManager(build_manager,1094 ExtendedMockExecutor)1095 execution_manager.executor.execute = mock.Mock(wraps=execution_manager.executor.execute)1096 execution_manager.running = True1097 execution_manager.submit("job1", {})1098 execution_manager.start_execution(inline=True)1099 # When1100 execution_manager.running = True1101 execution_manager.submit("job1", {})1102 self.assertEqual(execution_manager._work_queue.qsize(), 0)1103 execution_manager.start_execution(inline=True)1104 execution_manager.running = True1105 execution_manager.submit("job1", {}, force=True)1106 self.assertEqual(execution_manager._work_queue.qsize(), 1)1107 execution_manager.start_execution(inline=True)1108 # Then...
BuildPipe.py
Source:BuildPipe.py
1import os2import shutil3import subprocess4import sys5import tempfile6import typing7import zipfile8from abc import ABC9import json10HOME = os.path.dirname(__file__)11class ProjectView:12 """13 Helper class for in-memory file manipulation14 todo: export to common place / library15 """16 def __init__(self):17 self.path_lookup = {}18 self.path_cache = {}19 self.modified_files = set()20 def with_directory_source(21 self,22 directory: str,23 filter_files: typing.Callable[[str], bool] = lambda _: True,24 ):25 """26 Adds the files from a directory to the project view27 :param directory: the directory to load from28 :param filter_files: a filter function; useful when only special files should be included29 """30 for root, dirs, files in os.walk(directory):31 for file in files:32 whole_file = os.path.join(root, file).replace("\\", "/")33 f = os.path.relpath(whole_file, directory).replace("\\", "/")34 if filter_files(whole_file):35 self.path_lookup[f] = whole_file36 self.modified_files.add(f)37 return self38 def copy(self) -> "ProjectView":39 """40 Copies the ProjectView class; Writing to the files not loaded into memory are still changed in the view41 """42 instance = ProjectView()43 instance.path_lookup = self.path_lookup.copy()44 instance.path_cache = self.path_cache.copy()45 instance.modified_files = self.modified_files.copy()46 return instance47 def read(self, file: str, cache=False) -> bytes:48 """49 Reads a relative file50 :param file: the relative file51 :param cache: if the read data should be cached; useful when planning to access it multiple times in the near52 future53 :return: the data54 """55 if file in self.path_cache:56 return self.path_cache[file]57 with open(self.path_lookup[file], mode="rb") as f:58 data = f.read()59 if cache:60 self.path_cache[file] = data61 return data62 def write(self, file: str, data: bytes):63 """64 Writes data into the local cache, overriding existing data, and overriding the original file data previously65 accessable via read()66 :param file: the file67 :param data: the data to write68 """69 assert isinstance(data, bytes)70 self.path_cache[file] = data71 self.modified_files.add(file)72 def dump_into_directory(73 self,74 directory: str,75 file_filter: typing.Callable[[str], bool] = lambda _: True,76 ):77 """78 Writes all affected files into the given directory (all files accessed by with_directory_source79 and write()-en to)80 :param directory: the directory to write to81 :param file_filter: a filter for the files82 """83 for file in self.modified_files:84 if not file_filter(file):85 continue86 f = os.path.join(directory, file)87 d = os.path.dirname(f)88 if not os.path.exists(d):89 os.makedirs(d)90 if file in self.path_cache:91 with open(f, mode="wb") as f:92 f.write(self.path_cache[file])93 elif file in self.path_lookup:94 shutil.copyfile(self.path_lookup[file], f)95 else:96 print("skipping file", file, "as the file is not found")97 def dump_into_zipfile(98 self, file: str, file_filter: typing.Callable[[str], bool] = lambda _: True99 ):100 """101 Writes all affected files into the given zipfile (all files accessed by with_directory_source102 and write()-en to)103 :param file: the file to write into104 :param file_filter: a filter for the files105 """106 with zipfile.ZipFile(file, mode="w") as zip_file:107 for file in self.modified_files:108 if not file_filter(file):109 continue110 if file in self.path_cache:111 with zip_file.open(file, mode="w") as f:112 f.write(self.path_cache[file])113 elif file in self.path_lookup:114 zip_file.write(self.path_lookup[file], file)115 else:116 print("skipping file", file, "as the file is not found")117 def filter_in_place(self, file_filter: typing.Callable[[str], bool]):118 self.modified_files = set(filter(file_filter, self.modified_files))119 def merge(self, other: "ProjectView"):120 for file in other.modified_files:121 if file in other.path_cache:122 self.path_cache[file] = other.path_cache[file]123 elif file in other.path_lookup:124 self.path_lookup[file] = other.path_lookup[file]125 else:126 print("skipping file", file, "as the file is not found")127 continue128 self.modified_files.add(file)129 def print_stats(self):130 print(self.modified_files)131 def clear(self):132 self.modified_files.clear()133 self.path_cache.clear()134 self.path_lookup.clear()135class AbstractBuildStage(ABC):136 """137 Base class for a stage in the build system138 """139 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):140 raise NotImplementedError141class AbstractProjectPreparation(ABC):142 """143 Base class for a preparation stage; A stage affecting files, and not consuming a ProjectView.144 The ProjectView is created based on the results of this transformer145 """146 def execute_in(self, directory: str, build_manager):147 raise NotImplementedError148class ProjectBuildManager:149 """150 The manager, storing how to build certain things151 """152 def __init__(self):153 self.preparation_stages: typing.List[AbstractProjectPreparation] = []154 self.stages: typing.List[AbstractBuildStage] = []155 self.build_name = None156 self.version_id = None157 def add_preparation_stage(self, stage: AbstractProjectPreparation):158 """159 Adds such preparation stage to the internal list160 :param stage: the stage161 """162 self.preparation_stages.append(stage)163 return self164 def add_stage(self, stage: AbstractBuildStage):165 """166 Adds a normal stage into the internal list167 :param stage: the stage to add168 """169 self.stages.append(stage)170 return self171 def run(172 self,173 directory: str,174 build_output_dir: str,175 project_view_consumer: typing.Callable[[ProjectView], None] = None,176 ):177 """178 Runs the build configuration onto the given directory and outputs the data at the given directory179 :param directory: the directory to use as a source180 :param build_output_dir: the directory to output to181 :param project_view_consumer: a consumer for the project view, for additonal changes182 """183 if not os.path.isdir(build_output_dir):184 os.makedirs(build_output_dir)185 for preparation in self.preparation_stages:186 print("running preparation {}".format(preparation))187 preparation.execute_in(directory, self)188 view = ProjectView().with_directory_source(directory)189 if callable(project_view_consumer):190 project_view_consumer(view)191 for stage in self.stages:192 print("running stage {}".format(stage))193 try:194 stage.execute_on(view, build_output_dir, self)195 except:196 view.print_stats()197 raise198 return view199class BlackCodeFormattingPreparation(AbstractProjectPreparation):200 def execute_in(self, directory: str, build_manager):201 subprocess.call([sys.executable, "-m", "black", directory])202class PyMinifierTask(AbstractBuildStage):203 def __init__(self, special_config=None):204 if special_config is None:205 special_config = {}206 self.special_config = special_config207 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):208 try:209 import python_minifier210 except ImportError:211 subprocess.Popen(212 [sys.executable, "-m", "pip", "install", "python-minifier"]213 )214 import python_minifier215 for file in view.modified_files:216 if file.endswith(".py"):217 if file not in self.special_config:218 try:219 view.write(220 file,221 python_minifier.minify(222 view.read(file).decode("utf-8"),223 preserve_locals=["NAME"],224 remove_literal_statements=True,225 ).encode("utf-8"),226 )227 except:228 print(file)229 raise230 else:231 view.write(232 file,233 python_minifier.minify(234 view.read(file).decode("utf-8"), **self.special_config[file]235 ).encode("utf-8"),236 )237class JsonMinifierTask(AbstractBuildStage):238 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):239 for file in view.modified_files:240 if file.endswith(".json"):241 view.write(242 file,243 json.dumps(json.loads(view.read(file).decode("utf-8"))).encode(244 "utf-8"245 ),246 )247class BuildSplitStage(AbstractBuildStage):248 """249 A stage for splitting the current build chain into a sub-chain not modifying the base chain250 """251 def __init__(self, *parts: AbstractBuildStage, merge_back=False):252 self.parts = parts253 self.merge_back = merge_back254 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):255 view = view.copy() if not self.merge_back else view256 for part in self.parts:257 print("running stage {}".format(part))258 part.execute_on(view, build_output_dir, None)259class BuildSplitUsingManagerAndTMPCache(AbstractBuildStage):260 """261 Similar to BuildSplitStage, but takes a whole ProjectBuildManager.262 Data is written to a temporary directory263 """264 def __init__(self, manager: ProjectBuildManager, merge_back=False):265 self.manager = manager266 self.merge_back = merge_back267 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):268 directory = tempfile.TemporaryDirectory()269 view.dump_into_directory(directory.name)270 view2 = self.manager.run(directory.name, build_output_dir)271 if self.merge_back:272 view.merge(view2)273 directory.cleanup()274class DumpTask(AbstractBuildStage):275 """276 Task for dumping the whole file tree277 as_zip defines if the data should be written to a zip file or not278 file_filter is passed to the dump function279 """280 def __init__(281 self,282 file_or_dir_name: str,283 as_zip=True,284 file_filter: typing.Callable[[str], bool] = lambda _: True,285 ):286 self.file_or_dir_name = file_or_dir_name287 self.as_zip = as_zip288 self.file_filter = file_filter289 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):290 if self.as_zip:291 view.dump_into_zipfile(292 os.path.join(build_output_dir, self.file_or_dir_name).replace(293 "\\", "/"294 ),295 file_filter=self.file_filter,296 )297 else:298 view.dump_into_directory(299 os.path.join(build_output_dir, self.file_or_dir_name).replace(300 "\\", "/"301 ),302 file_filter=self.file_filter,303 )304class FileFilterTask(AbstractBuildStage):305 """306 Helper for filtering the project view by file name307 """308 def __init__(self, file_filter: typing.Callable[[str], bool]):309 self.file_filter = file_filter310 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):311 view.filter_in_place(self.file_filter)312class FileRenameTask(AbstractBuildStage):313 """314 Helper for renaming certain files in the tree315 """316 def __init__(self, rename: typing.Callable[[str], str]):317 self.rename = rename318 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):319 for file in view.modified_files:320 new_file = self.rename(file)321 if new_file != file:322 view.modified_files.remove(file)323 view.modified_files.add(new_file)324 if file in view.path_cache:325 view.path_cache[new_file] = view.path_cache.pop(file)326 if file in view.path_lookup:327 view.path_lookup[new_file] = view.path_lookup.pop(file)328class FilePrefixRenameTask(FileRenameTask):329 def __init__(self, renames_from, renames_to):330 self.renames_from = renames_from331 self.renames_to = renames_to332 super().__init__(self.rename_task)333 def rename_task(self, file: str) -> str:334 if file.startswith(self.renames_from):335 return self.renames_to + file.removeprefix(self.renames_from)336 return file337 def __repr__(self):338 return "FilePrefixRenameTask(from='{}',to='{}')".format(339 self.renames_from, self.renames_to340 )341PY_FILE_FILTER = FileFilterTask(lambda file: file.endswith(".py"))342class NuitkaBuild(AbstractBuildStage):343 def __init__(344 self,345 launch_args=None,346 executable_name: str = "result.exe",347 look_for_output="{tmp}/launch.exe",348 ):349 self.look_for_output = look_for_output350 self.executable_name = executable_name351 if launch_args is None:352 launch_args = []353 self.launch_args = launch_args354 def execute_on(self, view: ProjectView, build_output_dir: str, build_manager):355 args = [arg.format(output_dir=build_output_dir) for arg in self.launch_args]356 tmp = tempfile.TemporaryDirectory()357 view.dump_into_directory(tmp.name)358 subprocess.call(359 [sys.executable, "-m", "nuitka"] + args + [tmp.name + "/launch.py"]360 )361 shutil.copyfile(362 self.look_for_output.format(tmp=tmp.name, output_dir=build_output_dir),363 build_output_dir + "/" + self.executable_name,...
village.py
Source:village.py
1from page_objects.resource_fields_page import ResourceFieldsPage2from modell.build_manager import BuildManager3from PyQt5 import QtCore4import datetime5class Village(QtCore.QObject):6 def __init__(self, village_num: int):7 super(Village, self).__init__()8 self.village_num = village_num9 self.resource_fields = ResourceFieldsPage()10 self.build_manager = BuildManager()11 self.connect_build_manager_to_resource_fields()12 self.is_building_resource_field = bool13 self.next_resource_task_available_at = datetime.datetime14 self.village_name = str15 self.initialize_village()16 def initialize_village(self):17 village_elements = self.resource_fields.get_elements("//*[@class='villageList']//*[@class='iconAndNameWrapper']")18 self.village_name = village_elements[self.village_num].text()19 village_elements[self.village_num].click()20 self.build_manager.ask_building_status()21 self.build_manager.ask_for_fields()22 self.build_manager.ask_for_resources()23 self.is_building_resource_field = self.resource_fields.is_building_resource()24 self.set_next_resource_task()25 def connect_build_manager_to_resource_fields(self):26 self.build_manager.ask_building_status_signal.connect(self.resource_fields.read_resource_building_status)27 self.resource_fields.resource_building_status.connect(self.build_manager.set_building_status)28 self.build_manager.ask_resources_signal.connect(self.resource_fields.read_resources)29 self.resource_fields.resources_signal.connect(self.build_manager.set_resources)30 self.build_manager.ask_fields_signal.connect(self.resource_fields.read_fields)31 self.resource_fields.fields_signal.connect(self.build_manager.set_fields)32 self.build_manager.build_resource_field_signal.connect(self.resource_fields.build_field)33 def build_until(self):34 return self.build_manager.building_until35 def resource_fields_completed(self) -> bool:36 """37 bug for main village: if all resource fields are AT level 10, the bot will think, that it is completed38 """39 for resource_field in self.build_manager.fields:40 if resource_field.field_level != 10:41 return False42 return True43 def set_next_resource_task(self):44 self.next_resource_task_available_at =\45 datetime.datetime.now() + self.resource_fields.read_resource_building_status()46 # print(self.next_resource_task_available_at)47 def emit_building_done_at_signal(self):48 self.building_done_at_signal.emit(self.build_manager.building_until)49 def __repr__(self):...
Automation Testing Tutorials
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run prospector automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
