How to use get_page method in localstack
Best Python code snippet using localstack_python
test_api.py
Source:test_api.py 
1"""2Test APIs.3"""4from os.path import expanduser5import json6import pytest7from mock import patch, Mock8from coursera import api9from coursera import define10from coursera.test.utils import slurp_fixture, links_to_plain_text11from coursera.utils import BeautifulSoup12from requests.exceptions import HTTPError13from requests import Response14@pytest.fixture15def course():16 course = api.CourseraOnDemand(17 session=Mock(cookies={}), course_id='0', course_name='test_course')18 return course19@patch('coursera.api.get_page')20def test_extract_links_from_programming_http_error(get_page, course):21 """22 This test checks that downloader skips locked programming assignments23 instead of throwing an error. (Locked == returning 403 error code)24 """25 locked_response = Response()26 locked_response.status_code = define.HTTP_FORBIDDEN27 get_page.side_effect = HTTPError('Mocked HTTP error',28 response=locked_response)29 assert None == course.extract_links_from_programming('0')30@patch('coursera.api.get_page')31def test_extract_links_from_exam_http_error(get_page, course):32 """33 This test checks that downloader skips locked exams34 instead of throwing an error. (Locked == returning 403 error code)35 """36 locked_response = Response()37 locked_response.status_code = define.HTTP_FORBIDDEN38 get_page.side_effect = HTTPError('Mocked HTTP error',39 response=locked_response)40 assert None == course.extract_links_from_exam('0')41@patch('coursera.api.get_page')42def test_extract_links_from_supplement_http_error(get_page, course):43 """44 This test checks that downloader skips locked supplements45 instead of throwing an error. (Locked == returning 403 error code)46 """47 locked_response = Response()48 locked_response.status_code = define.HTTP_FORBIDDEN49 get_page.side_effect = HTTPError('Mocked HTTP error',50 response=locked_response)51 assert None == course.extract_links_from_supplement('0')52@patch('coursera.api.get_page')53def test_extract_links_from_lecture_http_error(get_page, course):54 """55 This test checks that downloader skips locked lectures56 instead of throwing an error. (Locked == returning 403 error code)57 """58 locked_response = Response()59 locked_response.status_code = define.HTTP_FORBIDDEN60 get_page.side_effect = HTTPError('Mocked HTTP error',61 response=locked_response)62 assert None == course.extract_links_from_lecture('fake_course_id', '0')63@patch('coursera.api.get_page')64def test_extract_links_from_quiz_http_error(get_page, course):65 """66 This test checks that downloader skips locked quizzes67 instead of throwing an error. (Locked == returning 403 error code)68 """69 locked_response = Response()70 locked_response.status_code = define.HTTP_FORBIDDEN71 get_page.side_effect = HTTPError('Mocked HTTP error',72 response=locked_response)73 assert None == course.extract_links_from_quiz('0')74@patch('coursera.api.get_page')75def test_extract_references_poll_http_error(get_page, course):76 """77 This test checks that downloader skips locked programming assignments78 instead of throwing an error. (Locked == returning 403 error code)79 """80 locked_response = Response()81 locked_response.status_code = define.HTTP_FORBIDDEN82 get_page.side_effect = HTTPError('Mocked HTTP error',83 response=locked_response)84 assert None == course.extract_references_poll()85@patch('coursera.api.get_page')86def test_extract_links_from_reference_http_error(get_page, course):87 """88 This test checks that downloader skips locked resources89 instead of throwing an error. (Locked == returning 403 error code)90 """91 locked_response = Response()92 locked_response.status_code = define.HTTP_FORBIDDEN93 get_page.side_effect = HTTPError('Mocked HTTP error',94 response=locked_response)95 assert None == course.extract_links_from_reference('0')96@patch('coursera.api.get_page')97def test_extract_links_from_programming_immediate_instructions_http_error(98 get_page, course):99 """100 This test checks that downloader skips locked programming immediate instructions101 instead of throwing an error. (Locked == returning 403 error code)102 """103 locked_response = Response()104 locked_response.status_code = define.HTTP_FORBIDDEN105 get_page.side_effect = HTTPError('Mocked HTTP error',106 response=locked_response)107 assert (108 None == course.extract_links_from_programming_immediate_instructions('0'))109@patch('coursera.api.get_page')110def test_ondemand_programming_supplement_no_instructions(get_page, course):111 no_instructions = slurp_fixture(112 'json/supplement-programming-no-instructions.json')113 get_page.return_value = json.loads(no_instructions)114 output = course.extract_links_from_programming('0')115 assert {} == output116@patch('coursera.api.get_page')117@pytest.mark.parametrize(118 "input_filename,expected_output", [119 ('peer-assignment-instructions-all.json', 'intro Review criteria section'),120 ('peer-assignment-instructions-no-title.json', 'intro section'),121 ('peer-assignment-instructions-only-introduction.json', 'intro'),122 ('peer-assignment-instructions-only-sections.json', 'Review criteria section'),123 ('peer-assignment-no-instructions.json', ''),124 ]125)126def test_ondemand_from_peer_assignment_instructions(127 get_page, course, input_filename, expected_output):128 instructions = slurp_fixture('json/%s' % input_filename)129 get_page.return_value = json.loads(instructions)130 output = course.extract_links_from_peer_assignment('0')131 assert expected_output == links_to_plain_text(output)132@patch('coursera.api.get_page')133def test_ondemand_from_programming_immediate_instructions_no_instructions(134 get_page, course):135 no_instructions = slurp_fixture(136 'json/supplement-programming-immediate-instructions-no-instructions.json')137 get_page.return_value = json.loads(no_instructions)138 output = course.extract_links_from_programming_immediate_instructions('0')139 assert {} == output140@patch('coursera.api.get_page')141def test_ondemand_programming_supplement_empty_instructions(get_page, course):142 empty_instructions = slurp_fixture(143 'json/supplement-programming-empty-instructions.json')144 get_page.return_value = json.loads(empty_instructions)145 output = course.extract_links_from_programming('0')146 # Make sure that SOME html content has been extracted, but remove147 # it immediately because it's a hassle to properly prepare test input148 # for it. FIXME later.149 assert 'html' in output150 del output['html']151 assert {} == output152@patch('coursera.api.get_page')153def test_ondemand_programming_immediate_instructions_empty_instructions(154 get_page, course):155 empty_instructions = slurp_fixture(156 'json/supplement-programming-immediate-instructions-empty-instructions.json')157 get_page.return_value = json.loads(empty_instructions)158 output = course.extract_links_from_programming_immediate_instructions('0')159 # Make sure that SOME html content has been extracted, but remove160 # it immediately because it's a hassle to properly prepare test input161 # for it. FIXME later.162 assert 'html' in output163 del output['html']164 assert {} == output165@patch('coursera.api.get_page')166def test_ondemand_programming_supplement_one_asset(get_page, course):167 one_asset_tag = slurp_fixture('json/supplement-programming-one-asset.json')168 one_asset_url = slurp_fixture('json/asset-urls-one.json')169 asset_json = json.loads(one_asset_url)170 get_page.side_effect = [json.loads(one_asset_tag),171 json.loads(one_asset_url)]172 expected_output = {'pdf': [(asset_json['elements'][0]['url'],173 'statement-pca')]}174 output = course.extract_links_from_programming('0')175 # Make sure that SOME html content has been extracted, but remove176 # it immediately because it's a hassle to properly prepare test input177 # for it. FIXME later.178 assert 'html' in output179 del output['html']180 assert expected_output == output181@patch('coursera.api.get_page')182def test_extract_references_poll(get_page, course):183 """184 Test extracting course references.185 """186 get_page.side_effect = [187 json.loads(slurp_fixture('json/references-poll-reply.json'))188 ]189 expected_output = json.loads(190 slurp_fixture('json/references-poll-output.json'))191 output = course.extract_references_poll()192 assert expected_output == output193@patch('coursera.api.get_page')194def test_ondemand_programming_immediate_instructions_one_asset(get_page, course):195 one_asset_tag = slurp_fixture(196 'json/supplement-programming-immediate-instructions-one-asset.json')197 one_asset_url = slurp_fixture('json/asset-urls-one.json')198 asset_json = json.loads(one_asset_url)199 get_page.side_effect = [json.loads(one_asset_tag),200 json.loads(one_asset_url)]201 expected_output = {'pdf': [(asset_json['elements'][0]['url'],202 'statement-pca')]}203 output = course.extract_links_from_programming_immediate_instructions('0')204 # Make sure that SOME html content has been extracted, but remove205 # it immediately because it's a hassle to properly prepare test input206 # for it. FIXME later.207 assert 'html' in output208 del output['html']209 assert expected_output == output210@patch('coursera.api.get_page')211def test_ondemand_programming_supplement_three_assets(get_page, course):212 three_assets_tag = slurp_fixture(213 'json/supplement-programming-three-assets.json')214 three_assets_url = slurp_fixture('json/asset-urls-three.json')215 get_page.side_effect = [json.loads(three_assets_tag),216 json.loads(three_assets_url)]217 expected_output = json.loads(slurp_fixture(218 'json/supplement-three-assets-output.json'))219 output = course.extract_links_from_programming('0')220 output = json.loads(json.dumps(output))221 # Make sure that SOME html content has been extracted, but remove222 # it immediately because it's a hassle to properly prepare test input223 # for it. FIXME later.224 assert 'html' in output225 del output['html']226 assert expected_output == output227@patch('coursera.api.get_page')228def test_extract_links_from_lecture_assets_typename_asset(get_page, course):229 open_course_assets_reply = slurp_fixture(230 'json/supplement-open-course-assets-reply.json')231 api_assets_v1_reply = slurp_fixture(232 'json/supplement-api-assets-v1-reply.json')233 get_page.side_effect = [json.loads(open_course_assets_reply),234 json.loads(api_assets_v1_reply)]235 expected_output = json.loads(slurp_fixture(236 'json/supplement-extract-links-from-lectures-output.json'))237 assets = ['giAxucdaEeWJTQ5WTi8YJQ']238 output = course._extract_links_from_lecture_assets(assets)239 output = json.loads(json.dumps(output))240 assert expected_output == output241@patch('coursera.api.get_page')242def test_extract_links_from_lecture_assets_typname_url_and_asset(get_page, course):243 """244 This test makes sure that _extract_links_from_lecture_assets grabs url245 links both from typename == 'asset' and == 'url'.246 """247 get_page.side_effect = [248 json.loads(slurp_fixture(249 'json/supplement-open-course-assets-typename-url-reply-1.json')),250 json.loads(slurp_fixture(251 'json/supplement-open-course-assets-typename-url-reply-2.json')),252 json.loads(slurp_fixture(253 'json/supplement-open-course-assets-typename-url-reply-3.json')),254 json.loads(slurp_fixture(255 'json/supplement-open-course-assets-typename-url-reply-4.json')),256 json.loads(slurp_fixture(257 'json/supplement-open-course-assets-typename-url-reply-5.json')),258 ]259 expected_output = json.loads(slurp_fixture(260 'json/supplement-extract-links-from-lectures-url-asset-output.json'))261 assets = ['Yry0spSKEeW8oA5fR3afVQ',262 'kMQyUZSLEeWj-hLVp2Pm8w',263 'xkAloZmJEeWjYA4jOOgP8Q']264 output = course._extract_links_from_lecture_assets(assets)265 output = json.loads(json.dumps(output))266 assert expected_output == output267@patch('coursera.api.get_page')268def test_list_courses(get_page, course):269 """270 Test course listing method.271 """272 get_page.side_effect = [273 json.loads(slurp_fixture('json/list-courses-input.json'))274 ]275 expected_output = json.loads(276 slurp_fixture('json/list-courses-output.json'))277 expected_output = expected_output['courses']278 output = course.list_courses()279 assert expected_output == output280@pytest.mark.parametrize(281 "input_filename,output_filename,subtitle_language,video_id", [282 ('video-reply-1.json', 'video-output-1.json',283 'en,zh-CN|zh-TW', "None"),284 ('video-reply-1.json', 'video-output-1-en.json',285 'zh-TW', "None"),286 ('video-reply-1.json', 'video-output-1-en.json',287 'en', "None"),288 ('video-reply-1.json', 'video-output-1-all.json',289 'all', "None"),290 ('video-reply-1.json', 'video-output-1-all.json',291 'zh-TW,all|zh-CN', "None"),292 ('video-reply-2.json', 'video-output-2.json',293 'en,zh-CN|zh-TW', "None"),294 ]295)296def test_extract_subtitles_from_video_dom(input_filename, output_filename, subtitle_language, video_id):297 video_dom = json.loads(slurp_fixture('json/%s' % input_filename))298 expected_output = json.loads(slurp_fixture('json/%s' % output_filename))299 course = api.CourseraOnDemand(300 session=Mock(cookies={}), course_id='0', course_name='test_course')301 actual_output = course._extract_subtitles_from_video_dom(302 video_dom, subtitle_language, video_id)303 actual_output = json.loads(json.dumps(actual_output))304 assert actual_output == expected_output305@pytest.mark.parametrize(306 "input_filename,output_filename", [307 ('empty-input.json', 'empty-output.txt'),308 ('answer-text-replaced-with-span-input.json',309 'answer-text-replaced-with-span-output.txt'),310 ('question-type-textExactMatch-input.json',311 'question-type-textExactMatch-output.txt'),312 ('question-type-regex-input.json', 'question-type-regex-output.txt'),313 ('question-type-mathExpression-input.json',314 'question-type-mathExpression-output.txt'),315 ('question-type-checkbox-input.json', 'question-type-checkbox-output.txt'),316 ('question-type-mcq-input.json', 'question-type-mcq-output.txt'),317 ('question-type-singleNumeric-input.json',318 'question-type-singleNumeric-output.txt'),319 ('question-type-reflect-input.json', 'question-type-reflect-output.txt'),320 ('question-type-mcqReflect-input.json',321 'question-type-mcqReflect-output.txt'),322 ('question-type-unknown-input.json', 'question-type-unknown-output.txt'),323 ('multiple-questions-input.json', 'multiple-questions-output.txt'),324 ]325)326def test_quiz_exam_to_markup_converter(input_filename, output_filename):327 quiz_json = json.loads(slurp_fixture(328 'json/quiz-to-markup/%s' % input_filename))329 expected_output = slurp_fixture(330 'json/quiz-to-markup/%s' % output_filename).strip()331 converter = api.QuizExamToMarkupConverter(session=None)332 actual_output = converter(quiz_json).strip()333 # print('>%s<' % expected_output)334 # print('>%s<' % actual_output)335 assert actual_output == expected_output336class TestMarkupToHTMLConverter:337 def _p(self, html):338 return BeautifulSoup(html).prettify()339 STYLE = None340 def setup_method(self, test_method):341 self.STYLE = self._p(342 "".join([define.INSTRUCTIONS_HTML_INJECTION_PRE,343 define.INSTRUCTIONS_HTML_MATHJAX_URL,344 define.INSTRUCTIONS_HTML_INJECTION_AFTER])345 )346 self.markup_to_html = api.MarkupToHTMLConverter(session=None)347 ALTERNATIVE_MATHJAX_CDN = "https://alternative/mathjax/cdn.js"348 self.STYLE_WITH_ALTER = self._p(349 "".join([define.INSTRUCTIONS_HTML_INJECTION_PRE,350 ALTERNATIVE_MATHJAX_CDN,351 define.INSTRUCTIONS_HTML_INJECTION_AFTER])352 )353 self.markup_to_html_with_alter_mjcdn = api.MarkupToHTMLConverter(354 session=None, mathjax_cdn_url=ALTERNATIVE_MATHJAX_CDN)355 def test_empty(self):356 output = self.markup_to_html("")357 output_with_alter_mjcdn = self.markup_to_html_with_alter_mjcdn("")358 markup = """359 <meta charset="UTF-8"/>360 """361 assert self._p(markup) + self.STYLE == output362 assert self._p(markup) + \363 self.STYLE_WITH_ALTER == output_with_alter_mjcdn364 def test_replace_text_tag(self):365 markup = """366 <co-content>367 <text>368 Test<text>Nested</text>369 </text>370 <text>371 Test2372 </text>373 </co-content>374 """375 result = """376 <meta charset="UTF-8"/>377 <co-content>378 <p>379 Test<p>Nested</p>380 </p>381 <p>382 Test2383 </p>384 </co-content>\n385 """386 output = self.markup_to_html(markup)387 output_with_alter_mjcdn = self.markup_to_html_with_alter_mjcdn(markup)388 assert self._p(result) + self.STYLE == output389 assert self._p(result) + \390 self.STYLE_WITH_ALTER == output_with_alter_mjcdn391 def test_replace_heading(self):392 output = self.markup_to_html("""393 <co-content>394 <heading level="1">Text</heading>395 <heading level="2">Text</heading>396 <heading level="3">Text</heading>397 <heading level="4">Text</heading>398 <heading level="5">Text</heading>399 <heading >Text</heading>400 </co-content>401 """)402 assert self._p("""403 <meta charset="UTF-8"/>404 <co-content>405 <h1 level="1">Text</h1>406 <h2 level="2">Text</h2>407 <h3 level="3">Text</h3>408 <h4 level="4">Text</h4>409 <h5 level="5">Text</h5>410 <h1>Text</h1>411 </co-content>\n412 """) + self.STYLE == output413 def test_replace_code(self):414 output = self.markup_to_html("""415 <co-content>416 <code>Text</code>417 <code>Text</code>418 </co-content>419 """)420 assert self._p("""421 <meta charset="UTF-8"/>422 <co-content>423 <pre>Text</pre>424 <pre>Text</pre>425 </co-content>\n426 """) + self.STYLE == output427 def test_replace_list(self):428 output = self.markup_to_html("""429 <co-content>430 <list bullettype="numbers">Text</list>431 <list bullettype="bullets">Text</list>432 </co-content>433 """)434 assert self._p("""435 <meta charset="UTF-8"/>436 <co-content>437 <ol bullettype="numbers">Text</ol>438 <ul bullettype="bullets">Text</ul>439 </co-content>\n440 """) + self.STYLE == output441 @patch('coursera.api.AssetRetriever')442 def test_replace_images(self, mock_asset_retriever):443 replies = {444 'nVhIAj61EeaGyBLfiQeo_w': Mock(data=b'a', content_type='image/png'),445 'vdqUTz61Eea_CQ5dfWSAjQ': Mock(data=b'b', content_type='image/png'),446 'nodata': Mock(data=None, content_type='image/png')447 }448 mock_asset_retriever.__call__ = Mock(return_value=None)449 mock_asset_retriever.__getitem__ = Mock(450 side_effect=replies.__getitem__)451 self.markup_to_html._asset_retriever = mock_asset_retriever452 output = self.markup_to_html("""453 <co-content>454 <text>\n\n</text>455 <img assetId=\"nVhIAj61EeaGyBLfiQeo_w\" alt=\"\"/>456 <text>\n\n</text>457 <img assetId=\"vdqUTz61Eea_CQ5dfWSAjQ\" alt=\"\"/>458 <text>\n\n</text>459 </co-content>460 """)461 assert self._p("""462 <meta charset="UTF-8"/>463 <co-content>464 <p></p>465 <img alt="" assetid="nVhIAj61EeaGyBLfiQeo_w" src="data:image/png;base64,YQ=="/>466 <p></p>467 <img alt="" assetid="vdqUTz61Eea_CQ5dfWSAjQ" src="data:image/png;base64,Yg=="/>468 <p></p>469 </co-content>\n470 """) + self.STYLE == output471 @patch('coursera.api.AssetRetriever')472 def test_replace_audios(self, mock_asset_retriever):473 replies = {474 'aWTK9sYwEeW7AxLLCrgDQQ': Mock(data=b'a', content_type='audio/mpeg'),475 'bWTK9sYwEeW7AxLLCrgDQQ': Mock(data=b'b', content_type='unknown')476 }477 mock_asset_retriever.__call__ = Mock(return_value=None)478 mock_asset_retriever.__getitem__ = Mock(479 side_effect=replies.__getitem__)480 self.markup_to_html._asset_retriever = mock_asset_retriever481 output = self.markup_to_html("""482 <co-content>483 <asset id=\"aWTK9sYwEeW7AxLLCrgDQQ\" name=\"M111\" extension=\"mp3\" assetType=\"audio\"/>484 <asset id=\"bWTK9sYwEeW7AxLLCrgDQQ\" name=\"M112\" extension=\"mp3\" assetType=\"unknown\"/>485 </co-content>486 """)487 assert self._p("""488 <meta charset="UTF-8"/>489 <co-content>490 <asset assettype="audio" extension="mp3" id="aWTK9sYwEeW7AxLLCrgDQQ" name="M111">491 </asset>492 <audio controls="">493 Your browser does not support the audio element.494 <source src="data:audio/mpeg;base64,YQ==" type="audio/mpeg">495 </source>496 </audio>497 <asset assettype="unknown" extension="mp3" id="bWTK9sYwEeW7AxLLCrgDQQ" name="M112">498 </asset>499 </co-content>\n500 """) + self.STYLE == output501def test_quiz_converter():502 pytest.skip()503 quiz_to_markup = api.QuizExamToMarkupConverter(session=None)504 markup_to_html = api.MarkupToHTMLConverter(session=None)505 quiz_data = json.load(open('quiz.json'))['contentResponseBody']['return']506 result = markup_to_html(quiz_to_markup(quiz_data))507 # from ipdb import set_trace; set_trace(context=20)508 print('RESULT', result)509 with open('quiz.html', 'w') as file:510 file.write(result)511def test_quiz_converter_all():512 pytest.skip()513 import os514 from coursera.coursera_dl import get_session515 from coursera.cookies import login516 session = None517 session = get_session()518 quiz_to_markup = api.QuizExamToMarkupConverter(session=session)519 markup_to_html = api.MarkupToHTMLConverter(session=session)520 path = 'quiz_json'521 for filename in ['quiz-audio.json']: # os.listdir(path):522 # for filename in ['all_question_types.json']:523 # if 'YV0W4' not in filename:524 # continue525 # if 'QVHj1' not in filename:526 # continue527 #quiz_data = json.load(open('quiz.json'))['contentResponseBody']['return']528 current = os.path.join(path, filename)529 print(current)530 quiz_data = json.load(open(current))531 result = markup_to_html(quiz_to_markup(quiz_data))532 # from ipdb import set_trace; set_trace(context=20)533 # print('RESULT', result)534 with open('quiz_html/' + filename + '.html', 'w') as f:535 f.write(result)536def create_session():537 from coursera.coursera_dl import get_session538 from coursera.credentials import get_credentials539 from coursera.cookies import login540 session = get_session()541 username, password = get_credentials(netrc=expanduser('~/.netrc'))542 login(session, username, password)543 return session544@patch('coursera.api.get_page')545@patch('coursera.api.get_reply')546def test_asset_retriever(get_reply, get_page):547 reply = json.loads(slurp_fixture('json/asset-retriever/assets-reply.json'))548 get_page.side_effect = [reply]549 get_reply.side_effect = [Mock(status_code=200, content='<...>',550 headers=Mock(get=Mock(return_value='image/png')))] * 4551 asset_ids = ['bWTK9sYwEeW7AxLLCrgDQQ',552 'VceKeChKEeaOMw70NkE3iw',553 'VcmGXShKEea4ehL5RXz3EQ',554 'vdqUTz61Eea_CQ5dfWSAjQ']555 expected_output = [556 api.Asset(id="bWTK9sYwEeW7AxLLCrgDQQ", name="M111.mp3", type_name="audio",557 url="url4", content_type="image/png", data="<...>"),558 api.Asset(id="VceKeChKEeaOMw70NkE3iw", name="09_graph_decomposition_problems_1.pdf",559 type_name="pdf", url="url7", content_type="image/png", data="<...>"),560 api.Asset(id="VcmGXShKEea4ehL5RXz3EQ", name="09_graph_decomposition_starter_files_1.zip",561 type_name="generic", url="url2", content_type="image/png", data="<...>"),562 api.Asset(id="vdqUTz61Eea_CQ5dfWSAjQ", name="Capture.PNG",563 type_name="image", url="url9", content_type="image/png", data="<...>"),564 ]565 retriever = api.AssetRetriever(session=None)566 actual_output = retriever(asset_ids)567 assert expected_output == actual_output568def test_debug_asset_retriever():569 pytest.skip()570 asset_ids = ['bWTK9sYwEeW7AxLLCrgDQQ',571 'bXCx18YwEeWicwr5JH8fgw',572 'bX9X18YwEeW7AxLLCrgDQQ',573 'bYHvf8YwEeWFNA5XwZEiOw',574 'tZmigMYxEeWFNA5XwZEiOw']575 asset_ids = asset_ids[0:5]576 more = ['VceKeChKEeaOMw70NkE3iw',577 'VcmGXShKEea4ehL5RXz3EQ']578 print('session')579 session = create_session()580 retriever = api.AssetRetriever(session)581 #assets = retriever.get(asset_ids)582 assets = retriever(more)...
ActionHandler.py
Source:ActionHandler.py
...103 for file_path in self.init_file_paths:104 if file_path: self.main_window.new_page(file_path) #load pages from file paths105 if Preferences.file_open() in self.init_file_paths:106 self.main_window.new_page(Preferences.file_open(), show=True)107 if not self.get_page(): self.main_window.new_page() #ensure that at least a blank page exists108 elif action == Actions.APPLICATION_QUIT:109 if self.main_window.close_pages():110 gtk.main_quit()111 exit(0)112 ##################################################113 # Selections114 ##################################################115 elif action == Actions.ELEMENT_SELECT:116 pass #do nothing, update routines below117 elif action == Actions.NOTHING_SELECT:118 self.get_flow_graph().unselect()119 ##################################################120 # Enable/Disable121 ##################################################122 elif action == Actions.BLOCK_ENABLE:123 if self.get_flow_graph().enable_selected(True):124 self.get_flow_graph().update()125 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())126 self.get_page().set_saved(False)127 elif action == Actions.BLOCK_DISABLE:128 if self.get_flow_graph().enable_selected(False):129 self.get_flow_graph().update()130 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())131 self.get_page().set_saved(False)132 ##################################################133 # Cut/Copy/Paste134 ##################################################135 elif action == Actions.BLOCK_CUT:136 Actions.BLOCK_COPY()137 Actions.ELEMENT_DELETE()138 elif action == Actions.BLOCK_COPY:139 self.clipboard = self.get_flow_graph().copy_to_clipboard()140 elif action == Actions.BLOCK_PASTE:141 if self.clipboard:142 self.get_flow_graph().paste_from_clipboard(self.clipboard)143 self.get_flow_graph().update()144 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())145 self.get_page().set_saved(False)146 ##################################################147 # Move/Rotate/Delete/Create148 ##################################################149 elif action == Actions.BLOCK_MOVE:150 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())151 self.get_page().set_saved(False)152 elif action == Actions.BLOCK_ROTATE_CCW:153 if self.get_flow_graph().rotate_selected(90):154 self.get_flow_graph().update()155 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())156 self.get_page().set_saved(False)157 elif action == Actions.BLOCK_ROTATE_CW:158 if self.get_flow_graph().rotate_selected(-90):159 self.get_flow_graph().update()160 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())161 self.get_page().set_saved(False)162 elif action == Actions.ELEMENT_DELETE:163 if self.get_flow_graph().remove_selected():164 self.get_flow_graph().update()165 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())166 Actions.NOTHING_SELECT()167 self.get_page().set_saved(False)168 elif action == Actions.ELEMENT_CREATE:169 self.get_flow_graph().update()170 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())171 Actions.NOTHING_SELECT()172 self.get_page().set_saved(False)173 elif action == Actions.BLOCK_INC_TYPE:174 if self.get_flow_graph().type_controller_modify_selected(1):175 self.get_flow_graph().update()176 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())177 self.get_page().set_saved(False)178 elif action == Actions.BLOCK_DEC_TYPE:179 if self.get_flow_graph().type_controller_modify_selected(-1):180 self.get_flow_graph().update()181 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())182 self.get_page().set_saved(False)183 elif action == Actions.PORT_CONTROLLER_INC:184 if self.get_flow_graph().port_controller_modify_selected(1):185 self.get_flow_graph().update()186 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())187 self.get_page().set_saved(False)188 elif action == Actions.PORT_CONTROLLER_DEC:189 if self.get_flow_graph().port_controller_modify_selected(-1):190 self.get_flow_graph().update()191 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())192 self.get_page().set_saved(False)193 ##################################################194 # Window stuff195 ##################################################196 elif action == Actions.ABOUT_WINDOW_DISPLAY:197 Dialogs.AboutDialog(self.get_flow_graph().get_parent())198 elif action == Actions.HELP_WINDOW_DISPLAY:199 Dialogs.HelpDialog()200 elif action == Actions.TYPES_WINDOW_DISPLAY:201 Dialogs.TypesDialog(self.get_flow_graph().get_parent())202 elif action == Actions.ERRORS_WINDOW_DISPLAY:203 Dialogs.ErrorsDialog(self.get_flow_graph())204 ##################################################205 # Param Modifications206 ##################################################207 elif action == Actions.BLOCK_PARAM_MODIFY:208 selected_block = self.get_flow_graph().get_selected_block()209 if selected_block:210 if PropsDialog(selected_block).run():211 #save the new state212 self.get_flow_graph().update()213 self.get_page().get_state_cache().save_new_state(self.get_flow_graph().export_data())214 self.get_page().set_saved(False)215 else:216 #restore the current state217 n = self.get_page().get_state_cache().get_current_state()218 self.get_flow_graph().import_data(n)219 self.get_flow_graph().update()220 ##################################################221 # Undo/Redo222 ##################################################223 elif action == Actions.FLOW_GRAPH_UNDO:224 n = self.get_page().get_state_cache().get_prev_state()225 if n:226 self.get_flow_graph().unselect()227 self.get_flow_graph().import_data(n)228 self.get_flow_graph().update()229 self.get_page().set_saved(False)230 elif action == Actions.FLOW_GRAPH_REDO:231 n = self.get_page().get_state_cache().get_next_state()232 if n:233 self.get_flow_graph().unselect()234 self.get_flow_graph().import_data(n)235 self.get_flow_graph().update()236 self.get_page().set_saved(False)237 ##################################################238 # New/Open/Save/Close239 ##################################################240 elif action == Actions.FLOW_GRAPH_NEW:241 self.main_window.new_page()242 elif action == Actions.FLOW_GRAPH_OPEN:243 file_paths = OpenFlowGraphFileDialog(self.get_page().get_file_path()).run()244 if file_paths: #open a new page for each file, show only the first245 for i,file_path in enumerate(file_paths):246 self.main_window.new_page(file_path, show=(i==0))247 elif action == Actions.FLOW_GRAPH_CLOSE:248 self.main_window.close_page()249 elif action == Actions.FLOW_GRAPH_SAVE:250 #read-only or undefined file path, do save-as251 if self.get_page().get_read_only() or not self.get_page().get_file_path():252 Actions.FLOW_GRAPH_SAVE_AS()253 #otherwise try to save254 else:255 try:256 ParseXML.to_file(self.get_flow_graph().export_data(), self.get_page().get_file_path())257 self.get_page().set_saved(True)258 except IOError:259 Messages.send_fail_save(self.get_page().get_file_path())260 self.get_page().set_saved(False)261 elif action == Actions.FLOW_GRAPH_SAVE_AS:262 file_path = SaveFlowGraphFileDialog(self.get_page().get_file_path()).run()263 if file_path is not None:264 self.get_page().set_file_path(file_path)265 Actions.FLOW_GRAPH_SAVE()266 elif action == Actions.FLOW_GRAPH_SCREEN_CAPTURE:267 file_path = SaveImageFileDialog(self.get_page().get_file_path()).run()268 if file_path is not None:269 pixbuf = self.get_flow_graph().get_drawing_area().get_pixbuf()270 pixbuf.save(file_path, IMAGE_FILE_EXTENSION[1:])271 ##################################################272 # Gen/Exec/Stop273 ##################################################274 elif action == Actions.FLOW_GRAPH_GEN:275 if not self.get_page().get_proc():276 if not self.get_page().get_saved() or not self.get_page().get_file_path():277 Actions.FLOW_GRAPH_SAVE() #only save if file path missing or not saved278 if self.get_page().get_saved() and self.get_page().get_file_path():279 generator = self.get_page().get_generator()280 try:281 Messages.send_start_gen(generator.get_file_path())282 generator.write()283 except Exception,e: Messages.send_fail_gen(e)284 else: self.generator = None285 elif action == Actions.FLOW_GRAPH_EXEC:286 if not self.get_page().get_proc():287 Actions.FLOW_GRAPH_GEN()288 if self.get_page().get_saved() and self.get_page().get_file_path():289 ExecFlowGraphThread(self)290 elif action == Actions.FLOW_GRAPH_KILL:291 if self.get_page().get_proc():292 try: self.get_page().get_proc().kill()293 except: print "could not kill process: %d"%self.get_page().get_proc().pid294 elif action == Actions.PAGE_CHANGE: #pass and run the global actions295 pass296 else: print '!!! Action "%s" not handled !!!'%action297 ##################################################298 # Global Actions for all States299 ##################################################300 #update general buttons301 Actions.ERRORS_WINDOW_DISPLAY.set_sensitive(not self.get_flow_graph().is_valid())302 Actions.ELEMENT_DELETE.set_sensitive(bool(self.get_flow_graph().get_selected_elements()))303 Actions.BLOCK_PARAM_MODIFY.set_sensitive(bool(self.get_flow_graph().get_selected_block()))304 Actions.BLOCK_ROTATE_CCW.set_sensitive(bool(self.get_flow_graph().get_selected_blocks()))305 Actions.BLOCK_ROTATE_CW.set_sensitive(bool(self.get_flow_graph().get_selected_blocks()))306 #update cut/copy/paste307 Actions.BLOCK_CUT.set_sensitive(bool(self.get_flow_graph().get_selected_blocks()))308 Actions.BLOCK_COPY.set_sensitive(bool(self.get_flow_graph().get_selected_blocks()))309 Actions.BLOCK_PASTE.set_sensitive(bool(self.clipboard))310 #update enable/disable311 Actions.BLOCK_ENABLE.set_sensitive(bool(self.get_flow_graph().get_selected_blocks()))312 Actions.BLOCK_DISABLE.set_sensitive(bool(self.get_flow_graph().get_selected_blocks()))313 #set the exec and stop buttons314 self.update_exec_stop()315 #saved status316 Actions.FLOW_GRAPH_SAVE.set_sensitive(not self.get_page().get_saved())317 self.main_window.update()318 try: #set the size of the flow graph area (if changed)319 new_size = self.get_flow_graph().get_option('window_size')320 if self.get_flow_graph().get_size() != tuple(new_size):321 self.get_flow_graph().set_size(*new_size)322 except: pass323 #draw the flow graph324 self.get_flow_graph().update_selected()325 self.get_flow_graph().queue_draw()326 return True #action was handled327 def update_exec_stop(self):328 """329 Update the exec and stop buttons.330 Lock and unlock the mutex for race conditions with exec flow graph threads.331 """332 sensitive = self.get_flow_graph().is_valid() and not self.get_page().get_proc()333 Actions.FLOW_GRAPH_GEN.set_sensitive(sensitive)334 Actions.FLOW_GRAPH_EXEC.set_sensitive(sensitive)335 Actions.FLOW_GRAPH_KILL.set_sensitive(self.get_page().get_proc() != None)336class ExecFlowGraphThread(Thread):337 """Execute the flow graph as a new process and wait on it to finish."""338 def __init__ (self, action_handler):339 """340 ExecFlowGraphThread constructor.341 @param action_handler an instance of an ActionHandler342 """343 Thread.__init__(self)344 self.update_exec_stop = action_handler.update_exec_stop345 self.flow_graph = action_handler.get_flow_graph()346 #store page and dont use main window calls in run347 self.page = action_handler.get_page()348 Messages.send_start_exec(self.page.get_generator().get_file_path())349 #get the popen350 try:351 self.p = self.page.get_generator().get_popen()352 self.page.set_proc(self.p)353 #update354 self.update_exec_stop()355 self.start()356 except Exception, e:357 Messages.send_verbose_exec(str(e))358 Messages.send_end_exec()359 def run(self):360 """361 Wait on the executing process by reading from its stdout....
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 localstack 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?
