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

How to use _write_plan method in tappy

Best Python code snippet using tappy_python

ai_demo.py

Source:ai_demo.py

...99            self._write_iteration_header(file_object)100            self._write_prioritized_goal_list(diary_entry, file_object)101            self._write_goal(diary_entry, file_object)102            self._write_resource_list(diary_entry, file_object)103            self._write_plan(diary_entry, file_object)104            self._write_action(diary_entry, file_object)105            self._write_world_state_before(diary_entry, file_object)106            self._write_world_state_after(diary_entry, file_object)107            # Close the file108            file_object.close()109    @staticmethod110    def _write_world_state_after(diary_entry, file_object):111        # write the world state after the action was taken112        file_object.writelines("\n")113        for world_state in list(diary_entry['the_world_state_after'].keys()):114            world_state_value = diary_entry['the_world_state_after'][world_state]115            after_element = f"\n<after {world_state}={world_state_value}>"116            file_object.writelines(after_element)117    @staticmethod118    def _write_world_state_before(diary_entry, file_object):119        # write the world state before the action was taken120        file_object.writelines("\n")121        for world_state in list(diary_entry['the_world_state_before'].keys()):122            world_state_value = diary_entry['the_world_state_before'][world_state]123            before_element = f"\n<before {world_state}={world_state_value}>"124            file_object.writelines(before_element)125    @staticmethod126    def _write_action(diary_entry, file_object):127        # create and write the action taken as part of the given diary entry128        if "SUCCESS" in str(diary_entry['action_status']):129            action_status = "success"130        else:131            action_status = "fail"132        if diary_entry['action_taken'] == "":133            action_name = "NONE"134        else:135            action_name = diary_entry['action_taken']136        file_object.writelines(f"\n<action name={action_name} status={action_status}>")137    @staticmethod138    def _write_plan(diary_entry, file_object):139        # create and write the plan to achieve the currently-selected goal140        plan_open_tag = f"\n\n<plan>"141        plan_close_tag = "\n<plan>"142        plan_content = ""143        # include the plan steps in the action plan144        if diary_entry['my_plan'] is None:145            # write an empty plan146            file_object.writelines(f"\n\n<plan state=NONE>")147        else:148            # build plan content149            for step in diary_entry['my_plan']:150                start = '<'151                end = ' object at'152                step_string = str(step)...

tracker.py

Source:tracker.py

...39        self._sanitized_table = str.maketrans(" \\/\n", "----")40        if self.streaming:41            self._write_tap_version(self.stream)42            if self.plan is not None:43                self._write_plan(self.stream)44    def _get_outdir(self):45        return self._outdir46    def _set_outdir(self, outdir):47        self._outdir = outdir48        if outdir and not os.path.exists(outdir):49            os.makedirs(outdir)50    outdir = property(_get_outdir, _set_outdir)51    def _track(self, class_name):52        """Keep track of which test cases have executed."""53        if self._test_cases.get(class_name) is None:54            if self.streaming and self.header:55                self._write_test_case_header(class_name, self.stream)56            self._test_cases[class_name] = []57            if self.combined:58                self.combined_test_cases_seen.append(class_name)59    def add_ok(self, class_name, description, directive=""):60        result = Result(61            ok=True,62            number=self._get_next_line_number(class_name),63            description=description,64            directive=Directive(directive),65        )66        self._add_line(class_name, result)67    def add_not_ok(self, class_name, description, directive="", diagnostics=None):68        result = Result(69            ok=False,70            number=self._get_next_line_number(class_name),71            description=description,72            diagnostics=diagnostics,73            directive=Directive(directive),74        )75        self._add_line(class_name, result)76    def add_skip(self, class_name, description, reason):77        directive = "SKIP {0}".format(reason)78        result = Result(79            ok=True,80            number=self._get_next_line_number(class_name),81            description=description,82            directive=Directive(directive),83        )84        self._add_line(class_name, result)85    def _add_line(self, class_name, result):86        self._track(class_name)87        if self.streaming:88            print(result, file=self.stream)89        self._test_cases[class_name].append(result)90    def _get_next_line_number(self, class_name):91        if self.combined or self.streaming:92            # This has an obvious side effect. Oh well.93            self.combined_line_number += 194            return self.combined_line_number95        else:96            try:97                return len(self._test_cases[class_name]) + 198            except KeyError:99                # A result is created before the call to _track so the test100                # case may not be tracked yet. In that case, the line is 1.101                return 1102    def set_plan(self, total):103        """Notify the tracker how many total tests there will be."""104        self.plan = total105        if self.streaming:106            # This will only write the plan if we haven't written it107            # already but we want to check if we already wrote a108            # test out (in which case we can't just write the plan out109            # right here).110            if not self.combined_test_cases_seen:111                self._write_plan(self.stream)112        elif not self.combined:113            raise ValueError(114                "set_plan can only be used with combined or streaming output"115            )116    def generate_tap_reports(self):117        """Generate TAP reports.118        The results are either combined into a single output file or119        the output file name is generated from the test case.120        """121        if self.streaming:122            # We're streaming but set_plan wasn't called, so we can only123            # know the plan now (at the end).124            if not self._plan_written:125                print("1..{0}".format(self.combined_line_number), file=self.stream)126                self._plan_written = True127            return128        if self.combined:129            combined_file = "testresults.tap"130            if self.outdir:131                combined_file = os.path.join(self.outdir, combined_file)132            with open(combined_file, "w") as out_file:133                self._write_tap_version(out_file)134                if self.plan is not None:135                    print("1..{0}".format(self.plan), file=out_file)136                for test_case in self.combined_test_cases_seen:137                    self.generate_tap_report(138                        test_case, self._test_cases[test_case], out_file139                    )140                if self.plan is None:141                    print("1..{0}".format(self.combined_line_number), file=out_file)142        else:143            for test_case, tap_lines in self._test_cases.items():144                with open(self._get_tap_file_path(test_case), "w") as out_file:145                    self._write_tap_version(out_file)146                    self.generate_tap_report(test_case, tap_lines, out_file)147    def generate_tap_report(self, test_case, tap_lines, out_file):148        self._write_test_case_header(test_case, out_file)149        for tap_line in tap_lines:150            print(tap_line, file=out_file)151        # For combined results, the plan is only output once after152        # all the test cases complete.153        if not self.combined:154            print("1..{0}".format(len(tap_lines)), file=out_file)155    def _write_tap_version(self, filename):156        """Write a Version 13 TAP row.157        ``filename`` can be a filename or a stream.158        """159        if ENABLE_VERSION_13:160            print("TAP version 13", file=filename)161    def _write_plan(self, stream):162        """Write the plan line to the stream.163        If we have a plan and have not yet written it out, write it to164        the given stream.165        """166        if self.plan is not None:167            if not self._plan_written:168                print("1..{0}".format(self.plan), file=stream)169            self._plan_written = True170    def _write_test_case_header(self, test_case, stream):171        print("# TAP results for {test_case}".format(test_case=test_case), file=stream)172    def _get_tap_file_path(self, test_case):173        """Get the TAP output file path for the test case."""174        sanitized_test_case = test_case.translate(self._sanitized_table)175        tap_file = sanitized_test_case + ".tap"...

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.