12031098063

Committed 26 Nov 2024 12:57PM UTC coverage: 71.382%. Remained the same

Build # 12031098063

Build Type

Pull #2167

github

Committed by

web-flow

Commit Message

Merge d01794849 into fed429968

Pull Request Pull Request #2167: allow equal-signs in CFEP-25 pins

Run Details

1210 of 1868 branches covered (64.78%)

3315 of 4644 relevant lines covered (71.38%)

0.71 hits per line

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

13.29

/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,
    get_cached_cfp_file_path,
)
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,
        conda_forge_pinning_file=(
            get_cached_cfp_file_path(".")[0]
            if args.user is None and args.organization == "conda-forge"
            else None
        ),
    )

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

conda-forge / conda-smithy / 12031098063

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