How to use get_headers method in tempest

Best Python code snippet using tempest_python

store.py

Source:store.py

...17        content = response.json()18        os.environ["ACCESS_TOKEN"] = content["access_token"]19        os.environ["EXPIRES"] = str(content["expires"])20    return os.environ["ACCESS_TOKEN"]21def get_headers():22    access_token = get_access_token()23    return {24        "Authorization": f"Bearer {access_token}",25        "Content-Type": "application/json",26    }27def create_product(product):28    payload = {"type": "product", "status": "live", **product}29    response = requests.post(30        f"{BASE_URL}/products", headers=get_headers(), json={"data": payload}31    )32    response.raise_for_status()33    return response.json()["data"]["id"]34def set_main_image(product_id, image_id):35    payload = {"type": "main_image", "id": image_id}36    response = requests.post(37        f"{BASE_URL}/products/{product_id}/relationships/main-image",38        headers=get_headers(),39        json={"data": payload},40    )41    response.raise_for_status()42def get_products():43    response = requests.get(f"{BASE_URL}/products", headers=get_headers())44    response.raise_for_status()45    return response.json()["data"]46def get_product(item_id):47    response = requests.get(f"{BASE_URL}/products/{item_id}", headers=get_headers())48    response.raise_for_status()49    return response.json()["data"]50def get_cart(cart_id):51    response = requests.get(f"{BASE_URL}/carts/{cart_id}", headers=get_headers())52    response.raise_for_status()53    return response.json()54def delete_cart(cart_id):55    response = requests.delete(f"{BASE_URL}/carts/{cart_id}", headers=get_headers())56    response.raise_for_status()57def add_to_cart(cart, item_id, quantity):58    headers = {**get_headers(), "X-MOLTIN-CURRENCY": "RUB"}59    payload = {"id": item_id, "type": "cart_item", "quantity": int(quantity)}60    response = requests.post(61        f"{BASE_URL}/carts/{cart}/items", headers=headers, json={"data": payload}62    )63    response.raise_for_status()64def update_cart_item(cart, item_id, quantity):65    payload = {"quantity": int(quantity)}66    response = requests.put(67        f"{BASE_URL}/carts/{cart}/items/{item_id}",68        headers=get_headers(),69        json={"data": payload},70    )71    response.raise_for_status()72def remove_from_cart(cart, item_id):73    response = requests.delete(74        f"{BASE_URL}/carts/{cart}/items/{item_id}", headers=get_headers()75    )76    response.raise_for_status()77def get_cart_items(cart):78    response = requests.get(f"{BASE_URL}/carts/{cart}/items", headers=get_headers())79    response.raise_for_status()80    return response.json()81def create_file(file_location):82    access_token = get_access_token()83    headers = {"Authorization": f"Bearer {access_token}"}84    files = {"file_location": (None, file_location)}85    response = requests.post(f"{BASE_URL}/files", headers=headers, files=files)86    response.raise_for_status()87    return response.json()["data"]["id"]88def get_file_link(file_id):89    response = requests.get(f"{BASE_URL}/files/{file_id}", headers=get_headers())90    response.raise_for_status()91    return response.json()["data"]["link"]["href"]92def create_customer(name, email):93    payload = {"type": "customer", "name": name, "email": email}94    response = requests.post(95        f"{BASE_URL}/customers", headers=get_headers(), json={"data": payload}96    )97    response.raise_for_status()98    return response.json()["data"]["id"]99def find_customer(email):100    payload = {"filter": f"eq(email,{email})"}101    response = requests.get(102        f"{BASE_URL}/customers", headers=get_headers(), params=payload103    )104    response.raise_for_status()105    return response.json()["data"]106def get_customers():107    response = requests.get(f"{BASE_URL}/customers", headers=get_headers())108    response.raise_for_status()109    return response.json()["data"]110def create_flow(entity):111    payload = {"type": "flow", "enabled": True, **entity}112    response = requests.post(113        f"{BASE_URL}/flows", headers=get_headers(), json={"data": payload}114    )115    response.raise_for_status()116    return response.json()["data"]["id"]117def create_field(field):118    payload = {"type": "field", "required": True, "enabled": True, **field}119    response = requests.post(120        f"{BASE_URL}/fields", headers=get_headers(), json={"data": payload}121    )122    response.raise_for_status()123def create_entry(flow_slug, entry):124    payload = {"type": "entry", **entry}125    response = requests.post(126        f"{BASE_URL}/flows/{flow_slug}/entries",127        headers=get_headers(),128        json={"data": payload},129    )130    response.raise_for_status()131    return response.json()["data"]["id"]132def update_entry(flow_slug, entry_id, field_slug, value):133    payload = {134        "type": "entry",135        "id": entry_id,136        field_slug: value,137    }138    response = requests.put(139        f"{BASE_URL}/flows/{flow_slug}/entries/{entry_id}",140        headers=get_headers(),141        json={"data": payload},142    )143    response.raise_for_status()144def get_entries(slug):145    response = requests.get(f"{BASE_URL}/flows/{slug}/entries", headers=get_headers())146    response.raise_for_status()147    return response.json()["data"]148def get_entry(slug, entry_id):149    response = requests.get(150        f"{BASE_URL}/flows/{slug}/entries/{entry_id}", headers=get_headers()151    )152    response.raise_for_status()153    return response.json()["data"]154def delete_entry(slug, entry_id):155    response = requests.delete(156        f"{BASE_URL}/flows/{slug}/entries/{entry_id}", headers=get_headers()157    )...

relationship.py

Source:relationship.py

...47    ref_state,48)49def get_relationship() -> dict[str, list[TableField]]:50    return {51        "agencies.csv": agencies.get_headers(),52        "NIBRS_ACTIVITY_TYPE.csv": nibrs_activity_type.get_headers(),53        "NIBRS_AGE.csv": nibrs_age.get_headers(),54        "NIBRS_ARREST_TYPE.csv": nibrs_arrest_type.get_headers(),55        "NIBRS_ARRESTEE.csv": nibrs_arrestee.get_headers(),56        "NIBRS_ARRESTEE_WEAPON.csv": nibrs_arrestee_weapon.get_headers(),57        "NIBRS_ASSIGNMENT_TYPE.csv": nibrs_assignment_type.get_headers(),58        "NIBRS_BIAS_LIST.csv": nibrs_bias_list.get_headers(),59        "NIBRS_BIAS_MOTIVATION.csv": nibrs_bias_motivation.get_headers(),60        "NIBRS_CIRCUMSTANCES.csv": nibrs_circumstances.get_headers(),61        "NIBRS_CLEARED_EXCEPT.csv": nibrs_cleared_except.get_headers(),62        "NIBRS_CRIMINAL_ACT.csv": nibrs_criminal_act.get_headers(),63        "NIBRS_CRIMINAL_ACT_TYPE.csv": nibrs_criminal_act_type.get_headers(),64        "NIBRS_DRUG_MEASURE_TYPE.csv": nibrs_drug_measure_type.get_headers(),65        "NIBRS_ETHNICITY.csv": nibrs_ethnicity.get_headers(),66        "NIBRS_incident.csv": nibrs_incident.get_headers(),67        "NIBRS_INJURY.csv": nibrs_injury.get_headers(),68        "NIBRS_JUSTIFIABLE_FORCE.csv": nibrs_justifiable_force.get_headers(),69        "NIBRS_LOCATION_TYPE.csv": nibrs_location_type.get_headers(),70        "NIBRS_month.csv": nibrs_month.get_headers(),71        "NIBRS_OFFENDER.csv": nibrs_offender.get_headers(),72        "NIBRS_OFFENSE.csv": nibrs_offense.get_headers(),73        "NIBRS_OFFENSE_TYPE.csv": nibrs_offense_type.get_headers(),74        "NIBRS_PROP_DESC_TYPE.csv": nibrs_prop_desc_type.get_headers(),75        "NIBRS_PROP_LOSS_TYPE.csv": nibrs_prop_loss_type.get_headers(),76        "NIBRS_PROPERTY.csv": nibrs_property.get_headers(),77        "NIBRS_PROPERTY_DESC.csv": nibrs_property_desc.get_headers(),78        "NIBRS_RELATIONSHIP.csv": nibrs_relationship.get_headers(),79        "NIBRS_SUSPECT_USING.csv": nibrs_suspect_using.get_headers(),80        "NIBRS_SUSPECTED_DRUG.csv": nibrs_suspected_drug.get_headers(),81        "NIBRS_SUSPECTED_DRUG_TYPE.csv": nibrs_suspected_drug_type.get_headers(),82        "NIBRS_USING_LIST.csv": nibrs_using_list.get_headers(),83        "NIBRS_VICTIM.csv": nibrs_victim.get_headers(),84        "NIBRS_VICTIM_CIRCUMSTANCES.csv": nibrs_victim_circumstances.get_headers(),85        "NIBRS_VICTIM_INJURY.csv": nibrs_victim_injury.get_headers(),86        "NIBRS_VICTIM_OFFENDER_REL.csv": nibrs_victim_offender_rel.get_headers(),87        "NIBRS_VICTIM_OFFENSE.csv": nibrs_victim_offense.get_headers(),88        "NIBRS_VICTIM_TYPE.csv": nibrs_victim_type.get_headers(),89        "NIBRS_WEAPON.csv": nibrs_weapon.get_headers(),90        "NIBRS_WEAPON_TYPE.csv": nibrs_weapon_type.get_headers(),91        "REF_RACE.csv": ref_race.get_headers(),92        "REF_STATE.csv": ref_state.get_headers(),...

headerdata.py

Source:headerdata.py

...4        """Creates HeaderData with input dictionary of headers."""5        if header_data is None:6            header_data = {}7        self.header_data = header_data8    def get_headers(self):9        return self.header_data10    def update_headers(self, new_headers: dict):11        self.header_data.update(new_headers)12    def remove_header(self, key: str):13        try:14            del self.header_data[key]15        except KeyError:16            pass17    def get_archive_frame_data(self):18        return {19            'reduction_level': self.get_reduction_level(),20            'observation_day': self.get_observation_day(),21            'observation_date': self.get_observation_date(),22            'proposal_id': self.get_proposal_id(),23            'instrument_id': self.get_instrument_id(),24            'target_name': self.get_target_name(),25            'site_id': self.get_site_id(),26            'telescope_id': self.get_telescope_id(),27            'exposure_time': self.get_exposure_time(),28            'primary_optical_element': self.get_primary_optical_element(),29            'public_date': self.get_public_date(),30            'configuration_type': self.get_configuration_type(),31            'observation_id': self.get_observation_id(),32            'request_id': self.get_request_id(),33            'related_frame_filenames': list(self.get_related_frames().values())34        }35    def headers_are_set(self, keys: list):36        """Check that the values for the provided headers are set."""37        headers = self.get_headers()38        empty_values = [None, '']39        values = [headers.get(key) for key in keys]40        return all([value not in empty_values for value in values])41    def get_related_frame_keys(self):42        return settings.RELATED_FRAME_KEYS43    def get_related_frames(self):44        headers = self.get_headers()45        keys = self.get_related_frame_keys()46        related_frames = {}47        for key in keys:48            if key in headers and headers[key]:49                related_frames[key] = headers[key]50        return related_frames51    def get_observation_day(self):52        return self.get_headers().get(settings.OBSERVATION_DAY_KEY)53    def get_observation_date(self):54        return self.get_headers().get(settings.OBSERVATION_DATE_KEY)55    def get_proposal_id(self):56        return self.get_headers().get(settings.PROPOSAL_ID_KEY, '')57    def get_configuration_type(self):58        return self.get_headers().get(settings.CONFIGURATION_TYPE_KEY, '')59    def get_exposure_time(self):60        return self.get_headers().get(settings.EXPOSURE_TIME_KEY)61    62    def get_public_date(self):63        return self.get_headers().get(settings.PUBLIC_DATE_KEY)64    def get_reduction_level(self):65        return self.get_headers().get(settings.REDUCTION_LEVEL_KEY, 0)66    def get_instrument_id(self):67        return self.get_headers().get(settings.INSTRUMENT_ID_KEY, '')68    def get_site_id(self):69        return self.get_headers().get(settings.SITE_ID_KEY, '')70    def get_primary_optical_element(self):71        return self.get_headers().get(settings.PRIMARY_OPTICAL_ELEMENT_KEY, '')72    def get_target_name(self):73        return self.get_headers().get(settings.TARGET_NAME_KEY, '')74    def get_telescope_id(self):75        return self.get_headers().get(settings.TELESCOPE_ID_KEY, '')76    def get_observation_id(self):77        return self.get_headers().get(settings.OBSERVATION_ID_KEY)78    def get_configuration_id(self):79        return self.get_headers().get(settings.CONFIGURATION_ID_KEY)80    def get_request_id(self):81        return self.get_headers().get(settings.REQUEST_ID_KEY)82    def get_requestgroup_id(self):...

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.