KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use store_response method in lettuce-tools

Best Python code snippet using lettuce-tools_python

document_manager.py

Source:document_manager.py

1"""Module for Stackl Documents"""2import json3from typing import List4from loguru import logger5from pydantic import parse_obj_as6from core.enums.stackl_codes import StatusCode7from core.models.configs.document_model import BaseDocument8from core.models.configs.environment_model import Environment9from core.models.configs.functional_requirement_model import FunctionalRequirement10from core.models.configs.location_model import Location11from core.models.configs.policy_template_model import PolicyTemplate12from core.models.configs.stack_application_template_model import StackApplicationTemplate13from core.models.configs.stack_infrastructure_template_model import StackInfrastructureTemplate14from core.models.configs.zone_model import Zone15from core.models.history.snapshot_model import Snapshot16from core.models.items.service_model import Service17from core.models.items.stack_instance_model import StackInstance18from core.utils.stackl_exceptions import InvalidDocTypeError, InvalidDocNameError19from .manager import Manager20types_categories = ["configs", "items", "history"]21types_configs = [22    "environment", "location", "zone", "stack_application_template",23    "stack_infrastructure_template", "functional_requirement",24    "resource_requirement", "authentication", "policy_template"25]26types_items = [27    "stack_instance", "stack_template", "infrastructure_target", "service"28]29types_history = ["snapshot", "log"]30types = types_configs + types_items + types_history31def _process_document_keys(keys):32    """33    Method processes and checks document keys.34    Supports fuzzy get - trying to determine keys from other keys35    """36    # process to lowercase37    keys = {38        k.lower() if isinstance(k, str) else k:39            v.lower() if isinstance(v, str) else v40        for k, v in keys.items()41    }42    if not keys.get("category", None):43        type_name = keys.get("type", "none")44        if type_name in types_configs:45            keys["category"] = "configs"46        elif type_name in types_items:47            keys["category"] = "items"48        elif type_name in types_history:49            keys["category"] = "history"50        else:51            raise InvalidDocTypeError(type_name)52    if not keys.get("type", None):53        name = keys.get("name", "none")54        derived_type_from_name_list = [55            poss_type for poss_type in types if poss_type in name56        ]57        if len(derived_type_from_name_list) == 1:58            keys["type"] = derived_type_from_name_list[0]59        else:60            raise InvalidDocNameError(name)61    logger.debug(62        f"[DocumentManager] _process_document_keys. After _process_document_keys '{keys}'"63    )64    return keys65class DocumentManager(Manager):66    """Manager for everything document related"""67    def get_document(self, **keys):68        """Get a document from the chosen store"""69        logger.debug(f"[DocumentManager] get_document. Keys '{keys}'")70        keys = _process_document_keys(keys)71        logger.debug(72            f"[DocumentManager] get_document. Processed keys '{keys}'")73        store_response = self.store.get(**keys)74        if store_response.status_code == StatusCode.NOT_FOUND:75            return {}76        return store_response.content77    def write_document(self, document, overwrite=False):78        """Writes a document to the store"""79        logger.debug(80            f"[DocumentManager] write_document.  Document: '{document}'")81        keys = _process_document_keys(document)82        document['category'] = keys.get("category")83        document['type'] = keys.get("type")84        document['name'] = keys.get("name")85        document['description'] = keys.get("description")86        logger.debug("[DocumentManager] Checking if document already exists ")87        store_response = self.store.get(**keys)88        prev_document = store_response.content89        if store_response.status_code == StatusCode.NOT_FOUND:90            logger.debug(91                f" No document found yet. Creating document with data: {json.dumps(document)}"92            )93            store_response = self.store.put(document)94            return store_response.status_code95        if overwrite:96            prev_document_string = json.dumps(prev_document)97            logger.debug(98                f"Updating document with original contents: {prev_document_string}"99            )100            doc_new_string = json.dumps(document)101            logger.debug(102                f"Updating document with modified contents: {doc_new_string}"103            )104            if sorted(prev_document_string) == sorted(105                    doc_new_string106            ):  # Sorted since the doc might've changed the ordering107                logger.debug(108                    "Original document and new document are the same! NOT updating"109                )110                return StatusCode.OK111            store_response = self.store.put(document)112            return store_response.status_code113        logger.debug(114            "Document already exists and overwrite is false. Returning."115        )116        return StatusCode.BAD_REQUEST117    def write_base_document(self, base_document: BaseDocument):118        """Writes a base document to the store"""119        store_response = self.store.put(base_document.dict())120        return store_response.content121    def delete_base_document(self, document_type, name):122        """Deletes a base document"""123        store_response = self.store.delete(type=document_type,124                                           category="configs",125                                           name=name)126        return store_response127    def get_policy_template(self, policy_name):128        """gets a PolicyTemplate from the store"""129        store_response = self.store.get(type="policy_template",130                                        name=policy_name,131                                        category="configs")132        if store_response.status_code == 404:133            return None134        policy = PolicyTemplate.parse_obj(store_response.content)135        return policy136    def get_policy_templates(self):137        """gets all PolicyTemplate from the store"""138        store_response = self.store.get_all(document_type="policy_template",139                                            category="configs")140        policy_templates = parse_obj_as(List[PolicyTemplate],141                                        store_response.content)142        return policy_templates143    def write_policy_template(self, policy):144        """writes a PolicyTemplate to the store145        """146        store_response = self.store.put(policy.dict())147        policy = PolicyTemplate.parse_obj(store_response.content)148        return policy149    def delete_policy_template(self, name):150        """Deletes a PolicyTemplate from the store"""151        store_response = self.store.delete(type="policy_template",152                                           category="configs",153                                           name=name)154        return store_response155    def get_stack_instance(self, stack_instance_name):156        """gets a StackInstance Object from the store"""157        store_response = self.store.get(type="stack_instance",158                                        name=stack_instance_name,159                                        category="items")160        if store_response.status_code == 404:161            logger.debug("not found returning none")162            return None163        stack_instance = StackInstance.parse_obj(store_response.content)164        return stack_instance165    def get_stack_instances(self):166        """Get all stack instances"""167        store_response = self.store.get_all(document_type="stack_instance",168                                            category="items")169        stack_instances = parse_obj_as(List[StackInstance],170                                       store_response.content)171        if store_response.status_code == 404:172            return None173        return stack_instances174    def write_stack_instance(self, stack_instance):175        """writes a StackInstance object to the store176        """177        store_response = self.store.put(stack_instance.dict())178        return store_response.status_code179    def delete_stack_instance(self, name):180        """Delete a stack instance by name"""181        store_response = self.store.delete(type="stack_instance",182                                           name=name,183                                           category="items")184        return store_response185    def get_stack_infrastructure_template(self,186                                          stack_infrastructure_template_name):187        """gets a StackInfrastructureTemplate Object from the store"""188        store_response = self.store.get(189            type="stack_infrastructure_template",190            name=stack_infrastructure_template_name,191            category="configs")192        if store_response.status_code == 404:193            return None194        stack_infrastructure_template = StackInfrastructureTemplate.parse_obj(195            store_response.content)196        return stack_infrastructure_template197    def get_stack_infrastructure_templates(self):198        """Get a stack infrastructure template"""199        store_response = self.store.get_all(200            document_type="stack_infrastructure_template", category="configs")201        sits = parse_obj_as(List[StackInfrastructureTemplate],202                            store_response.content)203        return sits204    def write_stack_infrastructure_template(205            self, stack_infrastructure_template: StackInfrastructureTemplate):206        """writes a StackInfrastructureTemplate object to the store207        """208        store_response = self.store.put(stack_infrastructure_template.dict())209        return store_response.content210    def delete_stack_infrastructure_template(self, name):211        """Deletes a SIT from the store"""212        store_response = self.store.delete(213            type="stack_infrastructure_template",214            category="configs",215            name=name)216        return store_response217    def get_stack_application_template(218            self, stack_application_template_name) -> StackApplicationTemplate:219        """Gets a StackApplicationTemplate Object from the store"""220        store_response = self.store.get(type="stack_application_template",221                                        name=stack_application_template_name,222                                        category="configs")223        if store_response.status_code == 404:224            return None225        stack_application_template = StackApplicationTemplate.parse_obj(226            store_response.content)227        return stack_application_template228    def get_stack_application_templates(229            self) -> List[StackApplicationTemplate]:230        """Gets a Stack Application Template from the store"""231        store_response = self.store.get_all(document_type="stack_application_template",232                                            category="configs")233        sats = parse_obj_as(List[StackApplicationTemplate],234                            store_response.content)235        return sats236    def write_stack_application_template(237            self, stack_application_template: StackApplicationTemplate):238        """writes a StackApplicationTemplate object to the store239        """240        store_response = self.store.put(stack_application_template.dict())241        return store_response.content242    def delete_stack_application_template(self, name):243        """Deletes a stack application Template"""244        store_response = self.store.delete(type="stack_application_template",245                                           category="configs",246                                           name=name)247        return store_response248    def get_environment(self, environment_name):249        """gets a Environment Object from the store"""250        store_response = self.store.get(type="environment",251                                        name=environment_name,252                                        category="configs")253        if store_response.status_code == 404:254            return None255        environment = Environment.parse_obj(store_response.content)256        return environment257    def get_environments(self):258        """Get all environments from the store"""259        store_response = self.store.get_all(document_type="environment",260                                            category="configs")261        environments = parse_obj_as(List[Environment], store_response.content)262        return environments263    def get_location(self, location_name):264        """gets a Location Object from the store"""265        store_response = self.store.get(type="location",266                                        name=location_name,267                                        category="configs")268        if store_response.status_code == 404:269            return None270        location = Location.parse_obj(store_response.content)271        return location272    def get_locations(self):273        """Gets all locations from the store"""274        store_response = self.store.get_all(document_type="location",275                                            category="configs")276        locations = parse_obj_as(List[Location], store_response.content)277        return locations278    def get_zone(self, zone_name):279        """gets a Zone Object from the store"""280        store_response = self.store.get(type="zone",281                                        name=zone_name,282                                        category="configs")283        if store_response.status_code == 404:284            return None285        zone = Zone.parse_obj(store_response.content)286        return zone287    def get_zones(self):288        """Gets all zones from the store"""289        store_response = self.store.get_all(document_type="zone", category="configs")290        zone = parse_obj_as(List[Zone], store_response.content)291        return zone292    def get_service(self, service_name):293        """gets a Service Object from the store"""294        store_response = self.store.get(type="service",295                                        name=service_name,296                                        category="items")297        if store_response.status_code == 404:298            return None299        service = Service.parse_obj(store_response.content)300        return service301    def get_services(self):302        """Gets all the services from the store"""303        store_response = self.store.get_all(document_type="service", category="items")304        services = parse_obj_as(List[Service], store_response.content)305        return services306    def write_service(self, service: Service):307        """writes a Service object to the store308        """309        store_response = self.store.put(service.dict())310        return store_response.content311    def delete_service(self, name):312        """Deletes a service from the store"""313        self.store.delete(type="service", name=name, category="items")314    def get_functional_requirement(self, functional_requirement_name):315        """gets a FunctionalRequirement Object from the store"""316        store_response = self.store.get(type="functional_requirement",317                                        name=functional_requirement_name,318                                        category="configs")319        if store_response.status_code == 404:320            return None321        fr = FunctionalRequirement.parse_obj(store_response.content)322        return fr323    def get_functional_requirements(self) -> List[FunctionalRequirement]:324        """gets a FunctionalRequirement Object from the store"""325        store_response = self.store.get_all(document_type="functional_requirement",326                                            category="configs")327        fr = parse_obj_as(List[FunctionalRequirement], store_response.content)328        return fr329    def write_functional_requirement(self, functional_requirement: FunctionalRequirement):330        """writes a Service object to the store331        """332        store_response = self.store.put(functional_requirement.dict())333        return store_response.content334    def delete_functional_requirement(self, name: str):335        """Deletes a functional requirement from the store"""336        self.store.delete(type="functional_requirement",337                          name=name,338                          category="configs")339    def get_snapshot(self, name) -> Snapshot:340        """Get a snapshot from the store"""341        store_response = self.store.get(category="history",342                                        type="snapshot",343                                        name=name)344        if store_response.status_code == 404:345            return None346        snapshot = Snapshot.parse_obj(store_response.content)347        return snapshot348    def get_snapshots(self, document_type, name):349        """Gets all snapshots from the store of a document"""350        store_response = self.store.get_all(document_type=document_type,351                                            category="history",352                                            wildcard_prefix=f"{name}")353        return store_response.content354    def delete_snapshot(self, name: str):355        """Delete a snapshot from the store"""...

questionnaire.py

Source:questionnaire.py

...10    def show_question(self):11        print(self.question)1213    # åå¨é®å·æéççæ¡14    def store_response(self, new_response):15        # self.response[self.question] = new_response16        self.response.append(new_response)171819# è¯·å®ä¾å Survey() ç±»ï¼å¹¶ä¸æ¾ç¤ºåºè¿æ¬¡çè°æ¥é®å·é®é¢çº¦ 2 è¡ä»£ç 20food_survey = Survey('ä½ æåæ¬¢çç¾é£æ¯ä»ä¹ï¼')  # ç±»çå®ä¾åï¼åæ¶ä¸ºåæ°questionèµå¼ã21food_survey.show_question()  # è°ç¨ç±»æ¹æ³æå°é®é¢2223# åå¨é®å·è°æ¥ççæ¡24while True:25    response = input('è¯·åçé®å·é®é¢(ä»ä¹ä¹ä¸å¡«æåè½¦é®éåº)ï¼')26    if response == '':27        break28    # è¯·å°çæ¡ç¨é®å·ç³»ç»åå¨èµ·æ¥,çº¦ 1 è¡ä»£ç ï¼åéåè§ä¸æ¹ã29    food_survey.store_response(response)3031# è¾åºæµè¯32for food in food_survey.response:33    print('ç¾é£ï¼' + food)3435print('-------------------------------------------æ¶éåååç±è´¯å°----------------------------------------')363738# åå»ºä¸ä¸ªé®å·è°æ¥ç±»39class Survey:40    # æ¶éè°æ¥é®å·ççæ¡41    def __init__(self, question):42        self.question = question43        # å®ä¹æ¶éé®å·é®é¢çæ¡çåè¡¨44        self.response = []4546    # æ¾ç¤ºè°æ¥é®å·çé¢ç®47    def show_question(self):48        print(self.question)4950    # åå¨é®å·æéççæ¡51    def store_response(self, new_response):52        # self.response[self.question] = new_response53        self.response.append(new_response)545556class Inquiry(Survey):57    # æ¶éè°æ¥é®å·ççæ¡58    def __init__(self, question):59        # super().__init__(question)60        Survey.__init__(self, question)61        # å®ä¹æ¶éé®å·çæ¡çåéï¼ä»£ç é1è¡62        self.response = {}6364    # åå¨é®å·æéççæ¡ï¼è¦çç¶ç±»çç±»æ¹æ³ï¼65    def store_response(self, name, new_response):  # é¤äº selfï¼è¿éè¦ä¸¤ä¸ªåæ°ã66        self.response[name] = new_response  # é®å¼å¯¹çæ°å¢676869survey = Inquiry('ä½ çç±è´¯å°æ¯åªï¼')70survey.show_question()71while True:72    response = input('è¯·åçé®å·é®é¢ï¼æ q é®éåºï¼')73    if response == 'q':74        break75    name = input('è¯·è¾å¥åçèå§åï¼')76    survey.store_response(name, response)  # è°ç¨ç±»æ¹æ³ï¼å°ä¸¤æ¬¡éè¿ input çåç¬¦ä¸²åå¥åå¸ã7778# è¾åºæµè¯79for name, value in survey.response.items():80    print(name + 'ï¼' + value)8182print("----------------------------------------------é£ä¸ªç·äººæ¥äº------------------------------------------")838485class Person:86    def __init__(self):87        self.name = input('è¯·è¾å¥æ¨çå§åï¼')88        print('å¤§å®¶æ³¨æäºï¼')8990    def show(self):
...

57.py

Source:57.py

1from survey import AnonymousSurvey2mine=AnonymousSurvey('who you like to meet')3mine.show_question()4mine.store_response('2')5mine.store_response('2')6mine.store_response('qi')7mine.store_response('qi')8mine.store_response('qi')9mine.store_response('qi')10mine.store_response('qi')11mine.store_response('qi')12mine.store_response('qi')...

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.