python类Settings()的实例源码

def report_status(my_settings: settings.Settings, req_type: str, reason: str, logical_res_id: str, status: str):
    raw_message = get_messages(my_settings)
    if len(raw_message) < 1:
        print("No message on queue... so we can't report back")
        return 1
    messages = [parse_message(m) for m in raw_message]
    filtered_messages = [m for m in messages if m.get('RequestType') == req_type]
    if logical_res_id != "":
        filtered_messages = [m for m in messages if m.get('LogicalResourceId') == logical_res_id]

    if len(filtered_messages) < 1:
        print("No message of type '{}', so unable to report back to CloudFormation".format(req_type))
        return 1

    for message in filtered_messages:
        response_for_cloud_formation = build_payload(message, status, reason)
        response_url_full = message.get('ResponseURL')
        response_url, response_params = response_url_full.split('?')
        json_response = json.dumps(response_for_cloud_formation)
        print("Posting SQS response\n{}".format(json_response))
        response = requests.put(
            url=response_url, params=response_params,
            data=str.encode(json_response)
        )
        if response.status_code != 200:
            print("Failed posting message {}".format(response))
            print("{}\n".format(response.text))

            return 1

        print("Deleting message", message.get('ReceiptHandle'))
        delete_messages(my_settings, message)

    return 0

def get_messages(my_settings: settings.Settings):
    sqs = boto3.client(service_name='sqs', region_name=my_settings.region)
    response = sqs.receive_message(
        QueueUrl=my_settings.pcf_pcfcustomresourcesqsqueueurl,
        MaxNumberOfMessages=10,
        VisibilityTimeout=1
    )
    print("SQS Messages")
    print(json.dumps(response, indent="  "))

    return response.get('Messages', [])

def configure_ert(my_settings: Settings):
    out, err, exit_code = om_manager.stage_product("cf", my_settings)
    if exit_code != 0:
        print("Failed to stage ERT")
        return out, err, exit_code

    out, err, exit_code = configure_tile_az(my_settings, 'cf')
    if exit_code != 0:
        print("Failed to configure az ERT")
        return out, err, exit_code

    out, err, exit_code = configure_ert_config(my_settings)
    if exit_code != 0:
        print("Failed to configure ERT")
        return out, err, exit_code

    out, err, exit_code = modify_vm_types(my_settings)
    if exit_code != 0:
        print("Failed to modify VM types for ERT")
        return out, err, exit_code

    out, err, exit_code = configure_ert_resources(my_settings)
    if exit_code != 0:
        print("Failed to configure ERT")
        return out, err, exit_code

    out, err, exit_code = configure_ert_multiaz_resources(my_settings)
    if exit_code != 0:
        print("Failed to configure Multi AZ ERT")
        return out, err, exit_code

    return create_required_databases(my_settings)

def configure_ert_resources(my_settings: Settings):
    prefix = my_settings.stack_name
    if my_settings.pcf_input_elbprefix != "":
        prefix = my_settings.pcf_input_elbprefix
    ert_resource_ctx = {
        "router_lb_name": "{prefix}".format(prefix=prefix)
    }
    with open("templates/ert_resources_config.j2.json", 'r') as f:
        ert_resource_template = Template(f.read())
    ert_resource_config = om_manager.format_om_json_str(ert_resource_template.render(ert_resource_ctx))
    cmd = "{om_with_auth} configure-product -n cf -pr '{ert_resources}'".format(
        om_with_auth=om_manager.get_om_with_auth(my_settings),
        ert_resources=ert_resource_config
    )
    return util.exponential_backoff_cmd(cmd)

def configure_ert_multiaz_resources(my_settings: Settings):
    if my_settings.pcf_pcfdeploymentsize == "Multi-AZ":
        with open("templates/ert_multiaz_resources_config.j2.json", 'r') as f:
            template = f.read()
    else:
        with open("templates/ert_singleaz_resources_config.j2.json", 'r') as f:
            template = f.read()

    ert_resource_config = om_manager.format_om_json_str(template)
    cmd = "{om_with_auth} configure-product -n cf -pr '{ert_resources}'".format(
        om_with_auth=om_manager.get_om_with_auth(my_settings),
        ert_resources=ert_resource_config
    )
    return util.exponential_backoff_cmd(cmd)

def configure_ert_config(my_settings: Settings):
    cert, key = generate_ssl_cert(my_settings)
    ert_config_template_ctx = {
        "pcf_rds_address": my_settings.pcf_rdsaddress,
        "pcf_rds_username": my_settings.pcf_rdsusername,
        "dns_suffix": my_settings.pcf_input_domain,
        "pcf_rds_password": my_settings.pcf_rdspassword,
        "admin_email": my_settings.pcf_input_adminemail,
        "pcf_elastic_runtime_s3_buildpacks_bucket": my_settings.pcf_elasticruntimes3buildpacksbucket,
        "pcf_elastic_runtime_s3_droplets_bucket": my_settings.pcf_elasticruntimes3dropletsbucket,
        "pcf_elastic_runtime_s3_packages_bucket": my_settings.pcf_elasticruntimes3packagesbucket,
        "pcf_elastic_runtime_s3_resources_bucket": my_settings.pcf_elasticruntimes3resourcesbucket,
        "pcf_iam_access_key_id": my_settings.pcf_iamuseraccesskey,
        "pcf_iam_secret_access_key": my_settings.pcf_iamusersecretaccesskey,
        "pcf_companyname": my_settings.pcf_pcfcompanyname,
        "s3_endpoint": my_settings.get_s3_endpoint(),
        "s3_region": my_settings.region,
        "pcf_skipsslvalidation": my_settings.pcf_input_skipsslvalidation,
        "cert": cert.replace("\n", "\\n"),
        "key": key.replace("\n", "\\n")
    }
    with open("templates/ert_config.j2.json", 'r') as f:
        ert_template = Template(f.read())
    ert_config = om_manager.format_om_json_str(ert_template.render(ert_config_template_ctx))
    cmd = "{om_with_auth} configure-product -n cf -p '{ert_config}'".format(
        om_with_auth=om_manager.get_om_with_auth(my_settings),
        ert_config=ert_config
    )
    return util.exponential_backoff_cmd(cmd)

def create_required_databases(my_settings: Settings):
    cmd = "mysql -h {hostname} --user={username} --port={port} --password={password} < templates/required_dbs.sql".format(
        hostname=my_settings.pcf_rdsaddress,
        username=my_settings.pcf_rdsusername,
        port=my_settings.pcf_rdsport,
        password=my_settings.pcf_rdspassword
    )

    return util.exponential_backoff_cmd(cmd)

def modify_vm_types(my_settings: Settings):
    path = '/api/v0/vm_types'
    out, err, return_code = om_manager.curl_get(my_settings, path)
    if return_code != 0:
        if out != "":
            print(out)
        if err != "":
            print(err)
        return out, err, return_code

    response_json = json.loads(out)
    m4_exists = False

    for vm_type in response_json["vm_types"]:
        if vm_type["name"].startswith("m3"):
            response_json["vm_types"].remove(vm_type)
        elif vm_type["name"].startswith("m4"):
            m4_exists = True

    if not m4_exists:
        with open("templates/ert_vm_types.json") as template:
            additional_types = json.load(template)
            for a in additional_types:
                response_json["vm_types"].append(a)

    out, err, return_code = om_manager.curl_payload(my_settings, path, json.dumps(response_json), 'PUT')
    if return_code != 0:
        if out != "":
            print(out)
        if err != "":
            print(err)

    return out, err, return_code

def generate_ssl_cert(my_settings: Settings):
    call("scripts/gen_ssl_certs.sh {}".format(my_settings.pcf_input_domain), shell=True)
    with open("{}.crt".format(my_settings.pcf_input_domain), 'r') as cert_file:
        cert = cert_file.read()

    with open("{}.key".format(my_settings.pcf_input_domain), 'r') as key_file:
        key = key_file.read()

    return cert, key

def setUp(self):
        self.settings = Mock(settings.Settings)
        self.settings.ert_release_version = "1.9.0"
        self.settings.ert_release_sha256 = "xyz123"
        self.settings.stemcell_release_version = "123"
        self.settings.stemcell_release_sha256 = "123"

def upload_stemcell(my_settings: settings.Settings, path: str):
    for stemcell in os.listdir(path):
        if stemcell.endswith(".tgz"):
            print("uploading stemcell {0}".format(stemcell))
            cmd = "{om_with_auth} upload-stemcell -s '{path}'".format(
                om_with_auth=om_manager.get_om_with_auth(my_settings), path=os.path.join(path, stemcell)
            )
            out, err, exit_code = util.exponential_backoff_cmd(cmd)
            if exit_code != 0:
                return out, err, exit_code

    return "", "", 0

def upload_assets(my_settings: settings.Settings, path: str):
    for tile in os.listdir(path):
        if tile.endswith(".pivotal"):
            print("uploading product {0}".format(tile))

            cmd = "{om_with_auth} -r 3600 upload-product -p '{path}'".format(
                om_with_auth=om_manager.get_om_with_auth(my_settings), path=os.path.join(path, tile))

            out, err, exit_code = util.exponential_backoff_cmd(cmd)
            if exit_code != 0:
                return out, err, exit_code

    return "", "", 0

def setUp(self):
        self.settings = Mock(Settings)
        self.settings.region = 'canada-1a'
        self.settings.pcf_pcfwaithandle = 'https://wait.example.com?signature=foo'

def setUp(self):
        self.settings = Mock(Settings)
        self.settings.opsman_url = 'https://cf.example.com'
        self.settings.opsman_user = 'admin'
        self.settings.pcf_opsmanageradminpassword = 'monkey-123'

def setUp(self):
        self.settings = Mock(Settings)
        self.settings.region = 'canada-1a'
        self.settings.pcf_elasticruntimes3buildpacksbucket = 'bucket-bp'
        self.settings.pcf_elasticruntimes3dropletsbucket = 'bucket-dp'
        self.settings.pcf_elasticruntimes3packagesbucket = 'bucket-pkg'
        self.settings.pcf_elasticruntimes3resourcesbucket = 'bucket-rsc'
        self.settings.pcf_iam_access_key_id = 'key-vale'
        self.settings.pcf_iam_secret_access_key = 'key-secret'
        self.settings.resources_created = True

def setUp(self):
        self.settings = Mock(Settings)
        self.settings.region = 'canada-1a'
        self.settings.pcf_pcfcustomresourcesqsqueueurl = 'https://queue.example.com'

        self.response = {'Messages':
            [
                {
                    'Body': '{\n  "Type" : "Notification",\n  "MessageId" : "3c71d3a0-7d25-59b1-93bc-8213196855a7",\n  "TopicArn" : "arn:aws:sns:us-west-2:540420658117:custom-resource-maybe-4-SNSTopic-11NVYUIXWZTXZ",\n  "Subject" : "AWS CloudFormation custom resource request",\n  "Message" : "{\\"RequestType\\":\\"Delete\\",\\"ServiceToken\\":\\"arn:aws:sns:us-west-2:540420658117:custom-resource-maybe-4-SNSTopic-11NVYUIXWZTXZ\\",\\"ResponseURL\\":\\"https://cloudformation-custom-resource-response-uswest2.s3-us-west-2.amazonaws.com/arn%3Aaws%3Acloudformation%3Aus-west-2%3A530420658117%3Astack/pcf-stack/1e820540-4c58-11e7-a965-50d5ca0184f2%7CMyCustomResource%7C4dd2c9a0-04cb-4218-908c-e3cdfad3c634?AWSAccessKeyId=AKIAI4KYMPPRGQACET5Q&Expires=1496940077&Signature=uLbeSl3rtkuDa2xf0y9oMFc1zBI%3D\\",\\"StackId\\":\\"arn:aws:cloudformation:us-west-2:540420658117:stack/pcf-stack/1e820540-4c58-11e7-a965-50d5ca0184f2\\",\\"RequestId\\":\\"4dd2c9a0-04cb-4218-908c-e2cdfad3c634\\",\\"LogicalResourceId\\":\\"MyCustomResource\\",\\"ResourceType\\":\\"Custom::CloudFoundryPhase1\\",\\"ResourceProperties\\":{\\"ServiceToken\\":\\"arn:aws:sns:us-west-2:540410658117:custom-resource-maybe-4-SNSTopic-11NVZUIXWZTXZ\\",\\"ExampleProperty2\\":\\"ExampleValue2\\",\\"ExampleProperty1\\":\\"ExampleValue1\\"}}",\n  "Timestamp" : "2017-06-08T14:41:17.469Z",\n  "SignatureVersion" : "1",\n  "Signature" : "eNbpNJ+mzhH4qRP89tIWj96cK0nCirmNkwjQe/3DAgWnuhKKYKvtkffZu04uTynb/Tjt/5O7y/dfUy5cTmauhAx9g8gFkwH4vQVGhJOgEI6hFEHqRgY+8FwfeKowDiZTJyixwuqY9O1wqXxOb9q0rsLpApZCVwHsIhXio3ATdmJVNuJZhpC+0Dae1jnAg7nRyLAOrjEI8yaGLLoA1s/KdrZakclwGIgNj+/as4Vscbd1+VH59cpzryoHpOIjQrESTsCqCSSHh9H4j10wjDAeFxDY3lAZOZu+SlWOsCshv+4xRiJMmWY+WCgJ2UuAXnj7N6QPKwXwnPV1UmdkXGaGWQ==",\n  "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-b95095beb82e8q6a046b3aafc7f4149a.pem",\n  "UnsubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-west-2:540420258117:custom-resource-maybe-4-SNSTopic-11NVYUIXWZTXZ:dc93952d-b252-4da3-a726-7df99631df02"\n}',
                    'MD5OfBody': '46e9054484367800fbd6bb8e13aaea36',
                    'MessageId': '6eef8c02-9874-41f8-8883-9788b911b773',
                    'ReceiptHandle': 'AQEBCLcjvnWL65ILiE9/L6WzthEatj3punqXWcxK/VGSfOhkjbfaMoy6GHtxPr/giOjO2gIW7buTv9Mkhk7/tpgbgJEm338nzoLOxqiW4s3fzoQWrjDu0HHpcD1KrMJBeVstxnLglEOOny2KRozfsLjbeH5HoXuo+8mrb0nwVUglIK2vBkAPHLOGu64/BPOR6dt2qgYK4hzytgXQprcLlS5rYrrpYkqBKjWt9PCuwSG244LuN3brNyRIgxPR9SQ/ja9CWocx7sS3Ri6tAVU8zP4OxjRmfdMj/EEdL3Wm5m4v7+hnGDnj0LSV/3UX6C1/ozIOtHY6bqN6HQ6nM48Dk6UTEm78ApFuYmOFnh5xfcAEJHHN9meqnMsYMe0l8hTEBRHDYII/W4K5APbUNNsuoU/R3uYgqWlNMFWDvxSORKPwy/2O0Kk51+ZE/fyGEaVncoof'
                },
                {
                    'Body': '{\n  "Type" : "Notification",\n  "MessageId" : "3c71d3a0-7d25-59b1-93bc-8213196855a7",\n  "TopicArn" : "arn:aws:sns:us-west-2:540420658117:custom-resource-maybe-4-SNSTopic-11NVYUIXWZTXZ",\n  "Subject" : "AWS CloudFormation custom resource request",\n  "Message" : "{\\"RequestType\\":\\"Create\\",\\"ServiceToken\\":\\"arn:aws:sns:us-west-2:540420658117:custom-resource-maybe-4-SNSTopic-11NVYUIXWZTXZ\\",\\"ResponseURL\\":\\"https://cloudformation-custom-resource-response-uswest2.s3-us-west-2.amazonaws.com/arn%3Aaws%3Acloudformation%3Aus-west-2%3A530420658117%3Astack/pcf-stack/1e820540-4c58-11e7-a965-50d5ca0184f2%7CMyCustomResource%7C4dd2c9a0-04cb-4218-908c-e3cdfad3c634?AWSAccessKeyId=AKIAI4KYMPPRGQACET5Q&Expires=1496940077&Signature=uLbeSl3rtkuDa2xf0y9oMFc1zBI%3D\\",\\"StackId\\":\\"arn:aws:cloudformation:us-west-2:540420658117:stack/pcf-stack/1e820540-4c58-11e7-a965-50d5ca0184f2\\",\\"RequestId\\":\\"4dd2c9a0-04cb-4218-908c-e2cdfad3c634\\",\\"LogicalResourceId\\":\\"MyCustomResource\\",\\"ResourceType\\":\\"Custom::CloudFoundryPhase1\\",\\"ResourceProperties\\":{\\"ServiceToken\\":\\"arn:aws:sns:us-west-2:540410658117:custom-resource-maybe-4-SNSTopic-11NVZUIXWZTXZ\\",\\"ExampleProperty2\\":\\"ExampleValue2\\",\\"ExampleProperty1\\":\\"ExampleValue1\\"}}",\n  "Timestamp" : "2017-06-08T14:41:17.469Z",\n  "SignatureVersion" : "1",\n  "Signature" : "eNbpNJ+mzhH4qRP89tIWj96cK0nCirmNkwjQe/3DAgWnuhKKYKvtkffZu04uTynb/Tjt/5O7y/dfUy5cTmauhAx9g8gFkwH4vQVGhJOgEI6hFEHqRgY+8FwfeKowDiZTJyixwuqY9O1wqXxOb9q0rsLpApZCVwHsIhXio3ATdmJVNuJZhpC+0Dae1jnAg7nRyLAOrjEI8yaGLLoA1s/KdrZakclwGIgNj+/as4Vscbd1+VH59cpzryoHpOIjQrESTsCqCSSHh9H4j10wjDAeFxDY3lAZOZu+SlWOsCshv+4xRiJMmWY+WCgJ2UuAXnj7N6QPKwXwnPV1UmdkXGaGWQ==",\n  "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-b95095beb82e8q6a046b3aafc7f4149a.pem",\n  "UnsubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-west-2:540420258117:custom-resource-maybe-4-SNSTopic-11NVYUIXWZTXZ:dc93952d-b252-4da3-a726-7df99631df02"\n}',
                    'MD5OfBody': '46e9054484367800fbd6bb8e13aaea36',
                    'MessageId': '6eef8c02-9874-41f8-8883-9788b911b773',
                    'ReceiptHandle': 'AQEBCLcjvnWL65ILiE9/L6WzthEatj3punqXWcxK/VGSfOhkjbfaMoy6GHtxPr/giOjO2gIW7buTv9Mkhk7/tpgbgJEm338nzoLOxqiW4s3fzoQWrjDu0HHpcD1KrMJBeVstxnLglEOOny2KRozfsLjbeH5HoXuo+8mrb0nwVUglIK2vBkAPHLOGu64/BPOR6dt2qgYK4hzytgXQprcLlS5rYrrpYkqBKjWt9PCuwSG244LuN3brNyRIgxPR9SQ/ja9CWocx7sS3Ri6tAVU8zP4OxjRmfdMj/EEdL3Wm5m4v7+hnGDnj0LSV/3UX6C1/ozIOtHY6bqN6HQ6nM48Dk6UTEm78ApFuYmOFnh5xfcAEJHHN9meqnMsYMe0l8hTEBRHDYII/W4K5APbUNNsuoU/R3uYgqWlNMFWDvxSORKPwy/2O0Kk51+ZE/fyGEaVncoof'
                }
            ],
            'ResponseMetadata': {
                'HTTPStatusCode': 200, 'HTTPHeaders': {
                    'connection': 'keep-alive', 'server': 'Server',
                    'content-length': '3342',
                    'content-type': 'text/xml',
                    'date': 'Thu, 08 Jun 2017 15:20:07 GMT',
                    'x-amzn-requestid': '116092e7-fee7-5580-b276-c3549380b837'
                },
                'RequestId': '116092e7-fee7-5580-b276-c3549380b837', 'RetryAttempts': 0
            }
        }

def accept_eulas(my_settings: settings.Settings):
    out, err, exit_code = accept_ert_eula(my_settings)
    if exit_code != 0:
        return out, err, exit_code
    return accept_stemcell_eula(my_settings)

def accept_stemcell_eula(my_settings: settings.Settings):
    response, result = util.exponential_backoff(
        functools.partial(post_eula, my_settings, "stemcells", my_settings.stemcell_release_id),
        check_eula_succeeded
    )
    if result == EULAResult.SUCCESS:
        return "Success", "", 0
    else:
        return "Failed to accept stemcell EULA; status code from Pivotal Network {}".format(response.status_code), "", 1

def config_opsman_auth(my_settings: settings.Settings):
    cmd = "om -k --target {url} configure-authentication --username '{user}' --password '{password}' --decryption-passphrase '{password}'".format(
        url=my_settings.opsman_url,
        user=my_settings.opsman_user,
        password=my_settings.pcf_opsmanageradminpassword
    )
    return util.exponential_backoff_cmd(cmd)

def add_new_gif(visible_gifs, gif):
    """Add new gif to visible gifs, this might push the oldest away
    from visible!
    """
    new_gif = {
        'id': gif['id'],
        'position': {'bottom': 40, 'left': 50},
        'height': gif['height'],
        'added': int(time.time())
    }

    if len(visible_gifs) >= settings['maxVisible']:
        oldest = visible_gifs.pop(0)
        logger.info('Popped oldest %s' % oldest['id'])
        # Take position from the oldest
        new_gif['position'] = oldest['position']
        new_gif['height'] = oldest['height']

    visible_gifs.append(new_gif)

def get_serial_list():
    baselist=[]
    baselist = baselist \
               + glob.glob("/dev/ttyUSB*") \
               + glob.glob("/dev/ttyACM*") \
               + glob.glob("/dev/tty.usb*") \
               + glob.glob("/dev/cu.*") \
               + glob.glob("/dev/rfcomm*")

    additionalPorts = WebConfig.settings().get(["serial", "additionalPorts"])
    for additional in additionalPorts:
        baselist += glob.glob(additional)

    if "AUTO" not in baselist:
        baselist.insert(0, "AUTO")
    prev = WebConfig.settings().get(["serial", "port"])
    if prev in baselist:
        baselist.remove(prev)
        baselist.insert(0, prev)
    if WebConfig.settings().getBoolean(["devel", "virtualPrinter", "enabled"]):
        baselist.append("VIRTUAL")
    return baselist

def __init__(self):
        handlers = [
            URLSpec(r'/', IndexHandler),
            URLSpec(r'/cat/(\d{1,2})', CatHandler, name="cat"),
            URLSpec(r'/detail/(\w+)', DetailHandler),
            URLSpec(r'/login', LoginHandler),
        ]

        tornado.web.Application.__init__(self, handlers, **settings.settings)
        self.db = torndb.Connection(host=settings.DBHOST, database=settings.DATABASE, user=settings.DBUSER, password=settings.DBPWD)

def __init__(self):
        logging.info("init MainApplication with settings: %s" % str(settings))
        tornado.web.Application.__init__(self, url_patterns, **settings)

def get_settings(env, start_response):
    start_response('200 OK', [('Content-Type', 'application/json')])
    return [json.dumps(settings)]


# Functions for responding to file requests etc.

def get_settings(env, start_response):
    start_response('200 OK', [('Content-Type', 'application/json')])
    return [json.dumps(settings)]


# Functions for responding to file requests etc.

def __init__(self):
        tornado.web.Application.__init__(self, url_patterns, **settings)

def set_local_control(self,status):
        print("Setting local control to "+str(status))
        self.usr_local_control = int(status)
                self.local_control = self.usr_local_control

        #fix into persistent memory
        settings[2] = self.usr_local_control
        f = open(self.cwd+'/../thermostat/settings.py','w')
        f.write('settings = ' + str(settings))
        f.close()

def read_temp(self):
        temp = self.instance.read_temp()
                return str(temp),str(settings[1])

def set_mode(self,mode):
        self.user_mode =str(mode)

        #fix into persistent memory
        settings[0] = self.user_mode
        f = open(self.cwd+'/../thermostat/settings.py','w')
        f.write('settings = ' + str(settings))
        f.close()

def set_setpoint(self,setpoint):
        self.instance.set_setpoint(float(setpoint))

        #fix into persistent memory
        settings[1] = float(setpoint)
        f = open(self.cwd+'/../thermostat/settings.py','w')
        f.write('settings = ' + str(settings))
        f.close()