(self)
| 163 | self._expiryKey = expiryKey |
| 164 | |
| 165 | def token(self): |
| 166 | fullCmd = self._cmd + (" ") + " ".join(self._args) |
| 167 | process = subprocess.Popen( |
| 168 | [self._cmd] + self._args, |
| 169 | stdout=subprocess.PIPE, |
| 170 | stderr=subprocess.PIPE, |
| 171 | universal_newlines=True) |
| 172 | (stdout, stderr) = process.communicate() |
| 173 | exit_code = process.wait() |
| 174 | if exit_code != 0: |
| 175 | msg = 'cmd-path: process returned %d' % exit_code |
| 176 | msg += "\nCmd: %s" % fullCmd |
| 177 | stderr = stderr.strip() |
| 178 | if stderr: |
| 179 | msg += '\nStderr: %s' % stderr |
| 180 | raise ConfigException(msg) |
| 181 | try: |
| 182 | data = json.loads(stdout) |
| 183 | except ValueError as de: |
| 184 | raise ConfigException( |
| 185 | 'exec: failed to decode process output: %s' % de) |
| 186 | A = namedtuple('A', ['token', 'expiry']) |
| 187 | return A( |
| 188 | token=data['credential']['access_token'], |
| 189 | expiry=parse_rfc3339(data['credential']['token_expiry'])) |
| 190 | |
| 191 | |
| 192 | class KubeConfigLoader: |
no test coverage detected