10728154446

Committed 05 Sep 2024 09:05PM CUT coverage: 71.467%. Remained the same

Build # 10728154446

Build Type

Pull #2053

github

Committed by

web-flow

Commit Message

Merge c168e015b into a77a55afc

Pull Request Pull Request #2053: Fix `./build-locally.py` in feedstocks

Run Details

1452 of 2145 branches covered (67.69%)

3181 of 4451 relevant lines covered (71.47%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

14.81

/conda_smithy/github.py

import os
from random import choice

import github
from git import Repo
from github import Github
from github.GithubException import GithubException
from github.Organization import Organization
from github.Team import Team

from conda_smithy.configure_feedstock import _load_forge_config
from conda_smithy.utils import (
    _get_metadata_from_feedstock_dir,
    get_feedstock_name_from_meta,
)


def gh_token():
    try:
        with open(os.path.expanduser("~/.conda-smithy/github.token")) as fh:
            token = fh.read().strip()
        if not token:
            raise ValueError()
    except (OSError, ValueError):
        msg = (
            "No github token. Go to https://github.com/settings/tokens/new and generate\n"
            "a token with repo access. Put it in ~/.conda-smithy/github.token"
        )
        raise RuntimeError(msg)
    return token


def create_team(org, name, description, repo_names=[]):
    # PyGithub creates secret teams, and has no way of turning that off! :(
    post_parameters = {
        "name": name,
        "description": description,
        "privacy": "closed",
        "permission": "push",
        "repo_names": repo_names,
    }
    headers, data = org._requester.requestJsonAndCheck(
        "POST", org.url + "/teams", input=post_parameters
    )
    return Team(org._requester, headers, data, completed=True)


def add_membership(team, member):
    headers, data = team._requester.requestJsonAndCheck(
        "PUT", team.url + "/memberships/" + member
    )
    return (headers, data)


def remove_membership(team, member):
    headers, data = team._requester.requestJsonAndCheck(
        "DELETE", team.url + "/memberships/" + member
    )
    return (headers, data)


def has_in_members(team, member):
    status, headers, data = team._requester.requestJson(
        "GET", team.url + "/members/" + member
    )
    return status == 204


def get_cached_team(org, team_name, description=""):
    cached_file = os.path.expanduser(
        f"~/.conda-smithy/{org.login}-{team_name}-team"
    )
    try:
        with open(cached_file) as fh:
            team_id = int(fh.read().strip())
            return org.get_team(team_id)
    except OSError:
        pass

    try:
        repo = org.get_repo(f"{team_name}-feedstock")
        team = next(
            (team for team in repo.get_teams() if team.name == team_name), None
        )
        if team:
            return team
    except GithubException:
        pass

    team = next(
        (team for team in org.get_teams() if team.name == team_name), None
    )
    if not team:
        if description:
            team = create_team(org, team_name, description, [])
        else:
            raise RuntimeError(f"Couldn't find team {team_name}")

    with open(cached_file, "w") as fh:
        fh.write(str(team.id))

    return team


def create_github_repo(args):
    token = gh_token()

    # Load the conda-forge config and read metadata from the feedstock recipe
    forge_config = _load_forge_config(args.feedstock_directory, None)
    metadata = _get_metadata_from_feedstock_dir(
        args.feedstock_directory, forge_config
    )

    feedstock_name = get_feedstock_name_from_meta(metadata)

    gh = Github(token)
    user_or_org = None
    if args.user is not None:
        pass
        # User has been defined, and organization has not.
        user_or_org = gh.get_user()
    else:
        # Use the organization provided.
        user_or_org = gh.get_organization(args.organization)

    repo_name = f"{feedstock_name}-feedstock"
    try:
        gh_repo = user_or_org.create_repo(
            repo_name,
            has_wiki=False,
            private=args.private,
            description=f"A conda-smithy repository for {feedstock_name}.",
        )
        print(f"Created {gh_repo.full_name} on github")
    except GithubException as gh_except:
        if (
            gh_except.data.get("errors", [{}])[0].get("message", "")
            != "name already exists on this account"
        ):
            raise
        gh_repo = user_or_org.get_repo(repo_name)
        print("Github repository already exists.")

    # Now add this new repo as a remote on the local clone.
    repo = Repo(args.feedstock_directory)
    remote_name = args.remote_name.strip()
    if remote_name:
        if remote_name in [remote.name for remote in repo.remotes]:
            existing_remote = repo.remotes[remote_name]
            if existing_remote.url != gh_repo.ssh_url:
                print(
                    f"Remote {remote_name} already exists, and doesn't point to {gh_repo.ssh_url} "
                    f"(it points to {existing_remote.url})."
                )
        else:
            repo.create_remote(remote_name, gh_repo.ssh_url)

    if args.extra_admin_users is not None:
        for user in args.extra_admin_users:
            gh_repo.add_to_collaborators(user, "admin")

    if args.add_teams:
        if isinstance(user_or_org, Organization):
            configure_github_team(
                metadata, gh_repo, user_or_org, feedstock_name
            )


def accept_all_repository_invitations(gh):
    user = gh.get_user()
    invitations = github.PaginatedList.PaginatedList(
        github.Invitation.Invitation,
        user._requester,
        user.url + "/repository_invitations",
        None,
    )
    for invite in invitations:
        invite._requester.requestJsonAndCheck("PATCH", invite.url)


def remove_from_project(gh, org, project):
    user = gh.get_user()
    repo = gh.get_repo(f"{org}/{project}")
    repo.remove_from_collaborators(user.login)


def configure_github_team(meta, gh_repo, org, feedstock_name, remove=True):
    # Add a team for this repo and add the maintainers to it.
    superlative = [
        "awesome",
        "slick",
        "formidable",
        "awe-inspiring",
        "breathtaking",
        "magnificent",
        "wonderous",
        "stunning",
        "astonishing",
        "superb",
        "splendid",
        "impressive",
        "unbeatable",
        "excellent",
        "top",
        "outstanding",
        "exalted",
        "standout",
        "smashing",
    ]

    maintainers = set(meta.meta.get("extra", {}).get("recipe-maintainers", []))
    maintainers = set(maintainer.lower() for maintainer in maintainers)
    maintainer_teams = set(m for m in maintainers if "/" in m)
    maintainers = set(m for m in maintainers if "/" not in m)

    # Try to get team or create it if it doesn't exist.
    team_name = feedstock_name
    current_maintainer_teams = list(gh_repo.get_teams())
    fs_team = next(
        (team for team in current_maintainer_teams if team.name == team_name),
        None,
    )
    current_maintainers = set()
    if not fs_team:
        fs_team = create_team(
            org,
            team_name,
            f"The {choice(superlative)} {team_name} contributors!",
        )
        fs_team.add_to_repos(gh_repo)

    current_maintainers = set([e.login.lower() for e in fs_team.get_members()])

    # Get the all-members team
    description = f"All of the awesome {org.login} contributors!"
    all_members_team = get_cached_team(org, "all-members", description)
    new_org_members = set()

    # Add only the new maintainers to the team.
    # Also add the new maintainers to all-members if not already included.
    for new_maintainer in maintainers - current_maintainers:
        add_membership(fs_team, new_maintainer)

        if not has_in_members(all_members_team, new_maintainer):
            add_membership(all_members_team, new_maintainer)
            new_org_members.add(new_maintainer)

    # Remove any maintainers that need to be removed (unlikely here).
    if remove:
        for old_maintainer in current_maintainers - maintainers:
            remove_membership(fs_team, old_maintainer)

    # Add any new maintainer teams
    maintainer_teams = set(
        m.split("/")[1]
        for m in maintainer_teams
        if m.startswith(str(org.login))
    )
    current_maintainer_team_objs = {
        team.slug: team for team in current_maintainer_teams
    }
    current_maintainer_teams = set(
        [team.slug for team in current_maintainer_teams]
    )
    for new_team in maintainer_teams - current_maintainer_teams:
        team = org.get_team_by_slug(new_team)
        team.add_to_repos(gh_repo)

    # remove any old teams
    if remove:
        for old_team in current_maintainer_teams - maintainer_teams:
            team = current_maintainer_team_objs.get(
                old_team, org.get_team_by_slug(old_team)
            )
            if team.name == fs_team.name:
                continue
            team.remove_from_repos(gh_repo)

    return maintainers, current_maintainers, new_org_members

1	import os	1✔
2	from random import choice	1✔
3
4	import github	1✔
5	from git import Repo	1✔
6	from github import Github	1✔
7	from github.GithubException import GithubException	1✔
8	from github.Organization import Organization	1✔
9	from github.Team import Team	1✔
10
11	from conda_smithy.configure_feedstock import _load_forge_config	1✔
12	from conda_smithy.utils import (	1✔
13	_get_metadata_from_feedstock_dir,
14	get_feedstock_name_from_meta,
15	)
16
17
18	def gh_token():	1✔
19	try:	×
20	with open(os.path.expanduser("~/.conda-smithy/github.token")) as fh:	×
21	token = fh.read().strip()	×
22	if not token:	×
23	raise ValueError()	×
24	except (OSError, ValueError):	×
25	msg = (	×
26	"No github token. Go to https://github.com/settings/tokens/new and generate\n"
27	"a token with repo access. Put it in ~/.conda-smithy/github.token"
28	)
29	raise RuntimeError(msg)	×
30	return token	×
31
32
33	def create_team(org, name, description, repo_names=[]):	1✔
34	# PyGithub creates secret teams, and has no way of turning that off! :(
35	post_parameters = {	×
36	"name": name,
37	"description": description,
38	"privacy": "closed",
39	"permission": "push",
40	"repo_names": repo_names,
41	}
42	headers, data = org._requester.requestJsonAndCheck(	×
43	"POST", org.url + "/teams", input=post_parameters
44	)
45	return Team(org._requester, headers, data, completed=True)	×
46
47
48	def add_membership(team, member):	1✔
49	headers, data = team._requester.requestJsonAndCheck(	×
50	"PUT", team.url + "/memberships/" + member
51	)
52	return (headers, data)	×
53
54
55	def remove_membership(team, member):	1✔
56	headers, data = team._requester.requestJsonAndCheck(	×
57	"DELETE", team.url + "/memberships/" + member
58	)
59	return (headers, data)	×
60
61
62	def has_in_members(team, member):	1✔
63	status, headers, data = team._requester.requestJson(	×
64	"GET", team.url + "/members/" + member
65	)
66	return status == 204	×
67
68
69	def get_cached_team(org, team_name, description=""):	1✔
70	cached_file = os.path.expanduser(	×
71	f"~/.conda-smithy/{org.login}-{team_name}-team"
72	)
73	try:	×
74	with open(cached_file) as fh:	×
75	team_id = int(fh.read().strip())	×
76	return org.get_team(team_id)	×
77	except OSError:	×
78	pass	×
79
80	try:	×
81	repo = org.get_repo(f"{team_name}-feedstock")	×
82	team = next(	×
83	(team for team in repo.get_teams() if team.name == team_name), None
84	)
85	if team:	×
86	return team	×
87	except GithubException:	×
88	pass	×
89
90	team = next(	×
91	(team for team in org.get_teams() if team.name == team_name), None
92	)
93	if not team:	×
94	if description:	×
95	team = create_team(org, team_name, description, [])	×
96	else:
97	raise RuntimeError(f"Couldn't find team {team_name}")	×
98
99	with open(cached_file, "w") as fh:	×
100	fh.write(str(team.id))	×
101
102	return team	×
103
104
105	def create_github_repo(args):	1✔
106	token = gh_token()	×
107
108	# Load the conda-forge config and read metadata from the feedstock recipe
109	forge_config = _load_forge_config(args.feedstock_directory, None)	×
110	metadata = _get_metadata_from_feedstock_dir(	×
111	args.feedstock_directory, forge_config
112	)
113
114	feedstock_name = get_feedstock_name_from_meta(metadata)	×
115
116	gh = Github(token)	×
117	user_or_org = None	×
118	if args.user is not None:	×
119	pass
120	# User has been defined, and organization has not.
121	user_or_org = gh.get_user()	×
122	else:
123	# Use the organization provided.
124	user_or_org = gh.get_organization(args.organization)	×
125
126	repo_name = f"{feedstock_name}-feedstock"	×
127	try:	×
128	gh_repo = user_or_org.create_repo(	×
129	repo_name,
130	has_wiki=False,
131	private=args.private,
132	description=f"A conda-smithy repository for {feedstock_name}.",
133	)
134	print(f"Created {gh_repo.full_name} on github")	×
135	except GithubException as gh_except:	×
136	if (	×
137	gh_except.data.get("errors", [{}])[0].get("message", "")
138	!= "name already exists on this account"
139	):
140	raise	×
141	gh_repo = user_or_org.get_repo(repo_name)	×
142	print("Github repository already exists.")	×
143
144	# Now add this new repo as a remote on the local clone.
145	repo = Repo(args.feedstock_directory)	×
146	remote_name = args.remote_name.strip()	×
147	if remote_name:	×
148	if remote_name in [remote.name for remote in repo.remotes]:	×
149	existing_remote = repo.remotes[remote_name]	×
150	if existing_remote.url != gh_repo.ssh_url:	×
151	print(	×
152	f"Remote {remote_name} already exists, and doesn't point to {gh_repo.ssh_url} "
153	f"(it points to {existing_remote.url})."
154	)
155	else:
156	repo.create_remote(remote_name, gh_repo.ssh_url)	×
157
158	if args.extra_admin_users is not None:	×
159	for user in args.extra_admin_users:	×
160	gh_repo.add_to_collaborators(user, "admin")	×
161
162	if args.add_teams:	×
163	if isinstance(user_or_org, Organization):	×
164	configure_github_team(	×
165	metadata, gh_repo, user_or_org, feedstock_name
166	)
167
168
169	def accept_all_repository_invitations(gh):	1✔
170	user = gh.get_user()	×
171	invitations = github.PaginatedList.PaginatedList(	×
172	github.Invitation.Invitation,
173	user._requester,
174	user.url + "/repository_invitations",
175	None,
176	)
177	for invite in invitations:	×
178	invite._requester.requestJsonAndCheck("PATCH", invite.url)	×
179
180
181	def remove_from_project(gh, org, project):	1✔
182	user = gh.get_user()	×
183	repo = gh.get_repo(f"{org}/{project}")	×
184	repo.remove_from_collaborators(user.login)	×
185
186
187	def configure_github_team(meta, gh_repo, org, feedstock_name, remove=True):	1✔
188	# Add a team for this repo and add the maintainers to it.
189	superlative = [	×
190	"awesome",
191	"slick",
192	"formidable",
193	"awe-inspiring",
194	"breathtaking",
195	"magnificent",
196	"wonderous",
197	"stunning",
198	"astonishing",
199	"superb",
200	"splendid",
201	"impressive",
202	"unbeatable",
203	"excellent",
204	"top",
205	"outstanding",
206	"exalted",
207	"standout",
208	"smashing",
209	]
210
211	maintainers = set(meta.meta.get("extra", {}).get("recipe-maintainers", []))	×
212	maintainers = set(maintainer.lower() for maintainer in maintainers)	×
213	maintainer_teams = set(m for m in maintainers if "/" in m)	×
214	maintainers = set(m for m in maintainers if "/" not in m)	×
215
216	# Try to get team or create it if it doesn't exist.
217	team_name = feedstock_name	×
218	current_maintainer_teams = list(gh_repo.get_teams())	×
219	fs_team = next(	×
220	(team for team in current_maintainer_teams if team.name == team_name),
221	None,
222	)
223	current_maintainers = set()	×
224	if not fs_team:	×
225	fs_team = create_team(	×
226	org,
227	team_name,
228	f"The {choice(superlative)} {team_name} contributors!",
229	)
230	fs_team.add_to_repos(gh_repo)	×
231
232	current_maintainers = set([e.login.lower() for e in fs_team.get_members()])	×
233
234	# Get the all-members team
235	description = f"All of the awesome {org.login} contributors!"	×
236	all_members_team = get_cached_team(org, "all-members", description)	×
237	new_org_members = set()	×
238
239	# Add only the new maintainers to the team.
240	# Also add the new maintainers to all-members if not already included.
241	for new_maintainer in maintainers - current_maintainers:	×
242	add_membership(fs_team, new_maintainer)	×
243
244	if not has_in_members(all_members_team, new_maintainer):	×
245	add_membership(all_members_team, new_maintainer)	×
246	new_org_members.add(new_maintainer)	×
247
248	# Remove any maintainers that need to be removed (unlikely here).
249	if remove:	×
250	for old_maintainer in current_maintainers - maintainers:	×
251	remove_membership(fs_team, old_maintainer)	×
252
253	# Add any new maintainer teams
254	maintainer_teams = set(	×
255	m.split("/")[1]
256	for m in maintainer_teams
257	if m.startswith(str(org.login))
258	)
259	current_maintainer_team_objs = {	×
260	team.slug: team for team in current_maintainer_teams
261	}
262	current_maintainer_teams = set(	×
263	[team.slug for team in current_maintainer_teams]
264	)
265	for new_team in maintainer_teams - current_maintainer_teams:	×
266	team = org.get_team_by_slug(new_team)	×
267	team.add_to_repos(gh_repo)	×
268
269	# remove any old teams
270	if remove:	×
271	for old_team in current_maintainer_teams - maintainer_teams:	×
272	team = current_maintainer_team_objs.get(	×
273	old_team, org.get_team_by_slug(old_team)
274	)
275	if team.name == fs_team.name:	×
276	continue
277	team.remove_from_repos(gh_repo)	×
278
279	return maintainers, current_maintainers, new_org_members	×

conda-forge / conda-smithy / 10728154446

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous