(self, client_func, expected_path, expected_body, source=None, success=True)
| 34 | TEST_JWT = 'test-secrets-key' |
| 35 | |
| 36 | def _test_client(self, client_func, expected_path, expected_body, source=None, success=True): |
| 37 | _request = {} |
| 38 | |
| 39 | class GetKernelRunSourceForCaipHandler(UserSessionHTTPHandler): |
| 40 | def set_request(self): |
| 41 | _request['path'] = self.path |
| 42 | content_len = int(self.headers.get('Content-Length')) |
| 43 | _request['body'] = json.loads(self.rfile.read(content_len)) |
| 44 | _request['headers'] = self.headers |
| 45 | |
| 46 | def get_response(self): |
| 47 | if success: |
| 48 | return {'result': {'source': source}, 'wasSuccessful': 'true'} |
| 49 | return {'wasSuccessful': 'false'} |
| 50 | |
| 51 | env = EnvironmentVarGuard() |
| 52 | env.set(_KAGGLE_USER_SECRETS_TOKEN_ENV_VAR_NAME, self.TEST_JWT) |
| 53 | with env: |
| 54 | with HTTPServer((self.SERVER_ADDRESS.hostname, self.SERVER_ADDRESS.port), GetKernelRunSourceForCaipHandler) as httpd: |
| 55 | threading.Thread(target=httpd.serve_forever).start() |
| 56 | |
| 57 | try: |
| 58 | # Make sure this server is pointed to by clients (the port may have been dynamically chosen): |
| 59 | env.set(_KAGGLE_URL_BASE_ENV_VAR_NAME, "http://"+httpd.server_address[0]+":"+str(httpd.server_address[1])) |
| 60 | client_func() |
| 61 | finally: |
| 62 | httpd.shutdown() |
| 63 | |
| 64 | path, headers, body = _request['path'], _request['headers'], _request['body'] |
| 65 | self.assertEqual( |
| 66 | path, |
| 67 | expected_path, |
| 68 | msg="Fake server did not receive the right request from UserSessionClient.") |
| 69 | self.assertEqual( |
| 70 | body, |
| 71 | expected_body, |
| 72 | msg="Fake server did not receive the right body from UserSessionClient.") |
| 73 | |
| 74 | def test_no_token_fails(self): |
| 75 | env = EnvironmentVarGuard() |
no outgoing calls
no test coverage detected