Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _run_hook method in ATX

Best Python code snippet using ATX

update_issues.py

Source:update_issues.py

...89        return list.__getitem__(self, commit_id)90    def url(self):91        return 'mock/repo'92class TestChangegroupHook(_TestBase):93    def _run_hook(self, commit_messages, warning_count=0, update_count=0):94        repo = _MockRepo(commit_messages)95        update_issues.changegroup_hook(self.ui, repo, 0)96        warnings = self.ui.warn.call_args_list97        updates = self.trac_proxy_mock.ticket.update.call_args_list98        self.assertEqual(len(warnings), warning_count)99        self.assertEqual(len(updates), update_count)100        return updates101    def test_commits_with_invalid_message_format_ignored(self):102        self._run_hook([103            '',104            'Issue #1337 - Extraneous characters in issue number',105            'Issue 1337',  # No dash, no message.106            'Issue 1337: Colon instead of dash',107            'Noissue no dash',108            'Issue 1337-No space around dash',109            'Fixes 1337 no dash',110        ], warning_count=7)111    def test_noissue_commits_ignored(self):112        self._run_hook(['Noissue - Foo', 'noissue - Bar'])  # No updates.113    def test_single_issue_referenced(self):114        updates = self._run_hook(['Issue 1337 - Foo'], update_count=1)115        self.assertEqual(updates[0][0][0], 1337)116    def test_multiline_commit_message(self):117        updates = self._run_hook(['Issue 1337 - Foo\nBar',118                                  'Issue 1337 - Bar.\nBaz',119                                  'Fixes 2448 - Foo\n\nBar',120                                  'Fixes 2448 - Bar\n \nBaz'],121                                 update_count=2)122        comment_1337 = updates[0][0][1]123        self.assertIn('Issue 1337 - Foo...]', comment_1337)124        self.assertIn('Issue 1337 - Bar...]', comment_1337)125        comment_2448 = updates[1][0][1]126        self.assertIn('Fixes 2448 - Foo]', comment_2448)127        self.assertIn('Fixes 2448 - Bar]', comment_2448)128    def test_multiline_commit_message_crlf(self):129        updates = self._run_hook(['Issue 1337 - Foo\r\nBar',130                                  'Issue 1337 - Bar.\r\nBaz',131                                  'Fixes 2448 - Foo\r\n\r\nBar',132                                  'Fixes 2448 - Bar\r\n \r\nBaz'],133                                 update_count=2)134        comment_1337 = updates[0][0][1]135        self.assertIn('Issue 1337 - Foo...]', comment_1337)136        self.assertIn('Issue 1337 - Bar...]', comment_1337)137        comment_2448 = updates[1][0][1]138        self.assertIn('Fixes 2448 - Foo]', comment_2448)139        self.assertIn('Fixes 2448 - Bar]', comment_2448)140    def test_missing_issue_referenced(self):141        self._run_hook(['Issue 42 - Bar'], warning_count=1)142    def test_multiple_issues_referenced(self):143        updates = self._run_hook(['Issue 1337, fixes 2448 - Foo'],144                                 update_count=2)145        self.assertEqual(updates[0][0][0], 1337)146        self.assertEqual(updates[1][0][0], 2448)147    def test_multiple_commits_for_issue(self):148        updates = self._run_hook(['Issue 1337 - Foo', 'Fixes 1337 - Bar'],149                                 update_count=1)150        comment = updates[0][0][1]151        self.assertIn('000000000000', comment)152        self.assertIn('000000000100', comment)153class TestPushkeyHook(_TestBase):154    def _run_hook(self, commit_messages, bookmark='master',155                  warning_count=0, update_count=0):156        repo = _MockRepo(['Base', 'Old'] + commit_messages)157        update_issues.pushkey_hook(self.ui, repo,158                                   namespace='bookmarks', key=bookmark,159                                   old=1, new=1 + len(commit_messages))160        warnings = self.ui.warn.call_args_list161        updates = self.trac_proxy_mock.ticket.update.call_args_list162        self.assertEqual(len(warnings), warning_count)163        self.assertEqual(len(updates), update_count)164        return updates165    def _check_update(self, update, issue_id, action='resolve',166                      milestone='current'):167        self.assertEqual(update[0][0], issue_id)168        changes = update[0][2]169        self.assertEqual(changes['action'], action)170        if milestone is None:171            self.assertNotIn('milestone', changes)172        else:173            self.assertEqual(changes['milestone'], milestone)174    def test_move_other_bookmark(self):175        self._run_hook(['Fixes 1337 - Foo'], bookmark='other')  # No updates.176    def test_one_issue_fixed(self):177        updates = self._run_hook(['Fixes 1337 - Foo'], update_count=1)178        self._check_update(updates[0], 1337)179    def test_fix_closed_issue(self):180        updates = self._run_hook(['fixes 2448 - Foo'], update_count=1)181        self._check_update(updates[0], 2448, action='leave', milestone='other')182    def test_fix_issue_noregexp(self):183        updates = self._run_hook(['Fixes 4670 - Foo'], update_count=1)184        self._check_update(updates[0], 4670, milestone=None)185    def test_fix_issue_no_matching_milestones(self):186        updates = self._run_hook(['Fixes 5781 - Foo'], update_count=1)187        self._check_update(updates[0], 5781, milestone=None)188    def test_fix_many(self):189        updates = self._run_hook(['Fixes 1337 - Foo', 'Fixes 2448 - Bar'],190                                 update_count=2)191        self._check_update(updates[0], 1337)192        self._check_update(updates[1], 2448, action='leave', milestone='other')193    def test_fix_nonexistent(self):194        self._run_hook(['Fixes 7331 - Foo'], warning_count=1)195    def test_fix_closed_with_assigned_milestone(self):196        self._run_hook(['fixes 3559 - Foo'])  # No updates.197if __name__ == '__main__':...

app.py

Source:app.py

...31        }32        self.args = args33        self.kwargs = kwargs34        logging.info(f'{self.title} [{self.version}] inited')35    def _run_hook(self, hook_name, *args):36        if not self.hooks[hook_name]:37            return38        self.hooks[hook_name](*args)39    def add_hook(self, hook_name, hook_func):40        if hook_name not in self.hooks.keys():41            logging.error(f'Unknown hook {hook_name}')42            return43        self.hooks[hook_name] = hook_func44    def _init_client(self, base_request, send):45        client = Client(base_request, send)46        self.clients[base_request.ws_id] = client47        return client48    def _remove_client(self, client):49        self.clients.pop(client.ws_id)50    def _make_request(self, base_request, message):51        try:52            parsed = json.loads(message['text'])53        except json.JSONDecodeError:54            raise ParseError55        if not validate_jrpc_request(parsed):56            raise InvalidRequest57        if parsed['method'] not in self.methods.keys():58            raise NotFound59        return base_request._make_request(parsed)60    async def _make_response(self, request, client):61        return await self.methods[request.method](request, client)62    def add_method(self, name, method):63        self.methods[name] = method64    def method(self, name):65        def wrapped(method):66            self.methods[name] = method67            return method68        return wrapped69    def add_user(self, client, user):70        client.user = user71        self.users[user.id] = client72    def remove_user(self, client):73        self.users.pop(client.user.id)74    def before(self, func):75        def decorator(method):76            async def wrapped(req, client):77                req, client = func(req, client)78                return await method(req, client)79            return wrapped80        return decorator81    def after(self, func):82        def decorator(method):83            async def wrapped(req, client):84                resp = await method(req, client)85                return func(req, resp, client)86            return wrapped87        return decorator88    async def __call__(self, scope, receive, send):89        if scope['type'] == 'lifespan':90            message = await receive()91            if message['type'] == 'lifespan.startup':92                self._run_hook('before_startup', scope, receive, send)93                await send({'type': 'lifespan.startup.complete'})94                self._run_hook('after_startup', scope, receive, send)95                return96            elif message['type'] == 'lifespan.shutdown':97                self._run_hook('before_shutdown', scope, receive, send)98                await send({'type': 'lifespan.shutdown.complete'})99                self._run_hook('after_shutdown', scope, receive, send)100                return101        elif scope['type'] == 'websocket':102            pass103        else:104            logging.error(f"Unsupported scope type - {scope['type']}")105            return106        message = await receive()107        if message['type'] == 'websocket.connect':108            self._run_hook('before_connection', scope, receive, send, message)109        else:110            return111        await send(jrpc_accept())112        self._run_hook('after_connection', scope, receive, send)113        base_request = Request(scope, receive, send)114        client = self._init_client(base_request, send)115        while True:116            message = await receive()117            if message['type'] in ('websocket.disconnect', 'websocket.close'):118                self._run_hook('on_disconnection', message, client)119                break120            elif message['type'] == 'websocket.receive':121                try:122                    request = self._make_request(base_request, message)123                except CorvusException as e:124                    await send(e.error())125                    continue126                self._run_hook('on_request', request, client)127                try:128                    response = await self._make_response(request, client)129                except CorvusException as e:130                    await send(e.error())131                    continue132                self._run_hook('on_response', response, client)133                if response:134                    await send(jrpc_response(response, request.id))135            else:136                await send(UnknownAsgiAction.error())...

hooks.py

Source:hooks.py

...46            if result:47                _LOGGER.debug("ouput of command %s:\n%s" % (command, output))48            return result, output49        return True, None50    def _run_hook(self, hook, node, *parameters):51        parameters = [ str(p) for p in list(parameters)]52        result, output = self._check_command_and_run(_CONFIGURATION_HOOKS.WORKING_FOLDER, hook, [ node ] + list(parameters))53        return result54    def pre_poweron(self, node, *parameters):55        return self._run_hook(_CONFIGURATION_HOOKS.PRE_POWERON, node, *parameters)56    def post_poweron(self, node, *parameters):57        return self._run_hook(_CONFIGURATION_HOOKS.POST_POWERON, node, *parameters)58    def pre_poweroff(self, node, *parameters):59        return self._run_hook(_CONFIGURATION_HOOKS.PRE_POWEROFF, node, *parameters)60    def post_poweroff(self, node, *parameters):61        return self._run_hook(_CONFIGURATION_HOOKS.POST_POWEROFF, node, *parameters)62    def poweredon(self, node, *parameters):63        return self._run_hook(_CONFIGURATION_HOOKS.POWEREDON, node, *parameters)64    def poweredoff(self, node, *parameters):65        return self._run_hook(_CONFIGURATION_HOOKS.POWEREDOFF, node, *parameters)66    def unexpected_poweron(self, node, *parameters):67        return self._run_hook(_CONFIGURATION_HOOKS.UNEXPECTED_POWERON, node, *parameters)68    def unexpected_poweroff(self, node, *parameters):69        return self._run_hook(_CONFIGURATION_HOOKS.UNEXPECTED_POWEROFF, node, *parameters)70    def onerr(self, node, *parameters):71        return self._run_hook(_CONFIGURATION_HOOKS.ONERR, node, *parameters)72    def offerr(self, node, *parameters):73        return self._run_hook(_CONFIGURATION_HOOKS.OFFERR, node, *parameters)74    def unknown(self, node, *parameters):75        return self._run_hook(_CONFIGURATION_HOOKS.UNKNOWN, node, *parameters)76    def idle(self, node, *parameters):77        return self._run_hook(_CONFIGURATION_HOOKS.IDLE, node, *parameters)78    def used(self, node, *parameters):79        return self._run_hook(_CONFIGURATION_HOOKS.USED, node, *parameters)80    def request(self, request):81        requests_string = ';'.join(request.resources.resources.requests)82        parameters = (request.resources.resources.slots, request.resources.resources.memory, request.resources.taskcount, request.resources.maxtaskspernode, requests_string)83        return self._run_hook(_CONFIGURATION_HOOKS.REQUEST, str(request.id), *parameters)84try:85    HOOKS86except:...

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.