11725709386

Committed 07 Nov 2024 03:07PM CUT coverage: 71.137%. Remained the same

Build # 11725709386

Build Type

push

github

Committed by

beckermr

Commit Message

Updated CHANGELOG for 3.44.2

Run Details

1474 of 2183 branches covered (67.52%)

3236 of 4549 relevant lines covered (71.14%)

0.71 hits per line

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

13.38

/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 _conda_forge_specific_repo_setup(gh_repo):
    # setup branch protections ruleset
    # default branch may not exist yet
    ruleset_name = "conda-forge-branch-protection"

    # first, check if the ruleset exists already
    rulesets_url = gh_repo.url + "/rulesets"
    _, ruleset_list = gh_repo._requester.requestJsonAndCheck(
        "GET", rulesets_url
    )
    ruleset_id = None
    for ruleset in ruleset_list:
        if ruleset["name"] == ruleset_name:
            ruleset_id = ruleset["id"]
            break

    if ruleset_id is not None:
        print("Updating branch protections")
        # update ruleset
        method = "PUT"
        url = f"{rulesets_url}/{ruleset_id}"
    else:
        print("Enabling branch protections")
        # new ruleset
        method = "POST"
        url = rulesets_url

    gh_repo._requester.requestJsonAndCheck(
        method,
        url,
        input={
            "name": ruleset_name,
            "target": "branch",
            "conditions": {
                "ref_name": {"exclude": [], "include": ["~DEFAULT_BRANCH"]}
            },
            "rules": [{"type": "deletion"}, {"type": "non_fast_forward"}],
            "enforcement": "active",
        },
    )


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
    is_conda_forge = False
    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)
        if args.organization == "conda-forge":
            is_conda_forge = True

    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}.",
        )

        if is_conda_forge:
            _conda_forge_specific_repo_setup(gh_repo)

        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 _conda_forge_specific_repo_setup(gh_repo):	1✔
106	# setup branch protections ruleset
107	# default branch may not exist yet
108	ruleset_name = "conda-forge-branch-protection"	×
109
110	# first, check if the ruleset exists already
111	rulesets_url = gh_repo.url + "/rulesets"	×
112	_, ruleset_list = gh_repo._requester.requestJsonAndCheck(	×
113	"GET", rulesets_url
114	)
115	ruleset_id = None	×
116	for ruleset in ruleset_list:	×
117	if ruleset["name"] == ruleset_name:	×
118	ruleset_id = ruleset["id"]	×
119	break	×
120
121	if ruleset_id is not None:	×
122	print("Updating branch protections")	×
123	# update ruleset
124	method = "PUT"	×
125	url = f"{rulesets_url}/{ruleset_id}"	×
126	else:
127	print("Enabling branch protections")	×
128	# new ruleset
129	method = "POST"	×
130	url = rulesets_url	×
131
132	gh_repo._requester.requestJsonAndCheck(	×
133	method,
134	url,
135	input={
136	"name": ruleset_name,
137	"target": "branch",
138	"conditions": {
139	"ref_name": {"exclude": [], "include": ["~DEFAULT_BRANCH"]}
140	},
141	"rules": [{"type": "deletion"}, {"type": "non_fast_forward"}],
142	"enforcement": "active",
143	},
144	)
145
146
147	def create_github_repo(args):	1✔
148	token = gh_token()	×
149
150	# Load the conda-forge config and read metadata from the feedstock recipe
151	forge_config = _load_forge_config(args.feedstock_directory, None)	×
152	metadata = _get_metadata_from_feedstock_dir(	×
153	args.feedstock_directory, forge_config
154	)
155
156	feedstock_name = get_feedstock_name_from_meta(metadata)	×
157
158	gh = Github(token)	×
159	user_or_org = None	×
160	is_conda_forge = False	×
161	if args.user is not None:	×
162	pass
163	# User has been defined, and organization has not.
164	user_or_org = gh.get_user()	×
165	else:
166	# Use the organization provided.
167	user_or_org = gh.get_organization(args.organization)	×
168	if args.organization == "conda-forge":	×
169	is_conda_forge = True	×
170
171	repo_name = f"{feedstock_name}-feedstock"	×
172	try:	×
173	gh_repo = user_or_org.create_repo(	×
174	repo_name,
175	has_wiki=False,
176	private=args.private,
177	description=f"A conda-smithy repository for {feedstock_name}.",
178	)
179
180	if is_conda_forge:	×
181	_conda_forge_specific_repo_setup(gh_repo)	×
182
183	print(f"Created {gh_repo.full_name} on github")	×
184	except GithubException as gh_except:	×
185	if (	×
186	gh_except.data.get("errors", [{}])[0].get("message", "")
187	!= "name already exists on this account"
188	):
189	raise	×
190	gh_repo = user_or_org.get_repo(repo_name)	×
191	print("Github repository already exists.")	×
192
193	# Now add this new repo as a remote on the local clone.
194	repo = Repo(args.feedstock_directory)	×
195	remote_name = args.remote_name.strip()	×
196	if remote_name:	×
197	if remote_name in [remote.name for remote in repo.remotes]:	×
198	existing_remote = repo.remotes[remote_name]	×
199	if existing_remote.url != gh_repo.ssh_url:	×
200	print(	×
201	f"Remote {remote_name} already exists, and doesn't point to {gh_repo.ssh_url} "
202	f"(it points to {existing_remote.url})."
203	)
204	else:
205	repo.create_remote(remote_name, gh_repo.ssh_url)	×
206
207	if args.extra_admin_users is not None:	×
208	for user in args.extra_admin_users:	×
209	gh_repo.add_to_collaborators(user, "admin")	×
210
211	if args.add_teams:	×
212	if isinstance(user_or_org, Organization):	×
213	configure_github_team(	×
214	metadata, gh_repo, user_or_org, feedstock_name
215	)
216
217
218	def accept_all_repository_invitations(gh):	1✔
219	user = gh.get_user()	×
220	invitations = github.PaginatedList.PaginatedList(	×
221	github.Invitation.Invitation,
222	user._requester,
223	user.url + "/repository_invitations",
224	None,
225	)
226	for invite in invitations:	×
227	invite._requester.requestJsonAndCheck("PATCH", invite.url)	×
228
229
230	def remove_from_project(gh, org, project):	1✔
231	user = gh.get_user()	×
232	repo = gh.get_repo(f"{org}/{project}")	×
233	repo.remove_from_collaborators(user.login)	×
234
235
236	def configure_github_team(meta, gh_repo, org, feedstock_name, remove=True):	1✔
237	# Add a team for this repo and add the maintainers to it.
238	superlative = [	×
239	"awesome",
240	"slick",
241	"formidable",
242	"awe-inspiring",
243	"breathtaking",
244	"magnificent",
245	"wonderous",
246	"stunning",
247	"astonishing",
248	"superb",
249	"splendid",
250	"impressive",
251	"unbeatable",
252	"excellent",
253	"top",
254	"outstanding",
255	"exalted",
256	"standout",
257	"smashing",
258	]
259
260	maintainers = set(meta.meta.get("extra", {}).get("recipe-maintainers", []))	×
261	maintainers = set(maintainer.lower() for maintainer in maintainers)	×
262	maintainer_teams = set(m for m in maintainers if "/" in m)	×
263	maintainers = set(m for m in maintainers if "/" not in m)	×
264
265	# Try to get team or create it if it doesn't exist.
266	team_name = feedstock_name	×
267	current_maintainer_teams = list(gh_repo.get_teams())	×
268	fs_team = next(	×
269	(team for team in current_maintainer_teams if team.name == team_name),
270	None,
271	)
272	current_maintainers = set()	×
273	if not fs_team:	×
274	fs_team = create_team(	×
275	org,
276	team_name,
277	f"The {choice(superlative)} {team_name} contributors!",
278	)
279	fs_team.add_to_repos(gh_repo)	×
280
281	current_maintainers = set([e.login.lower() for e in fs_team.get_members()])	×
282
283	# Get the all-members team
284	description = f"All of the awesome {org.login} contributors!"	×
285	all_members_team = get_cached_team(org, "all-members", description)	×
286	new_org_members = set()	×
287
288	# Add only the new maintainers to the team.
289	# Also add the new maintainers to all-members if not already included.
290	for new_maintainer in maintainers - current_maintainers:	×
291	add_membership(fs_team, new_maintainer)	×
292
293	if not has_in_members(all_members_team, new_maintainer):	×
294	add_membership(all_members_team, new_maintainer)	×
295	new_org_members.add(new_maintainer)	×
296
297	# Remove any maintainers that need to be removed (unlikely here).
298	if remove:	×
299	for old_maintainer in current_maintainers - maintainers:	×
300	remove_membership(fs_team, old_maintainer)	×
301
302	# Add any new maintainer teams
303	maintainer_teams = set(	×
304	m.split("/")[1]
305	for m in maintainer_teams
306	if m.startswith(str(org.login))
307	)
308	current_maintainer_team_objs = {	×
309	team.slug: team for team in current_maintainer_teams
310	}
311	current_maintainer_teams = set(	×
312	[team.slug for team in current_maintainer_teams]
313	)
314	for new_team in maintainer_teams - current_maintainer_teams:	×
315	team = org.get_team_by_slug(new_team)	×
316	team.add_to_repos(gh_repo)	×
317
318	# remove any old teams
319	if remove:	×
320	for old_team in current_maintainer_teams - maintainer_teams:	×
321	team = current_maintainer_team_objs.get(	×
322	old_team, org.get_team_by_slug(old_team)
323	)
324	if team.name == fs_team.name:	×
325	continue
326	team.remove_from_repos(gh_repo)	×
327
328	return maintainers, current_maintainers, new_org_members	×

conda-forge / conda-smithy / 11725709386

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