10161783611

Committed 30 Jul 2024 11:50AM UTC coverage: 70.749% (+0.1%) from 70.605%

Build # 10161783611

Build Type

Pull #1968

github

Committed by

web-flow

Commit Message

Merge 2f56c9670 into 77a83f8db

Pull Request Pull Request #1968: reduce input variants to only _used_ input variants

Run Details

1362 of 2048 branches covered (66.5%)

25 of 27 new or added lines in 1 file covered. (92.59%)

124 existing lines in 1 file now uncovered.

3033 of 4287 relevant lines covered (70.75%)

0.71 hits per line

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

14.93

/conda_smithy/github.py

import os
from random import choice

import conda_build.api
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.utils import 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()
    meta = conda_build.api.render(
        args.feedstock_directory,
        permit_undefined_jinja=True,
        finalize=False,
        bypass_env_check=True,
        trim_skip=False,
    )[0][0]

    feedstock_name = get_feedstock_name_from_meta(meta)

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

conda-forge / conda-smithy / 10161783611

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