(self, parsed_args, parsed_globals)
| 118 | ] |
| 119 | |
| 120 | def _run_main(self, parsed_args, parsed_globals): |
| 121 | s3_client = self._session.create_client( |
| 122 | "s3", |
| 123 | config=Config(signature_version='s3v4'), |
| 124 | region_name=parsed_globals.region, |
| 125 | verify=parsed_globals.verify_ssl, |
| 126 | ) |
| 127 | |
| 128 | template_path = parsed_args.template_file |
| 129 | if not os.path.isfile(template_path): |
| 130 | raise exceptions.InvalidTemplatePathError( |
| 131 | template_path=template_path |
| 132 | ) |
| 133 | |
| 134 | bucket = parsed_args.s3_bucket |
| 135 | |
| 136 | self.s3_uploader = S3Uploader( |
| 137 | s3_client, |
| 138 | bucket, |
| 139 | parsed_args.s3_prefix, |
| 140 | parsed_args.kms_key_id, |
| 141 | parsed_args.force_upload, |
| 142 | ) |
| 143 | # attach the given metadata to the artifacts to be uploaded |
| 144 | self.s3_uploader.artifact_metadata = parsed_args.metadata |
| 145 | |
| 146 | output_file = parsed_args.output_template_file |
| 147 | use_json = parsed_args.use_json |
| 148 | exported_str = self._export(template_path, use_json) |
| 149 | |
| 150 | sys.stdout.write("\n") |
| 151 | self.write_output(output_file, exported_str) |
| 152 | |
| 153 | if output_file: |
| 154 | msg = self.MSG_PACKAGED_TEMPLATE_WRITTEN.format( |
| 155 | output_file_name=output_file, |
| 156 | output_file_path=os.path.abspath(output_file), |
| 157 | ) |
| 158 | sys.stdout.write(msg) |
| 159 | |
| 160 | sys.stdout.flush() |
| 161 | return 0 |
| 162 | |
| 163 | def _export(self, template_path, use_json): |
| 164 | template = Template(template_path, os.getcwd(), self.s3_uploader) |
nothing calls this directly
no test coverage detected