11259992998

Committed 09 Oct 2024 05:25PM UTC coverage: 70.966%. Remained the same

Build # 11259992998

Build Type

push

github

Committed by

beckermr

Commit Message

Merge branch 'main' of https://github.com/conda-forge/conda-smithy

Run Details

1434 of 2134 branches covered (67.2%)

1 of 17 new or added lines in 2 files covered. (5.88%)

48 existing lines in 1 file now uncovered.

3158 of 4450 relevant lines covered (70.97%)

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
NEW 108	ruleset_name = "conda-forge-branch-protection"	×
109
110	# first, check if the ruleset exists already
NEW 111	rulesets_url = gh_repo.url + "/rulesets"	×
NEW 112	_, ruleset_list = gh_repo._requester.requestJsonAndCheck(	×
113	"GET", rulesets_url
114	)
NEW 115	ruleset_id = None	×
NEW 116	for ruleset in ruleset_list:	×
NEW 117	if ruleset["name"] == ruleset_name:	×
NEW 118	ruleset_id = ruleset["id"]	×
NEW 119	break	×
120
NEW 121	if ruleset_id is not None:	×
NEW 122	print("Updating branch protections")	×
123	# update ruleset
NEW 124	method = "PUT"	×
NEW 125	url = f"{rulesets_url}/{ruleset_id}"	×
126	else:
NEW 127	print("Enabling branch protections")	×
128	# new ruleset
NEW 129	method = "POST"	×
NEW 130	url = rulesets_url	×
131
NEW 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
UNCOV 156	feedstock_name = get_feedstock_name_from_meta(metadata)	×
157
158	gh = Github(token)	×
UNCOV 159	user_or_org = None	×
160	is_conda_forge = False	×
UNCOV 161	if args.user is not None:	×
162	pass
163	# User has been defined, and organization has not.
UNCOV 164	user_or_org = gh.get_user()	×
165	else:
166	# Use the organization provided.
UNCOV 167	user_or_org = gh.get_organization(args.organization)	×
UNCOV 168	if args.organization == "conda-forge":	×
UNCOV 169	is_conda_forge = True	×
170
UNCOV 171	repo_name = f"{feedstock_name}-feedstock"	×
UNCOV 172	try:	×
UNCOV 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
UNCOV 180	if is_conda_forge:	×
181	_conda_forge_specific_repo_setup(gh_repo)	×
182
UNCOV 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	×
UNCOV 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)	×
UNCOV 195	remote_name = args.remote_name.strip()	×
UNCOV 196	if remote_name:	×
197	if remote_name in [remote.name for remote in repo.remotes]:	×
UNCOV 198	existing_remote = repo.remotes[remote_name]	×
UNCOV 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
UNCOV 207	if args.extra_admin_users is not None:	×
UNCOV 208	for user in args.extra_admin_users:	×
UNCOV 209	gh_repo.add_to_collaborators(user, "admin")	×
210
UNCOV 211	if args.add_teams:	×
UNCOV 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✔
UNCOV 219	user = gh.get_user()	×
UNCOV 220	invitations = github.PaginatedList.PaginatedList(	×
221	github.Invitation.Invitation,
222	user._requester,
223	user.url + "/repository_invitations",
224	None,
225	)
UNCOV 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", []))	×
UNCOV 261	maintainers = set(maintainer.lower() for maintainer in maintainers)	×
UNCOV 262	maintainer_teams = set(m for m in maintainers if "/" in m)	×
UNCOV 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	×
UNCOV 267	current_maintainer_teams = list(gh_repo.get_teams())	×
UNCOV 268	fs_team = next(	×
269	(team for team in current_maintainer_teams if team.name == team_name),
270	None,
271	)
UNCOV 272	current_maintainers = set()	×
UNCOV 273	if not fs_team:	×
UNCOV 274	fs_team = create_team(	×
275	org,
276	team_name,
277	f"The {choice(superlative)} {team_name} contributors!",
278	)
UNCOV 279	fs_team.add_to_repos(gh_repo)	×
280
UNCOV 281	current_maintainers = set([e.login.lower() for e in fs_team.get_members()])	×
282
283	# Get the all-members team
UNCOV 284	description = f"All of the awesome {org.login} contributors!"	×
UNCOV 285	all_members_team = get_cached_team(org, "all-members", description)	×
UNCOV 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.
UNCOV 290	for new_maintainer in maintainers - current_maintainers:	×
UNCOV 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).
UNCOV 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
UNCOV 303	maintainer_teams = set(	×
304	m.split("/")[1]
305	for m in maintainer_teams
306	if m.startswith(str(org.login))
307	)
UNCOV 308	current_maintainer_team_objs = {	×
309	team.slug: team for team in current_maintainer_teams
310	}
UNCOV 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:	×
UNCOV 315	team = org.get_team_by_slug(new_team)	×
UNCOV 316	team.add_to_repos(gh_repo)	×
317
318	# remove any old teams
319	if remove:	×
UNCOV 320	for old_team in current_maintainer_teams - maintainer_teams:	×
UNCOV 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 / 11259992998

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