(self)
| 47 | |
| 48 | class KaggleJwtHandler(BaseHTTPRequestHandler): |
| 49 | def do_POST(self): |
| 50 | if self.path.endswith("AttachDatasourceUsingJwtRequest"): |
| 51 | content_length = int(self.headers["Content-Length"]) |
| 52 | request = json.loads(self.rfile.read(content_length)) |
| 53 | model_ref = request["modelRef"] |
| 54 | |
| 55 | self.send_response(200) |
| 56 | self.send_header("Content-type", "application/json") |
| 57 | self.end_headers() |
| 58 | |
| 59 | if model_ref['ModelSlug'] == 'unknown': |
| 60 | self.wfile.write(bytes(json.dumps({ |
| 61 | "wasSuccessful": False, |
| 62 | }), "utf-8")) |
| 63 | return |
| 64 | |
| 65 | # Load the files |
| 66 | mount_slug = f"{model_ref['ModelSlug']}/{model_ref['Framework']}/{model_ref['InstanceSlug']}/{model_ref['VersionNumber']}" |
| 67 | os.makedirs(os.path.dirname(os.path.join(MOUNT_PATH, mount_slug))) |
| 68 | os.symlink('/input/tests/data/saved_model/', os.path.join(MOUNT_PATH, mount_slug), target_is_directory=True) |
| 69 | |
| 70 | # Return the response |
| 71 | self.wfile.write(bytes(json.dumps({ |
| 72 | "wasSuccessful": True, |
| 73 | "result": { |
| 74 | "mountSlug": mount_slug, |
| 75 | }, |
| 76 | }), "utf-8")) |
| 77 | else: |
| 78 | self.send_response(404) |
| 79 | self.wfile.write(bytes(f"Unhandled path: {self.path}", "utf-8")) |
| 80 | |
| 81 | class TestKaggleModuleResolver(unittest.TestCase): |
| 82 | def test_kaggle_resolver_long_url_succeeds(self): |
nothing calls this directly
no outgoing calls
no test coverage detected