25827687_第6章 男人激斗，女人吸吮，激烈的空战与口交（2）

admin

.parse.urlunsplit((scheme, netloc, path, query, ""))
        return url, rev, user_pass

    @staticmethod
    def make_rev_args(
        username: Optional[str], password: Optional[HiddenText]
    ) -> CommandArgs:
        """
        Return the RevOptions "extra arguments" to use in obtain().
        """
        return []

    def get_url_rev_options(self, url: HiddenText) -> Tuple[HiddenText, RevOptions]:
        """
        Return the URL and RevOptions object to use in obtain(),
        as a tuple (url, rev_options).
        """
        secret_url, rev, user_pass = self.get_url_rev_and_auth(url.secret)
        username, secret_password = user_pass
        password: Optional[HiddenText] = None
        if secret_password is not None:
            password = hide_value(secret_password)
        extra_args = self.make_rev_args(username, password)
        rev_options = self.make_rev_options(rev, extra_args=extra_args)

        return hide_url(secret_url), rev_options

    @staticmethod
    def normalize_url(url: str) -> str:
        """
        Normalize a URL for comparison by unquoting it and removing any
        trailing slash.
        """
        return urllib.parse.unquote(url).rstrip("/")

    @classmethod
    def compare_urls(cls, url1: str, url2: str) -> bool:
        """
        Compare two repo URLs for identity, ignoring incidental differences.
        """
        return cls.normalize_url(url1) == cls.normalize_url(url2)

    def fetch_new(
        self, dest: str, url: HiddenText, rev_options: RevOptions, verbosity: int
    ) -> None:
        """
        Fetch a revision from a repository, in the case that this is the
        first fetch from the repository.

        Args:
          dest: the directory to fetch the repository to.
          rev_options: a RevOptions object.
          verbosity: verbosity level.
        """
        raise NotImplementedError

    def switch(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
        """
        Switch the repo at ``dest`` to point to ``URL``.

        Args:
          rev_options: a RevOptions object.
        """
        raise NotImplementedError

    def update(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
        """
        Update an already-existing repo to the given ``rev_options``.

        Args:
          rev_options: a RevOptions object.
        """
        raise NotImplementedError

    @classmethod
    def is_commit_id_equal(cls, dest: str, name: Optional[str]) -> bool:
        """
        Return whether the id of the current commit equals the given name.

        Args:
          dest: the repository directory.
          name: a string name.
        """
        raise NotImplementedError

    def obtain(self, dest: str, url: HiddenText, verbosity: int) -> None:
        """
        Install or update in editable mode the package represented by this
        VersionControl object.

        :param dest: the repository directory in which to install or update.
        :param url: the repository URL starting with a vcs prefix.
        :param verbosity: verbosity level.
        """
        url, rev_options = self.get_url_rev_options(url)

        if not os.path.exists(dest):
            self.fetch_new(dest, url, rev_options, verbosity=verbosity)
            return

        rev_display = rev_options.to_display()
        if self.is_repository_directory(dest):
            existing_url = self.get_remote_url(dest)
            if self.compare_urls(existing_url, url.secret):
                logger.debug(
                    "%s in %s exists, and has correct URL (%s)",
                    self.repo_name.title(),
                    display_path(dest),
                    url,
                )
                if not self.is_commit_id_equal(dest, rev_options.rev):
                    logger.info(
                        "Updating %s %s%s",
                        display_path(dest),
                        self.repo_name,
                        rev_display,
                    )
                    self.update(dest, url, rev_options)
                else:
                    logger.info("Skipping because already up-to-date.")
                return

            logger.warning(
                "%s %s in %s exists with URL %s",
                self.name,
                self.repo_name,
                display_path(dest),
                existing_url,
            )
            prompt = ("(s)witch, (i)gnore, (w)ipe, (b)ackup ", ("s", "i", "w", "b"))
        else:
            logger.warning(
                "Directory %s already exists, and is not a %s %s.",
                dest,
                self.name,
                self.repo_name,
            )
            # https://github.com/python/mypy/issues/1174
            prompt = ("(i)gnore, (w)ipe, (b)ackup ", ("i", "w", "b"))  # type: ignore

        logger.warning(
            "The plan is to install the %s repository %s",
            self.name,
            url,
        )
        response = ask_path_exists(f"What to do?  {prompt[0]}", prompt[1])

        if response == "a":
            sys.exit(-1)

        if response == "w":
            logger.warning("Deleting %s", display_path(dest))
            rmtree(dest)
            self.fetch_new(dest, url, rev_options, verbosity=verbosity)
            return

        if response == "b":
            dest_dir = backup_dir(dest)
            logger.warning("Backing up %s to %s", display_path(dest), dest_dir)
            shutil.move(dest, dest_dir)
            self.fetch_new(dest, url, rev_options, verbosity=verbosity)
            return

        # Do nothing if the response is "i".
        if response == "s":
            logger.info(
                "Switching %s %s to %s%s",
                self.repo_name,
                display_path(dest),
                url,
                rev_display,
            )
            self.switch(dest, url, rev_options)

    def unpack(self, location: str, url: HiddenText, verbosity: int) -> None:
        """
        Clean up current location and download the url repository
        (and vcs infos) into location

        :param url: the repository URL starting with a vcs prefix.
        :param verbosity: verbosity level.
        """
        if os.path.exists(location):
            rmtree(location)
        self.obtain(location, url=url, verbosity=verbosity)

    @classmethod
    def get_remote_url(cls, location: str) -> str:
        """
        Return the url used at location

        Raises RemoteNotFoundError if the repository does not have a remote
        url configured.
        """
        raise NotImplementedError

    @classmethod
    def get_revision(cls, location: str) -> str:
        """
        Return the current commit id of the files at the given location.
        """
        raise NotImplementedError

    @classmethod
    def run_command(
        cls,
        cmd: Union[List[str], CommandArgs],
        show_stdout: bool = True,
        cwd: Optional[str] = None,
        on_returncode: 'Literal["raise", "warn", "ignore"]' = "raise",
        extra_ok_returncodes: Optional[Iterable[int]] = None,
        command_desc: Optional[str] = None,
        extra_environ: Optional[Mapping[str, Any]] = None,
        spinner: Optional[SpinnerInterface] = None,
        log_failed_cmd: bool = True,
        stdout_only: bool = False,
    ) -> str:
        """
        Run a VCS subcommand
        This is simply a wrapper around call_subprocess that adds the VCS
        command name, and checks that the VCS is available
        """
        cmd = make_command(cls.name, *cmd)
        if command_desc is None:
            command_desc = format_command_args(cmd)
        try:
            return call_subprocess(
                cmd,
                show_stdout,
                cwd,
                on_returncode=on_returncode,
                extra_ok_returncodes=extra_ok_returncodes,
                command_desc=command_desc,
                extra_environ=extra_environ,
                unset_environ=cls.unset_environ,
                spinner=spinner,
                log_failed_cmd=log_failed_cmd,
                stdout_only=stdout_only,
            )
        except NotADirectoryError:
            raise BadCommand(f"Cannot find command {cls.name!r} - invalid PATH")
        except FileNotFoundError:
            # errno.ENOENT = no such file or directory
            # In other words, the VCS executable isn't available
            raise BadCommand(
                f"Cannot find command {cls.name!r} - do you have "
                f"{cls.name!r} installed and in your PATH?"
            )
        except PermissionError:
            # errno.EACCES = Permission denied
            # This error occurs, for instance, when the command is installed
            # only for another user. So, the current user don't have
            # permission to call the other user command.
            raise BadCommand(
                f"No permission to execute {cls.name!r} - install it "
                f"locally, globally (ask admin), or check your PATH. "
                f"See possible solutions at "
                f"https://pip.pypa.io/en/latest/reference/pip_freeze/"
                f"#fixing-permission-denied."
            )

    @classmethod
    def is_repository_directory(cls, path: str) -> bool:
        """
        Return whether a directory path is a repository directory.
        """
        logger.debug("Checking in %s for %s (%s)...", path, cls.dirname, cls.name)
        return os.path.exists(os.path.join(path, cls.dirname))

    @classmethod
    def get_repository_root(cls, location: str) -> Optional[str]:
        """
        Return the "root" (top-level) directory controlled by the vcs,
        or `None` if the directory is not in any.

        It is meant to be overridden to implement smarter detection
        mechanisms for specific vcs.

        This can do more than is_repository_directory() alone. For
        example, the Git override checks that Git is actually available.
        """
        if cls.is_repository_directory(location):
            return location
        return None