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

Copied to Clipboard!

How to use fetch_task method in lisa

Best Python code snippet using lisa_python

test_magpie.py

Source:test_magpie.py

...29    assert backlog_len() == 030    response = magpie.request("user1", "/task_add task42")31    assert response == "â task42 has been added."32    assert backlog_len() == 133    assert fetch_task("task42") == Task("task42", set(), TaskStatus.NEW)34def test_add_task2():35    clear_enviroment()36    magpie = Magpie()37    assert backlog_len() == 038    response = magpie.request("user1", "/task_add task1 tag2 tag1")39    assert response == "â task1 has been added.\ntask1 relates to tag1, tag2."40    assert backlog_len() == 141    assert fetch_task("task1") == Task(42        "task1", {"tag1", "tag2"}, TaskStatus.NEW)43def test_add_tasks1():44    clear_enviroment()45    magpie = Magpie()46    assert backlog_len() == 047    magpie.request("user1", "/task_add task1 tag1 tag2")48    magpie.request("user1", "/task_add task2 tag2")49    assert backlog_len() == 250    assert fetch_task("task1") == Task(51        "task1", {"tag1", "tag2"}, TaskStatus.NEW)52    assert fetch_task("task2") == Task("task2", {"tag2"}, TaskStatus.NEW)53def test_backlog():54    clear_enviroment()55    magpie = Magpie()56    response = magpie.request("user1", "/backlog")57    assert response == "backlog is empty."58    magpie.request("user1", "/task_add task1 tag1 tag2")59    magpie.request("user1", "/task_add task2 tag2")60    magpie.request("user1", "/task_add task3 tag1 tag2 tag3")61    response = magpie.request("user1", "/backlog")62    assert response == "backlog:\n" \63                       "ð¦ NEW(3):\n" \64                       "task1 relates to tag1, tag2.\n" \65                       "task2 relates to tag2.\n" \66                       "task3 relates to tag1, tag2, tag3."67    magpie.request("user1", "/task_add task4")68    magpie.request("user2", "/task_start task1")69    magpie.request("user3", "/task_done task2")70    magpie.request("user3", "/task_start task1")71    magpie.request("user3", "/task_stop task1")72    magpie.request("user3", "/task_start task4")73    magpie.request("user3", "/task_stop task4")74    response = magpie.request("user1", "/backlog")75    assert backlog_len() == 476    assert response == "backlog:\n" \77                       "ð¦ NEW(1):\ntask3 relates to tag1, tag2, tag3.\n\n" \78                       "â¸ SUSPENDED(1):\ntask4 relates to .\n\n" \79                       "ð  IN PROGRESS(1):\ntask1 relates to tag1, tag2.\n\n" \80                       "â DONE(1):\ntask2 relates to tag2."81def test_start_stop_single_user():82    clear_enviroment()83    magpie = Magpie()84    magpie.request("user1", "/task_add task1 tag1 tag2 tag3")85    response = magpie.request("user1", "/task_start task1")86    assert fetch_task("task1") == Task(87        "task1", {"tag1", "tag2", "tag3"}, TaskStatus.IN_PROGRESS)88    assert fetch_task_perform("task1", "user1") == TaskPerform(89        "user1", "task1", 0, [0])90    assert storage.fetch_all_task_sessions("task1") == {"user1"}91    assert response == "ð  you started working on task1.\n" \92                       "task1 relates to tag1, tag2, tag3.\n" \93                       "â no one else currently working on task1."94    run_time_machine(4)95    response = magpie.request("user1", "/task_stop task1")96    assert not storage.is_have_any_sessions()97    assert fetch_task("task1") == Task(98        "task1", {"tag1", "tag2", "tag3"}, TaskStatus.SUSPENDED)99    assert fetch_task_perform("task1", "user1") == TaskPerform(100        "user1", "task1", 4, [4])101    assert response == "â¸ you have finished work on task1.\n" \102                       "a total of 4.0 hours were spent on task1.\n" \103                       "today you spent on task1 4.0 hours.\n" \104                       "please mark the time spent."105def test_time_format():106    clear_enviroment()107    magpie = Magpie()108    magpie.request("user1", "/task_add task1 tag1 tag2 tag3")109    magpie.request("user1", "/task_start task1")110    run_time_machine(0.49)111    response = magpie.request("user1", "/task_stop task1")112    assert response == "â¸ you have finished work on task1.\n" \113                       "a total of 0.5 hours were spent on task1.\n" \114                       "today you spent on task1 0.5 hours.\n" \115                       "please mark the time spent."116def test_start_stop_many_users():117    clear_enviroment()118    magpie = Magpie()119    assert not storage.is_have_any_sessions()120    magpie.request("manager", "/task_add task1 tag1 tag2")121    run_time_machine(1)122    magpie.request("developer1", "/task_start task1")123    assert storage.fetch_all_task_sessions("task1") == {"developer1"}124    run_time_machine(2)125    response = magpie.request("developer2", "/task_start task1")126    assert storage.fetch_all_task_sessions("task1") == {"developer1", "developer2"}127    assert response == "ð  you started working on task1.\n" \128                       "task1 relates to tag1, tag2.\n" \129                       "ð¤ developer1 currently working on task1 also."130    response = magpie.request("developer3", "/task_start task1")131    assert storage.fetch_all_task_sessions("task1") == {"developer1", "developer2", "developer3"}132    assert response == "ð  you started working on task1.\n" \133                       "task1 relates to tag1, tag2.\n" \134                       "ð¤ developer1, developer2 currently working on task1 also."135    magpie.request("developer3", "/task_stop task1")136    run_time_machine(3)137    magpie.request("developer1", "/task_stop task1")138    assert storage.fetch_all_task_sessions("task1") == {"developer2"}139    assert fetch_task("task1") == Task(140        "task1", {"tag1", "tag2"}, TaskStatus.IN_PROGRESS)141    assert fetch_task_perform("task1", "developer1") == TaskPerform(142        "developer1", "task1", 5, [5])143    assert fetch_task_perform("task1", "developer2") == TaskPerform(144        "developer2", "task1", 3, [3])145    run_time_machine(1)146    magpie.request("developer2", "/task_stop task1")147    assert not storage.is_have_any_sessions()148    assert fetch_task("task1") == Task(149        "task1", {"tag1", "tag2"}, TaskStatus.SUSPENDED)150    assert fetch_task_perform("task1", "developer1") == TaskPerform(151        "developer1", "task1", 5, [5])152    assert fetch_task_perform("task1", "developer2") == TaskPerform(153        "developer2", "task1", 3 + 1, [3 + 1])154def test_add_tags():155    clear_enviroment()156    magpie = Magpie()157    magpie.request("manager", "/task_add task1")158    assert len(fetch_task("task1").tags) == 0159    response = magpie.request("manager", "/tag_add task1 tag1")160    assert fetch_task("task1").tags == {"tag1"}161    assert response == "tags for task1 updated. task1 now relates to tag1."162    response = magpie.request("developer", "/tag_add task1 tag2 tag3")163    assert fetch_task("task1").tags == {"tag1", "tag2", "tag3"}164    assert response == "tags for task1 updated. task1 now relates to tag1, tag2, tag3."165def test_events_spent_time():166    clear_enviroment()167    magpie = Magpie()168    magpie.request("manager", "/task_add task1 tag1 tag2")169    magpie.request("manager", "/task_add task2 tag1 tag3")170    magpie.request("developer1", "/task_start task1")171    run_time_machine(4)172    magpie.request("developer2", "/task_start task1")173    run_time_machine(5)174    magpie.request("developer1", "/task_stop task1")175    run_time_machine(1)176    magpie.request("developer2", "/task_stop task1")177    response = magpie.request("manager", "/events")178    assert response == "events for task1:\n" \179    "â± a total of 15.0 hours were spent on task.\n" \180               "â¹ task relates to tag1, tag2.\n" \181               "â  developer1 spent 9.0 hours on task in a single session.\n" \182               "\n" \183               "no events for task2.\n" \184               "\n"185def test_events_new_tags():186    clear_enviroment()187    magpie = Magpie()188    magpie.request("manager", "/task_add task1 tag1 tag2")189    run_time_machine(2)190    magpie.request("developer1", "/tag_add task1 tag3")191    run_time_machine(4)192    response = magpie.request("manager", "/events")193    assert response == "events for task1:\n" \194    "â¹ task relates to tag1, tag2, tag3.\n" \195               "â developer1 added new tags for task: tag3.\n\n"196def test_crunch_reminder():197    clear_enviroment()198    EVENT_HANDLERS[EventType.CRUNCH] = crunch_reminder199    magpie = Magpie()200    magpie.request("manager", "/task_add task1 tag1 tag2")201    magpie.request("developer1", "/task_start task1")202    run_time_machine(7)203    assert len(EVENTS_LOG) == 0204    assert len(MAILBOX) == 0205    run_time_machine(1.5)206    assert len(EVENTS_LOG) == 0207    assert len(MAILBOX) == 1208    assert MAILBOX == [209    ("developer1", "â  you are working on task1 over 8 hours.")]210def test_daily_report():211    clear_enviroment()212    magpie = Magpie()213    response = magpie.request("manager", "/daily_report")214    assert response == "there is no implementations for '/daily_report' command. YET."215def test_weekly_report():216    clear_enviroment()217    magpie = Magpie()218    response = magpie.request("manager", "/weekly_report")219    assert response == "there is no implementations for '/weekly_report' command. YET."220def test_task_done():221    clear_enviroment()222    magpie = Magpie()223    magpie.request("manager", "/task_add task1 tag1")224    assert fetch_task("task1") == Task("task1", {"tag1"}, TaskStatus.NEW)225    response = magpie.request("manager", "/task_done task1")226    assert fetch_task("task1") == Task("task1", {"tag1"}, TaskStatus.DONE)227    assert response == "you marked task1 as done."228def test_events_task_done():229    clear_enviroment()230    magpie = Magpie()231    magpie.request("manager", "/task_add task1 tag1")232    response = magpie.request("developer", "/task_done task1")233    response = magpie.request("manager", "/events")234    assert response == "events for task1:\n" \235	    	       "â¹ task relates to tag1.\n" \...

backup_play.py

Source:backup_play.py

...62                supported_action_name = self.SUPPORTED_MODULE_ARGS.keys()63                # if action name is file, ignore state option in directory and absent64                if task.action == 'file':65                    fetch_path = self._get_dest_path(task)66                    fetch_task = self._make_fetch_task(task, fetch_path)67                    if 'state' in task.args:68                        if not task.args['state'] in ('directory', 'absent'):69                            new_block.block.append(fetch_task)70                    else:71                        new_block.block.append(fetch_task)72                    #stat_task = self._make_stat_task(task, fetch_path)73                    #new_block.block.append(stat_task)74                #suppored action eg. copy, template75                elif task.action in supported_action_name:76                    fetch_path = self._get_dest_path(task)77                    fetch_task = self._make_fetch_task(task, fetch_path)78                    new_block.block.append(fetch_task)79                    #stat_task = self._make_stat_task(task, fetch_path)80                    #new_block.block.append(stat_task)81                #allowed action eg. meta, import_tasks82                elif task.action in self.ALLOWED_ACTION:83                    new_block.block.append(task)84                # TODO: support include_tasks85            # if Block object, compile tasks recursion86            elif isinstance(block, Block):87                task = self._backup_compilie_block(task)88                new_block.block.append(task)89            else:90                new_block.block.append(task)91        return new_block92    def _get_dest_path(self, task):93        dest_path = ""94        for arg in self.SUPPORTED_MODULE_ARGS[task.action]:95            if arg in task.args:96                dest_path = task.args[arg]97                break98            msg = '%s module require [%s] in args' % (task.action, self.SUPPORTED_MODULE_ARGS[task.action])99            raise Exception(msg)100        if dest_path.endswith("/"):101            if 'src' in task.args:102                file_name = os.path.basename(task.args['src'])103                dest_path = os.path.join(dest_path, file_name)104            else:105                msg = '%s module is must not be directory' % task.action106                raise Exception(msg)107        return dest_path108    def _make_fetch_task(self, task, path):109        fetch_task = deepcopy(task)110        fetch_task.name = "<fetch|%s> %s" % (task.action, task.name)111        fetch_task.action = "fetch"112        fetch_task.args = {113            'src': path,114            'dest': self.dest115        }116        fetch_task.changed_when = []117        fetch_task.failed_when = []118        fetch_task.notify = None119        return fetch_task120    def _make_stat_task(self, task, path):121        stat_task = deepcopy(task)122        stat_task.name = "<stat|%s> %s" % (task.action, task.name)...

image_loader.py

Source:image_loader.py

1from PyQt5 import QtNetwork2from PyQt5.QtCore import QUrl, Qt3from PyQt5.QtGui import QPixmap4from PyQt5.QtWidgets import QLabel, QApplication, QVBoxLayout, QMainWindow, QWidget5import threading6class Glide:7    def __init__(self, label: QLabel, url: str, width=None, height=None):8        super().__init__()9        self.label = label10        self.req = QtNetwork.QNetworkRequest(QUrl(url))11        self.width = width12        self.height = height13        self.fetch_task = None14    def start(self, network_access_manager: QtNetwork.QNetworkAccessManager):15        t = threading.current_thread()16        print(f'start at {t.ident}, {t.name}')17        self.fetch_task = network_access_manager.get(self.req)18        self.fetch_task.finished.connect(self.resolve_fetch)19    def resolve_fetch(self):20        t = threading.current_thread()21        print(f'resolve at {t.ident} {t.name}')22        img_binary = self.fetch_task.readAll()23        pixmap = QPixmap()24        pixmap.loadFromData(img_binary)25        if self.width is not None and self.height is not None:26            pixmap.scaled(self.width, self.height, aspectRatioMode=Qt.KeepAspectRatio,27                          transformMode=Qt.SmoothTransformation)28        self.label.setPixmap(pixmap)29class TestWindow(QMainWindow):30    def __init__(self):31        super().__init__()32        self.network_access_manager = QtNetwork.QNetworkAccessManager()33        widget = QWidget()34        v_layout = QVBoxLayout()35        label1 = QLabel()36        v_layout.addWidget(label1)37        Glide(label1, "https://live.staticflickr.com/65535/49251422908_591245c64a_c_d.jpg")\38            .start(self.network_access_manager)39        widget.setLayout(v_layout)40        self.setCentralWidget(widget)41def test():42    app = QApplication([])43    window = TestWindow()44    window.show()45    app.exec()46if __name__ == '__main__':...

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.