How to use get_input_element method in pypom_form

Best Python code snippet using pypom_form_python

do_fns_error_handling.py

Source:do_fns_error_handling.py

...8This file contains DoFn classes representing functions like map, flatMap and filter but with error handling.9Errors are caught in a except bloc and a failure object are put in a side output.10The failure contains the current input element and exception.11"""12def get_input_element(element) -> str:13    """14    Get the input element for the Failure object.15    For a dict a json string is returned16    For other object, the string representation of the object is returned.17    """18    return json.dumps(element) if isinstance(element, dict) else str(element)19class FlatMap(beam.DoFn):20    """21    Custom DnFn class representing a flatMap operation with error handling.22    """23    def __init__(self,24                 step: str,25                 input_element_mapper: Callable[[Any], Any],26                 setup_action: Callable[[], None] = lambda: None,27                 start_bundle_action: Callable[[], None] = lambda: None,28                 finish_bundle_action: Callable[[], None] = lambda: None,29                 teardown_action: Callable[[], None] = lambda: None):30        self.step = step31        self.input_element_mapper = input_element_mapper32        self.setup_action = setup_action33        self.start_bundle_action = start_bundle_action34        self.finish_bundle_action = finish_bundle_action35        self.teardown_action = teardown_action36        super().__init__()37    def setup(self):38        self.setup_action()39    def start_bundle(self):40        self.start_bundle_action()41    def finish_bundle(self):42        self.finish_bundle_action()43    def teardown(self):44        self.teardown_action()45    def process(self, element, *args, **kwargs):46        try:47            results: Iterable = self.input_element_mapper(element, *args, **kwargs)48            for result in results:49                yield result50        except Exception as err:51            failure = Failure(52                pipeline_step=self.step,53                input_element=get_input_element(element),54                exception=err55            )56            yield pvalue.TaggedOutput(FAILURES, failure)57class Map(beam.DoFn):58    """59    Custom DnFn class representing a map operation with error handling.60    """61    def __init__(self,62                 step: str,63                 input_element_mapper: Callable[[Any], Any],64                 setup_action: Callable[[], None] = lambda: None,65                 start_bundle_action: Callable[[], None] = lambda: None,66                 finish_bundle_action: Callable[[], None] = lambda: None,67                 teardown_action: Callable[[], None] = lambda: None):68        self.step = step69        self.input_element_mapper = input_element_mapper70        self.setup_action = setup_action71        self.start_bundle_action = start_bundle_action72        self.finish_bundle_action = finish_bundle_action73        self.teardown_action = teardown_action74        super().__init__()75    def setup(self):76        self.setup_action()77    def start_bundle(self):78        self.start_bundle_action()79    def finish_bundle(self):80        self.finish_bundle_action()81    def teardown(self):82        self.teardown_action()83    def process(self, element, *args, **kwargs):84        try:85            yield self.input_element_mapper(element, *args, **kwargs)86        except Exception as err:87            failure = Failure(88                pipeline_step=self.step,89                input_element=get_input_element(element),90                exception=err91            )92            yield pvalue.TaggedOutput(FAILURES, failure)93class Filter(beam.DoFn):94    """95    Custom DnFn class representing a filter operation with error handling.96    """97    def __init__(self,98                 step: str,99                 input_element_predicate: Callable[[Any], bool]):100        self.step = step101        self.input_element_predicate = input_element_predicate102        super().__init__()103    def process(self, element, *args, **kwargs):104        try:105            if self.input_element_predicate(element, *args, **kwargs):106                yield element107        except Exception as err:108            failure = Failure(109                pipeline_step=self.step,110                input_element=get_input_element(element),111                exception=err112            )...

main.py

Source:main.py

...43        self.users_nav_bar()44        if self.input_frame:45            self.input_frame.destroy()46        if(self.socket.current_room):47            self.get_input_element(self.main_frame).grid(row=1, column=0)48    def get_input_element(self, master):49        self.input_frame = tk.Frame(master=master, background="gray", bd=3)50        #51        input_field = tk.Text(self.input_frame, height=2, width=50)52        input_field.pack(side=tk.LEFT)53        input_field.focus_force()54        input_field.bind("<Return>", lambda x: self.socket.send_message(input_field.get("1.0", "end-1c")))55        #56        send_button = tk.Button(self.main_frame, text="send", relief="raised", command=lambda: self.socket.send_message(input_field.get("1.0", "end-1c")))57        send_button.grid(row=1, column=1)58        return self.input_frame59    def render_messages(self, master):60        ui.messages(master, self.messages_accessor.get_messages()[-15:], self.user_id)61    def users_nav_bar(self):62        if self.frame_users_nav:...

functional_test.py

Source:functional_test.py

...10        self.browser.quit()11    def check_h1(self, text):12        self.assertIn(text, self.browser.find_element_by_tag_name('h1').text)13    14    def get_input_element(self):15        return self.browser.find_element_by_id('answer_box')16    def send_answer(self, text):17        self.get_input_element().send_keys(text)18        self.get_input_element().send_keys(Keys.ENTER)19    def test_winning_the_game(self):20        # want to play the game21        # open the browser and go on the game website22        self.browser.get('http://localhost:8080/')23        # start page opens24        # "Central Corridor" room25        # read the text26        self.assertIn('Gothons From Planet Percal #25', self.browser.title)27        self.check_h1('Central Corridor')28        # type the right phrase 'tell a joke' and hit enter29        self.send_answer('tell a joke')30        # see the "Laser Weapon Armory" room page31        # read the text32        # type the right phrase '0132' and hit enter...

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.