Merge branch 'Langenfeld:master' into master

gitea/apiobject.py

@@ -1,11 +1,35 @@
 import logging
 from datetime import datetime
 from typing import List, Tuple, Dict, Sequence, Optional, Union, Set
-
+from dataclasses import dataclass, fields
 from .baseapiobject import ReadonlyApiObject, ApiObject
 from .exceptions import *
 
 
+@dataclass(frozen=True)
+class RepoUnits:
+    code: str = "none"
+    issues: str = "none"
+    ext_issues: str = "none"
+    wiki: str = "none"
+    pulls: str = "none"
+    releases: str = "none"
+    ext_wiki: str = "none"
+
+    def to_dict(self) -> dict[str, str]:
+        """Return the correctly prefixed (added "repo.") representation for gitea Repository runit Rights"""
+        return {
+            f"repo.{field.name}": getattr(self, field.name) for field in fields(self)
+        }
+
+    @classmethod
+    def from_dict(cls, unit_dict: dict[str, str]) -> "RepoUnits":
+        """Parse all known repo units from the dictionary returned by the api"""
+        return RepoUnits(
+            **{k[5:]: v for k, v in unit_dict.items() if k[5:] in fields(cls)}
+        )
+
+
 class Organization(ApiObject):
     """see https://try.gitea.io/api/swagger#/organization/orgGetAll"""
 
@@ -119,10 +143,12 @@ class Organization(ApiObject):
             setattr(t, "_organization", self)
         return teams
 
-    def get_team(self, name) -> "Team":
+    def get_team(self, name, ignore_case: bool = False) -> "Team":
         teams = self.get_teams()
         for team in teams:
-            if team.name == name:
+            if (not ignore_case and team.name == name) or (
+                ignore_case and team.name.lower() == name.lower()
+            ):
                 return team
         raise NotFoundException("Team not existent in organization.")
 
@@ -344,6 +370,8 @@ class Repository(ApiObject):
     REPO_ISSUES = """/repos/{owner}/{repo}/issues"""  # <owner, reponame>
     REPO_DELETE = """/repos/%s/%s"""  # <owner>, <reponame>
     REPO_TIMES = """/repos/%s/%s/times"""  # <owner>, <reponame>
+    REPO_TOPICS = """/repos/%s/%s/topics"""  # <owner, <reponame>
+    REPO_TOPIC = """/repos/%s/%s/topics/%s"""  # <owner, <reponame>, <topicname>
     REPO_USER_TIME = """/repos/%s/%s/times/%s"""  # <owner>, <reponame>, <username>
     REPO_COMMITS = "/repos/%s/%s/commits"  # <owner>, <reponame>
     REPO_TRANSFER = "/repos/{owner}/{repo}/transfer"
@@ -432,7 +460,7 @@ class Repository(ApiObject):
         try:
             results = self.gitea.requests_get_paginated(
                 Repository.REPO_COMMITS % (self.owner.username, self.name),
-                page_limit= page_limit
+                page_limit=page_limit,
             )
         except ConflictException as err:
             logging.warning(err)
@@ -465,6 +493,24 @@ class Repository(ApiObject):
         )
         return results
 
+    def get_topics(self) -> list[str]:
+        results = self.gitea.requests_get(
+            Repository.REPO_TOPICS % (self.owner.username, self.name)
+        )
+        return results["topics"]
+
+    def add_topic(self, topic: str):
+        """Add a topic to the repository"""
+        self.gitea.requests_put(
+            Repository.REPO_TOPIC % (self.owner.username, self.name, topic)
+        )
+
+    def del_topic(self, topic: str):
+        """Delete a topic to the repository"""
+        self.gitea.requests_delete(
+            Repository.REPO_TOPIC % (self.owner.username, self.name, topic)
+        )
+
     def get_user_time(self, username) -> float:
         if isinstance(username, User):
             username = username.username
@@ -930,7 +976,8 @@ class Team(ApiObject):
         return hash(self.organization) ^ hash(self.id)
 
     _fields_to_parsers = {
-        "organization": lambda gitea, o: Organization.parse_response(gitea, o)
+        "organization": lambda gitea, o: Organization.parse_response(gitea, o),
+        "units_map": lambda gitea, o: RepoUnits.from_dict(o),
     }
 
     _patchable_fields = {

gitea/gitea.py

@@ -1,12 +1,11 @@
 import logging
 import json
 from typing import List, Dict, Union
-
 from immutabledict import immutabledict
 import requests
 import urllib3
 
-from .apiobject import User, Organization, Repository, Team
+from .apiobject import User, Organization, Repository, Team, RepoUnits
 from .exceptions import NotFoundException, ConflictException, AlreadyExistsException
 
 
@@ -22,7 +21,14 @@ class Gitea:
     CREATE_TEAM = """/orgs/%s/teams"""  # <orgname>
 
     def __init__(
-        self, gitea_url: str, token_text=None, auth=None, verify=True, log_level="INFO"
+        self,
+        gitea_url: str,
+        token_text=None,
+        auth=None,
+        verify=True,
+        log_level="INFO",
+        # example: "socks5h://127.0.0.1:9050"
+        proxy=None,
     ):
         """Initializing Gitea-instance
 
@@ -43,6 +49,12 @@ class Gitea:
         self.url = gitea_url
         self.requests = requests.Session()
 
+        if proxy:
+            self.requests.proxies = {
+                "http": proxy,
+                "https": proxy,
+            }
+
         # Manage authentification
         if not token_text and not auth:
             raise ValueError("Please provide auth or token_text, but not both")
@@ -352,6 +364,7 @@ class Gitea:
             "repo.releases",
             "repo.ext_wiki",
         ),
+        units_map: "RepoUnits" = RepoUnits(),
     ):
         """Creates a Team.
 
@@ -370,6 +383,7 @@ class Gitea:
                 "can_create_org_repo": can_create_org_repo,
                 "includes_all_repositories": includes_all_repositories,
                 "units": units,
+                "units_map": units_map.to_dict(),
             },
         )
         if "id" in result:

requirements.txt

@@ -1,3 +1,3 @@
-requests
+requests[socks]
 pytest
 immutabledict

tests/test_api.py

@@ -348,6 +348,19 @@ def test_team_get_org(instance):
     assert org.username == teams[0].organization.name
 
 
+def test_topic_functions(instance):
+    user = User.request(instance, test_user)
+    repo = Repository.request(instance, user.username, test_repo)
+    repo.add_topic("rings")
+    repo.add_topic("swords")
+    repo.add_topic("dragons")
+    assert "swords" in repo.get_topics()
+    repo.del_topic("swords")
+    assert "swords" not in repo.get_topics()
+    assert "dragons" in repo.get_topics()
+    assert "rings" in repo.get_topics()
+
+
 def test_delete_repo_userowned(instance):
     user = User.request(instance, test_user)
     repo = Repository.request(instance, user.username, test_repo)