13228537945

Committed 09 Feb 2025 06:34PM UTC coverage: 35.131%. Remained the same

Build # 13228537945

Build Type

Pull #644

github

Committed by

web-flow

Commit Message

Merge 8b3aeb3fa into 16f1fc1f8

Pull Request Pull Request #644: Ignore archived projects

Run Details

437 of 1188 branches covered (36.78%)

645 of 1836 relevant lines covered (35.13%)

3.48 hits per line

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

0.0

/tools/copy_maintainers.py

# -*- coding: utf-8 -*-
# License AGPLv3 (https://www.gnu.org/licenses/agpl-3.0-standalone.html)
"""
oca-copy-maintainers

Copy the users from the teams configured on community.odoo.com to the
GitHub teams

"""

from __future__ import absolute_import, print_function

import argparse
import errno
import os
import sys
from operator import attrgetter

from . import colors, github_login, odoo_login

COPY_USERS_BLACKLIST = os.environ.get(
    "COPY_USERS_BLACKLIST", "~/.config/oca-copy-maintainers/copy_users_blacklist.txt"
)


class FakeProject(object):
    """mock project to represent the 'CLA' team"""

    def __init__(self, name, members):
        self.name = name
        self._members = members

    @property
    def user_id(self):
        return False
        return self._members[0] if self._members else False

    @property
    def members(self):
        return self._members[1:]


def get_cla_project(odoo):
    Partner = odoo.model("res.partner")
    domain = [
        ("github_name", "!=", False),
        "|",
        ("category_id.name", "in", ("ECLA", "ICLA")),
        ("parent_id.category_id.name", "=", "ECLA"),
    ]
    members = Partner.browse(domain)
    return FakeProject("OCA Contributors", members)


class GHTeamList(object):
    def __init__(self, gh_cnx=None, org="oca", dry_run=False):
        if gh_cnx is None:
            gh_cnx = github_login.login()
        self._gh = gh_cnx
        self._org = self._gh.organization("oca")
        self._load_teams()
        self.dry_run = dry_run

    def _load_teams(self):
        self._teams = {t.name: t for t in self._org.teams()}

    def get_project_team(self, project):
        team = self._teams.get(project.name)
        if team:
            return team
        team = self._teams.get(project.name + " Maintainers")
        if team:
            return team
        team = self._teams.get("Local " + project.name + " Maintainers")
        return team

    def get_project_psc_team(self, project):
        main_team = self.get_project_team(project)
        if not main_team:
            return None
        name = main_team.name + " PSC Representative"
        team = self._teams.get(name)
        if team is None and main_team is not None:
            team = self.create_psc_team(project, name, main_team)
        # sync repositories
        if team:
            for repo in main_team.repositories():
                repo_name = "%s/%s" % (repo.owner.login, repo.name)
                if not team.has_repository(repo_name):
                    if not self.dry_run:
                        status = team.add_repository(repo_name)
                    else:
                        status = False
                    print(
                        "Added repo %s to team %s -> %s"
                        % (repo_name, team.name, "OK" if status else "NOK")
                    )
        if team:
            print(list(r.name for r in team.repositories()))
        else:
            print("no team found for project %", project)
        return team

    def create_psc_team(self, project, team_name, main_team):
        repo_names = [
            "%s/%s" % (r.owner.login, r.name) for r in main_team.repositories()
        ]
        if not self.dry_run:
            self._org.create_team(name=team_name, repo_names=repo_names)
        self._load_teams()
        return self._teams.get(team_name)


def get_members_project(odoo):
    Partner = odoo.model("res.partner")
    domain = [
        ("github_name", "!=", False),
        ("membership_state", "in", ("paid", "free")),
    ]
    members = Partner.browse(domain)
    return FakeProject("OCA Members", members)


def get_copy_users_blacklist(filename=COPY_USERS_BLACKLIST):
    """read the blacklist file, if it exists (tries to create one if none).
    File format 1 github login per line, comments begin with # until end of line
    """
    try:
        fobj = open(filename)
    except IOError as exc:
        if exc.errno == errno.ENOENT:  # no such file or directories
            os.makedirs(os.path.dirname(filename))
            fobj = open(filename, "w")
            fobj.close()
            fobj = open(filename)
        else:
            raise
    blacklist = set()
    for line in fobj:
        login = line.split("#", 1)[0].strip().lower()
        blacklist.add(login)
    return blacklist


def copy_users(odoo, team=None, dry_run=False):
    gh = github_login.login()

    # on odoo, the model is a project, but they are teams on GitHub
    Project = odoo.model("project.project")
    base_domain = [
        ("privacy_visibility", "!=", "followers"),
    ]
    if team == "OCA Contributors":
        projects = [get_cla_project(odoo)]
    elif team == "OCA Members":
        projects = [get_members_project(odoo)]
    elif team:
        domain = [("name", "=", team)] + base_domain
        projects = Project.browse(domain)
        if not projects:
            sys.exit("Project %s not found. (%s)" % (team, domain))
    else:
        projects = list(Project.browse(base_domain))
        projects.append(get_cla_project(odoo))
        projects.append(get_members_project(odoo))
    github_teams = GHTeamList(gh, org="oca", dry_run=dry_run)
    valid = []
    not_found = []
    for odoo_project in sorted(projects, key=attrgetter("name")):
        team = github_teams.get_project_team(odoo_project)
        if team and odoo_project.user_id:
            psc_team = github_teams.get_project_psc_team(odoo_project)
        else:
            psc_team = False
        if team:
            valid.append((odoo_project, team, psc_team))
        else:
            not_found.append(odoo_project)

    black_list = get_copy_users_blacklist()

    no_github_login = set()
    for odoo_project, github_team, psc_team in valid:
        print()
        print('Syncing project "%s"' % odoo_project.name)
        psc_users = [odoo_project.user_id] if odoo_project.user_id else []
        users = psc_users + list(odoo_project.members)
        user_logins = set(
            [
                "oca-transbot",
                "OCA-git-bot",
                "oca-travis",
            ]
        )
        psc_user_logins = set()
        for user in users:
            if user.github_name:
                if user.github_name.lower() not in black_list:
                    user_logins.add(user.github_name)
            else:
                no_github_login.add("%s (%s)" % (user.name, user.login))
        for user in psc_users:
            if user.github_name and user.github_name not in black_list:
                psc_user_logins.add(user.github_name)
        sync_team(github_team, user_logins, dry_run)
        if psc_team:
            sync_team(psc_team, psc_user_logins, dry_run)

    if no_github_login:
        print()
        print("Following users miss GitHub login:")
        print(colors.FAIL + "\n".join(no_github_login) + colors.ENDC)

    if not_found:
        print()
        print("The following Odoo projects have no team in GitHub:")
        for project in not_found:
            print(project.name)


def sync_team(team, logins, dry_run=False):
    print(team.name)
    current_logins = set(user.login for user in team.members())

    keep_logins = logins.intersection(current_logins)
    remove_logins = current_logins - logins
    add_logins = logins - current_logins
    print("Add   ", (colors.GREEN + ", ".join(add_logins) + colors.ENDC))
    print("Keep  ", ", ".join(keep_logins))
    print("Remove", (colors.FAIL + ", ".join(remove_logins) + colors.ENDC))
    if not dry_run:
        for login in add_logins:
            try:
                team.add_or_update_membership(login)
            except Exception as exc:
                print("Failed to invite %s: %s" % (login, exc))
        for login in remove_logins:
            try:
                team.remove_member(login)
            except Exception as exc:
                print("Failed to remove %s: %s" % (login, exc))


def main():
    parser = argparse.ArgumentParser(parents=[odoo_login.get_parser()])
    group = parser.add_argument_group("Copy maintainers options")
    group.add_argument("-t", "--team", help="Name of the team to synchronize.")
    group.add_argument(
        "-n",
        "--dry-run",
        action="store_true",
        help="Prints the actions to do, " "but does not apply them",
    )
    args = parser.parse_args()

    odoo = odoo_login.login(args.username, args.store)
    copy_users(odoo, team=args.team, dry_run=args.dry_run)


if __name__ == "__main__":
    main()

1	# -- coding: utf-8 --
2	# License AGPLv3 (https://www.gnu.org/licenses/agpl-3.0-standalone.html)
3	"""
4	oca-copy-maintainers
5
6	Copy the users from the teams configured on community.odoo.com to the
7	GitHub teams
8
9	"""
10
11	from __future__ import absolute_import, print_function	×
12
13	import argparse	×
14	import errno	×
15	import os	×
16	import sys	×
17	from operator import attrgetter	×
18
19	from . import colors, github_login, odoo_login	×
20
21	COPY_USERS_BLACKLIST = os.environ.get(	×
22	"COPY_USERS_BLACKLIST", "~/.config/oca-copy-maintainers/copy_users_blacklist.txt"
23	)
24
25
26	class FakeProject(object):	×
27	"""mock project to represent the 'CLA' team"""
28
29	def __init__(self, name, members):	×
30	self.name = name	×
31	self._members = members	×
32
33	@property	×
34	def user_id(self):
35	return False	×
36	return self._members[0] if self._members else False
37
38	@property	×
39	def members(self):
40	return self._members[1:]	×
41
42
43	def get_cla_project(odoo):	×
44	Partner = odoo.model("res.partner")	×
45	domain = [	×
46	("github_name", "!=", False),
47	"\|",
48	("category_id.name", "in", ("ECLA", "ICLA")),
49	("parent_id.category_id.name", "=", "ECLA"),
50	]
51	members = Partner.browse(domain)	×
52	return FakeProject("OCA Contributors", members)	×
53
54
55	class GHTeamList(object):	×
56	def __init__(self, gh_cnx=None, org="oca", dry_run=False):	×
57	if gh_cnx is None:	×
58	gh_cnx = github_login.login()	×
59	self._gh = gh_cnx	×
60	self._org = self._gh.organization("oca")	×
61	self._load_teams()	×
62	self.dry_run = dry_run	×
63
64	def _load_teams(self):	×
65	self._teams = {t.name: t for t in self._org.teams()}	×
66
67	def get_project_team(self, project):	×
68	team = self._teams.get(project.name)	×
69	if team:	×
70	return team	×
71	team = self._teams.get(project.name + " Maintainers")	×
72	if team:	×
73	return team	×
74	team = self._teams.get("Local " + project.name + " Maintainers")	×
75	return team	×
76
77	def get_project_psc_team(self, project):	×
78	main_team = self.get_project_team(project)	×
79	if not main_team:	×
80	return None	×
81	name = main_team.name + " PSC Representative"	×
82	team = self._teams.get(name)	×
83	if team is None and main_team is not None:	×
84	team = self.create_psc_team(project, name, main_team)	×
85	# sync repositories
86	if team:	×
87	for repo in main_team.repositories():	×
88	repo_name = "%s/%s" % (repo.owner.login, repo.name)	×
89	if not team.has_repository(repo_name):	×
90	if not self.dry_run:	×
91	status = team.add_repository(repo_name)	×
92	else:
93	status = False	×
94	print(	×
95	"Added repo %s to team %s -> %s"
96	% (repo_name, team.name, "OK" if status else "NOK")
97	)
98	if team:	×
99	print(list(r.name for r in team.repositories()))	×
100	else:
101	print("no team found for project %", project)	×
102	return team	×
103
104	def create_psc_team(self, project, team_name, main_team):	×
105	repo_names = [	×
106	"%s/%s" % (r.owner.login, r.name) for r in main_team.repositories()
107	]
108	if not self.dry_run:	×
109	self._org.create_team(name=team_name, repo_names=repo_names)	×
110	self._load_teams()	×
111	return self._teams.get(team_name)	×
112
113
114	def get_members_project(odoo):	×
115	Partner = odoo.model("res.partner")	×
116	domain = [	×
117	("github_name", "!=", False),
118	("membership_state", "in", ("paid", "free")),
119	]
120	members = Partner.browse(domain)	×
121	return FakeProject("OCA Members", members)	×
122
123
124	def get_copy_users_blacklist(filename=COPY_USERS_BLACKLIST):	×
125	"""read the blacklist file, if it exists (tries to create one if none).
126	File format 1 github login per line, comments begin with # until end of line
127	"""
128	try:	×
129	fobj = open(filename)	×
130	except IOError as exc:	×
131	if exc.errno == errno.ENOENT: # no such file or directories	×
132	os.makedirs(os.path.dirname(filename))	×
133	fobj = open(filename, "w")	×
134	fobj.close()	×
135	fobj = open(filename)	×
136	else:
137	raise	×
138	blacklist = set()	×
139	for line in fobj:	×
140	login = line.split("#", 1)[0].strip().lower()	×
141	blacklist.add(login)	×
142	return blacklist	×
143
144
145	def copy_users(odoo, team=None, dry_run=False):	×
146	gh = github_login.login()	×
147
148	# on odoo, the model is a project, but they are teams on GitHub
149	Project = odoo.model("project.project")	×
150	base_domain = [	×
151	("privacy_visibility", "!=", "followers"),
152	]
153	if team == "OCA Contributors":	×
154	projects = [get_cla_project(odoo)]	×
155	elif team == "OCA Members":	×
156	projects = [get_members_project(odoo)]	×
157	elif team:	×
158	domain = [("name", "=", team)] + base_domain	×
159	projects = Project.browse(domain)	×
160	if not projects:	×
161	sys.exit("Project %s not found. (%s)" % (team, domain))	×
162	else:
163	projects = list(Project.browse(base_domain))	×
164	projects.append(get_cla_project(odoo))	×
165	projects.append(get_members_project(odoo))	×
166	github_teams = GHTeamList(gh, org="oca", dry_run=dry_run)	×
167	valid = []	×
168	not_found = []	×
169	for odoo_project in sorted(projects, key=attrgetter("name")):	×
170	team = github_teams.get_project_team(odoo_project)	×
171	if team and odoo_project.user_id:	×
172	psc_team = github_teams.get_project_psc_team(odoo_project)	×
173	else:
174	psc_team = False	×
175	if team:	×
176	valid.append((odoo_project, team, psc_team))	×
177	else:
178	not_found.append(odoo_project)	×
179
180	black_list = get_copy_users_blacklist()	×
181
182	no_github_login = set()	×
183	for odoo_project, github_team, psc_team in valid:	×
184	print()	×
185	print('Syncing project "%s"' % odoo_project.name)	×
186	psc_users = [odoo_project.user_id] if odoo_project.user_id else []	×
187	users = psc_users + list(odoo_project.members)	×
188	user_logins = set(	×
189	[
190	"oca-transbot",
191	"OCA-git-bot",
192	"oca-travis",
193	]
194	)
195	psc_user_logins = set()	×
196	for user in users:	×
197	if user.github_name:	×
198	if user.github_name.lower() not in black_list:	×
199	user_logins.add(user.github_name)	×
200	else:
201	no_github_login.add("%s (%s)" % (user.name, user.login))	×
202	for user in psc_users:	×
203	if user.github_name and user.github_name not in black_list:	×
204	psc_user_logins.add(user.github_name)	×
205	sync_team(github_team, user_logins, dry_run)	×
206	if psc_team:	×
207	sync_team(psc_team, psc_user_logins, dry_run)	×
208
209	if no_github_login:	×
210	print()	×
211	print("Following users miss GitHub login:")	×
212	print(colors.FAIL + "\n".join(no_github_login) + colors.ENDC)	×
213
214	if not_found:	×
215	print()	×
216	print("The following Odoo projects have no team in GitHub:")	×
217	for project in not_found:	×
218	print(project.name)	×
219
220
221	def sync_team(team, logins, dry_run=False):	×
222	print(team.name)	×
223	current_logins = set(user.login for user in team.members())	×
224
225	keep_logins = logins.intersection(current_logins)	×
226	remove_logins = current_logins - logins	×
227	add_logins = logins - current_logins	×
228	print("Add ", (colors.GREEN + ", ".join(add_logins) + colors.ENDC))	×
229	print("Keep ", ", ".join(keep_logins))	×
230	print("Remove", (colors.FAIL + ", ".join(remove_logins) + colors.ENDC))	×
231	if not dry_run:	×
232	for login in add_logins:	×
233	try:	×
234	team.add_or_update_membership(login)	×
235	except Exception as exc:	×
236	print("Failed to invite %s: %s" % (login, exc))	×
237	for login in remove_logins:	×
238	try:	×
239	team.remove_member(login)	×
240	except Exception as exc:	×
241	print("Failed to remove %s: %s" % (login, exc))	×
242
243
244	def main():	×
245	parser = argparse.ArgumentParser(parents=[odoo_login.get_parser()])	×
246	group = parser.add_argument_group("Copy maintainers options")	×
247	group.add_argument("-t", "--team", help="Name of the team to synchronize.")	×
248	group.add_argument(	×
249	"-n",
250	"--dry-run",
251	action="store_true",
252	help="Prints the actions to do, " "but does not apply them",
253	)
254	args = parser.parse_args()	×
255
256	odoo = odoo_login.login(args.username, args.store)	×
257	copy_users(odoo, team=args.team, dry_run=args.dry_run)	×
258
259
260	if __name__ == "__main__":	×
261	main()	×

OCA / maintainer-tools / 13228537945

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