Function resolve_pip_dev

testing/bin/run_tests.py:166–362

(logger)

def resolve_pip_dev(logger):
    # type: (Logger) -> Tuple[Version, Optional[SpecifierSet], List[str], str]

    build_system_requires = []  # type: List[str]
    pip_adhoc_build_system_requires = os.environ.get("_PEX_PIP_ADHOC_BUILD_SYSTEM_REQUIRES")
    if pip_adhoc_build_system_requires:
        build_system_requires.extend(json.loads(pip_adhoc_build_system_requires))

    extra_log_lines = []  # type: List[Text]

    pip_adhoc_requirement = os.environ.get("_PEX_PIP_ADHOC_REQUIREMENT")
    if pip_adhoc_requirement:
        pip_requirement = parse_requirement_string(pip_adhoc_requirement)
        pip_from = "_PEX_PIP_ADHOC_REQUIREMENT={requirement}".format(requirement=pip_requirement)
    else:
        data = json.loads(
            subprocess.check_output(
                args=[
                    "gh",
                    "run",
                    "-R",
                    PIP_REPO,
                    "list",
                    "-b",
                    PIP_BRANCH,
                    "-w",
                    PIP_WORKFLOW,
                    "-e",
                    "push",
                    "-s",
                    "success",
                    "-L1",
                    "--json",
                    "createdAt,displayTitle,headSha",
                ]
            )
        )
        if not data:
            raise RunError(
                "There were no green commits found for the {repo} {branch} branch in the "
                "{workflow} workflow.".format(
                    repo=PIP_REPO, branch=PIP_BRANCH, workflow=PIP_WORKFLOW
                )
            )
        green_ci_run = data[0]
        commit = green_ci_run["headSha"]
        pip_requirement = parse_requirement_string(
            "pip @ git+https://github.com/{repo}@{sha}".format(repo=PIP_REPO, sha=commit)
        )

        pip_from = "green {workflow} commit: https://github.com/{repo}/commit/{commit}".format(
            workflow=PIP_WORKFLOW, repo=PIP_REPO, commit=commit
        )
        extra_log_lines.append(
            to_unicode("> {created_at}: {pr_title}").format(
                created_at=dateutil.parser.isoparse(green_ci_run["createdAt"]).strftime("%c"),
                pr_title=green_ci_run["displayTitle"],
            )