Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.