Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
