119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
# Copyright © 2022 Aravinth Manivannan <realaravinth@batsense.net>
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
from .entity import User
|
|
from .factory import Factory
|
|
from .alert import Message, MessageCode
|
|
|
|
|
|
class GroupPolicy:
|
|
"""
|
|
Payment policy to govern a group of actors
|
|
"""
|
|
|
|
def __init__(self, f: Factory):
|
|
self.f = f
|
|
self.validator = f.val
|
|
self.alert = f.alert
|
|
self.forge = f.forge
|
|
|
|
def apply(self):
|
|
"""
|
|
Apply policies to group
|
|
Account paying and non-paying users. Notify admin if non-paying members are
|
|
are exceeding their gratis quota
|
|
"""
|
|
|
|
raise NotImplementedError()
|
|
|
|
|
|
class UserPolicy(GroupPolicy):
|
|
"""
|
|
Payment policy to govern a group of actors, where an actor is a Gitea user
|
|
"""
|
|
|
|
def __init__(self, f: Factory):
|
|
super().__init__(f=f)
|
|
|
|
def apply(self):
|
|
"""
|
|
Account paying and non-paying users. Notify admin if non-paying members are
|
|
are exceeding their gratis quota
|
|
"""
|
|
|
|
users = self.forge.get_all_users()
|
|
for user in users:
|
|
if self.validator.is_paying(user=user):
|
|
continue
|
|
|
|
repos = self.forge.get_all_user_repos(user.username)
|
|
for repo in repos:
|
|
if not repo.is_fork:
|
|
self.alert.alert(
|
|
Message(
|
|
code=MessageCode.USER_NO_PURCHASE_NON_FORK_REPO,
|
|
customer=user,
|
|
repo=repo,
|
|
)
|
|
)
|
|
continue
|
|
|
|
forks = []
|
|
cur_repo = repo
|
|
forks.append(cur_repo)
|
|
|
|
while cur_repo.is_fork:
|
|
parent = self.forge.parent(repo=cur_repo) # cur_repo.parent()
|
|
parent_owner = User(username=parent.owner)
|
|
cur_repo = parent
|
|
forks.append(cur_repo)
|
|
if self.validator.is_paying(user=parent_owner):
|
|
break
|
|
|
|
if not self.validator.is_paying(
|
|
user=self.forge.get_user(cur_repo.owner)
|
|
):
|
|
self.alert.alert(
|
|
Message(
|
|
code=MessageCode.USER_NO_PURCHASE_FORK_HISTORY, forks=forks
|
|
)
|
|
)
|
|
|
|
|
|
class OrgPolicy(GroupPolicy):
|
|
"""
|
|
Payment policy to govern a group of actors, where an actor is a Gitea organisation
|
|
"""
|
|
|
|
def __init__(self, f: Factory):
|
|
super().__init__(f=f)
|
|
|
|
def apply(self):
|
|
"""
|
|
Account paying and non-paying users. Notify admin if non-paying members are
|
|
are exceeding their gratis quota
|
|
"""
|
|
|
|
orgs = self.forge.get_all_orgs()
|
|
for org in orgs:
|
|
members = self.forge.org_members(org_name=org.username) # org.members()
|
|
paying = 0
|
|
for member in members:
|
|
if self.validator.is_paying(user=member):
|
|
paying += 1
|
|
if not paying >= (len(members) / 2):
|
|
self.alert.alert(
|
|
Message(code=MessageCode.ORG_MAJORITY_NO_PURCHASE, org=org)
|
|
)
|