(
self,
pip_configuration, # type: PipConfiguration
refresh=False, # type: bool
locked_choice=LockedChoice.AUTO, # type: LockedChoice.Value
)
| 386 | return True |
| 387 | |
| 388 | def _resolve_script( |
| 389 | self, |
| 390 | pip_configuration, # type: PipConfiguration |
| 391 | refresh=False, # type: bool |
| 392 | locked_choice=LockedChoice.AUTO, # type: LockedChoice.Value |
| 393 | ): |
| 394 | # type: (...) -> Union[str, Error] |
| 395 | |
| 396 | if "file" == self.url.scheme: |
| 397 | if not os.path.exists(self.url.path): |
| 398 | return Error( |
| 399 | "The path {path} pointed at by {entry_point} does not exist.".format( |
| 400 | path=self.url.path, entry_point=self.url |
| 401 | ) |
| 402 | ) |
| 403 | return self.url.path |
| 404 | |
| 405 | download_url = self.url.download_url |
| 406 | cache_dir = CacheDir.RUN.path( |
| 407 | "scripts", hashlib.sha1(download_url.encode("utf-8")).hexdigest() |
| 408 | ) |
| 409 | script_name = posixpath.basename(self.url.path) |
| 410 | |
| 411 | if refresh: |
| 412 | safe_rmtree(cache_dir) |
| 413 | with atomic_directory(cache_dir) as atomic_dir: |
| 414 | if not atomic_dir.is_finalized(): |
| 415 | fetcher = URLFetcher( |
| 416 | network_configuration=pip_configuration.network_configuration, |
| 417 | password_entries=pip_configuration.repos_configuration.password_entries, |
| 418 | ) |
| 419 | with fetcher.get_body_stream(download_url) as src_fp, open( |
| 420 | os.path.join(atomic_dir.work_dir, script_name), "wb" |
| 421 | ) as dst_fp: |
| 422 | shutil.copyfileobj(src_fp, dst_fp) |
| 423 | if locked_choice is not LockedChoice.IGNORE: |
| 424 | name, _ = os.path.splitext(script_name) |
| 425 | for lock in "pylock.{name}.toml".format(name=name), "pylock.toml": |
| 426 | if self._resolve_lock(fetcher, atomic_dir.work_dir, lock): |
| 427 | break |
| 428 | return os.path.join(cache_dir, script_name) |
| 429 | |
| 430 | def resolve( |
| 431 | self, |
no test coverage detected