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

How to use put_inventory method in localstack

Best Python code snippet using localstack_python

compliance_uploader.py

Source:compliance_uploader.py

1from botocore.exceptions import ClientError2import logging3import os4from patch_common import client_selector5from patch_common.constant_repository import ExitCodes6from patch_common.exceptions import PatchManagerError7logger = logging.getLogger()8DEFAULT_HASH_LOCATION = "/var/log/amazon/ssm/pb-hashes"9def upload_report(region,10                  instance_id,11                  compliance):12    """13    Upload an inventory report for the instance containing the summary and compliance items. Handles inventory14    hash change logic15    :param ssm_client_selector: to hit inventory16    :param instance_id: for the items17    :param summary_item: to be uploaded with put_inventory18    :param compliance_item: to be uploaded with put_inventory, empty if hash has not changed19    """20    logger.info("Start to upload patch compliance.")21    logger.debug("Compliance: %s", str(compliance))22    compliance_contents = compliance["Items"]23    clobber = False24    retries = 025    success = False26    while retries <= 1 and not success:27        retries = retries + 128        if clobber or _has_compliance_hash_changed(compliance):29            logger.info("Attempting full upload")30            compliance["Items"] = compliance_contents31        else:32            logger.info("Report is unchanged, attempting partial upload")33            del (compliance["Items"])34        try:35            default_ssm_client = client_selector.get_default_client(instance_id, region)36            items = []37            if "Items" in compliance:38                items = compliance["Items"]39            default_ssm_client.put_compliance_items(40                ResourceId=compliance["ResourceId"],41                ResourceType=compliance["ResourceType"],42                ComplianceType=compliance["ComplianceType"],43                ExecutionSummary=compliance["ExecutionSummary"],44                Items=items,45                UploadType="PARTIAL",46                ItemContentHash=compliance["ItemContentHash"]47            )48            logger.info("Upload complete.")49            success = True50            _persist_inventory_hash(compliance)51        except (ClientError, PatchManagerError) as e:52            # Only worry about a specific service error code53            # TODO - check these Exception types with put compliance.54            if isinstance(e, ClientError) and e.response["Error"]["Code"] in ["InvalidItemContentException", "ItemContentMismatchException"]:55                logger.warn("Hash mismatch on upload, retry", exc_info=True)56                clobber = True57            elif not client_selector.is_managed_instance(instance_id):58                _upload_report_with_fallback_client(region, instance_id, compliance)59                return60            else:61                raise PatchManagerError("Unable to upload the inventory:", ExitCodes.PUT_INVENTORY_ERROR, e)62        except Exception as e:63            raise PatchManagerError("Encounter service side error when uploading the inventory", ExitCodes.PUT_INVENTORY_ERROR, e)64    if not success:65        raise PatchManagerError("Could not upload inventory report after retries.", ExitCodes.PUT_INVENTORY_ERROR)66def _upload_report_with_fallback_client(region,67                                   instance_id,68                                   compliance):69    """70    Upload an inventory report using fallback ssm client71    :param instance_id: for the items72    :param summary_item: to be uploaded with put_inventory73    :param compliance_item: to be uploaded with put_inventory, empty if hash has not changed74    """75    fallback_ssm_client = client_selector.get_fallback_client(region)76    try:77        items = []78        if "Items" in compliance:79            items = compliance["Items"]80        fallback_ssm_client.put_compliance_items(81            ResourceId=compliance["ResourceId"],82            ResourceType=compliance["ResourceType"],83            ComplianceType=compliance["ComplianceType"],84            ExecutionSummary=compliance["ExecutionSummary"],85            Items=items,86            ItemContentHash=compliance["ItemContentHash"]87        )88        logger.info("Upload complete.")89        _persist_inventory_hash(compliance)90    except Exception as e:91        # TODO - check these errors.92        raise PatchManagerError("Unable to upload the inventory using fallback creds:", ExitCodes.PUT_INVENTORY_ERROR, e)93def _persist_inventory_hash(item):94    """95    Persists the inventory hash for future comparison, cleans previous hashes96    :param item: to get ContentHash from97    """98    try:99        if os.path.exists(DEFAULT_HASH_LOCATION):100            hashes = os.listdir(DEFAULT_HASH_LOCATION)101            if len(hashes) > 1:102                logger.warn("Found multiple old hashes in directory %s", hashes)103            for directory in hashes:104                os.rmdir(os.path.join(DEFAULT_HASH_LOCATION, directory))105        os.makedirs(os.path.join(DEFAULT_HASH_LOCATION, item['ItemContentHash']))106    except:107        logger.warn("Could not persist inventory hash", exc_info=True)108def _has_compliance_hash_changed(compliance):109    """110    checks the previous hash uploaded, if any111    :param compliance: to get ContentHash from112    :return:113    """114    new_hash = compliance['ItemContentHash']115    hash_path = os.path.join(DEFAULT_HASH_LOCATION, new_hash)...

inventory_uploader.py

Source:inventory_uploader.py

...34            logger.info("Report is unchanged, attempting partial upload")35            del (compliance_item["Content"])36        try:37            default_ssm_client = client_selector.get_default_client(instance_id, region)38            default_ssm_client.put_inventory(39                InstanceId=instance_id,40                Items=[summary_item, compliance_item]41            )42            logger.info("Upload complete.")43            success = True44            _persist_inventory_hash(compliance_item)45        except (ClientError, PatchManagerError) as e:46            # Only worry about a specific service error code47            if isinstance(e, ClientError) and e.response["Error"]["Code"] in ["InvalidItemContentException", "ItemContentMismatchException"]:48                logger.warn("Hash mismatch on upload, retry", exc_info=True)49                clobber = True50            elif isinstance(e, ClientError) and e.response["Error"]["Code"] == "ItemSizeLimitExceededException":51                logger.warn("Exceeded Inventory Size Limit, retry without uploading CVE data")52                compliance_contents = _remove_cves_from_report(compliance_contents)53            elif not client_selector.is_managed_instance(instance_id):54                _upload_report_with_fallback_client(region, instance_id, summary_item, compliance_item)55                return56            else:57                raise PatchManagerError("Unable to upload the inventory:", ExitCodes.PUT_INVENTORY_ERROR, e)58        except Exception as e:59            raise PatchManagerError("Encounter service side error when uploading the inventory", ExitCodes.PUT_INVENTORY_ERROR, e)60    if not success:61        raise PatchManagerError("Could not upload inventory report after retries.", ExitCodes.PUT_INVENTORY_ERROR)62def _upload_report_with_fallback_client(region,63                                   instance_id,64                                   summary_item,65                                   compliance_item):66    """67    Upload an inventory report using fallback ssm client68    :param instance_id: for the items69    :param summary_item: to be uploaded with put_inventory70    :param compliance_item: to be uploaded with put_inventory, empty if hash has not changed71    """72    fallback_ssm_client = client_selector.get_fallback_client(region)73    try:74        fallback_ssm_client.put_inventory(75            InstanceId=instance_id,76            Items=[summary_item, compliance_item]77        )78        logger.info("Upload complete.")79        _persist_inventory_hash(compliance_item)80    except Exception as e:81        raise PatchManagerError("Unable to upload the inventory using fallback creds:", ExitCodes.PUT_INVENTORY_ERROR, e)82def _persist_inventory_hash(item):83    """84    Persists the inventory hash for future comparison, cleans previous hashes85    :param item: to get ContentHash from86    """87    try:88        if os.path.exists(DEFAULT_HASH_LOCATION):...

main.py

Source:main.py

...62    with open("config.json", "r") as file:63        config = json.load(file)64    modifiedInv = FileUtils.modifyInventory(inv.json(), config)65    time.sleep(25)66    val.put_inventory(modifiedInv)6768    while True:69        FileUtils.InventoryLoop()70        inv = val.get_inventory()71        with open("config.json", "r") as file:72            config = json.load(file)73        modifiedInv = FileUtils.modifyInventory(inv.json(), config)74        val.put_inventory(modifiedInv)757677787980if __name__ == '__main__':
...

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.