15637235389

Committed 13 Jun 2025 02:40PM UTC coverage: 72.416%. Remained the same

Build # 15637235389

Build Type

Pull #2333

github

Committed by

web-flow

Commit Message

Merge e6094fbeb into 1d22535c7

Pull Request Pull Request #2333: Use black's native linelength (Option A)

Run Details

1330 of 1998 branches covered (66.57%)

110 of 155 new or added lines in 17 files covered. (70.97%)

32 existing lines in 9 files now uncovered.

3510 of 4847 relevant lines covered (72.42%)

0.72 hits per line

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

13.04

/conda_smithy/github.py

import os
from random import choice

import github
import pygit2
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,
    file_permissions,
    get_feedstock_name_from_meta,
)


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

conda-forge / conda-smithy / 15637235389

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