🚧 Refactoring of the GogsService & adaptation on new features.

- extraction of all GogsApi code into an adapter subclass
- removal of custom features to use the RepositoryServices equivalents

Fixes: #18
Signed-off-by: Guyzmo <guyzmo+github+pub@m0g.net>

Author: guyzmo

git_repo/services/ext/gogs.py

@@ -6,51 +6,94 @@
 from ..service import register_target, RepositoryService, os
 from ...exceptions import ResourceError, ResourceExistsError, ResourceNotFoundError
 
-import gogs_client
-import requests
+from gogs_client import GogsApi, GogsRepo, Token, UsernamePassword, ApiFailure
+from requests import Session, HTTPError
 from urllib.parse import urlparse, urlunparse
 import functools
 
 from git import config as git_config
 from git.exc import GitCommandError
 
+class GogsClient(GogsApi):
+    def __init__(self, *args, **kwarg):
+        self.session = Session()
+        super().__init__(*args, session=self.session, **kwarg)
+
+    def set_token(self, token):
+        self.auth = Token(token)
+
+    def set_default_private(self, p):
+        self.default_private = p
+
+    def setup_session(self, ssl_config, proxy=dict()):
+        self.session.verify = ssl_config
+        self.session.proxies.update(proxy)
+
+    @property
+    def username(self):
+        if not hasattr(self, '_username'):
+            self._username = self.authenticated_user(self.auth).username
+        return self._username
+
+    def orgs(self):
+        orgs = self._check_ok(self._get('/user/orgs', auth=self.auth)).json()
+        #return [gogs_client.GogsUser.from_json(org) for org in orgs]
+        return [org['username'] for org in orgs]
+
+    def create_repository(self, user, repo):
+        if user == self.username:
+            repository = self.create_repo(self.auth, name=repo, private=self.default_private)
+        elif user in self.orgs():
+            data = dict(name=repo, private=self.default_private)
+            response = self._post('/org/{}/repos'.format(user), auth=self.auth, data=data)
+            repository = GogsRepo.from_json(self._check_ok(response).json())
+        else:
+            data = dict(name=repo, private=self.default_private)
+            response = self._post('/admin/users/{}/repos'.format(user), auth=self.auth, data=data)
+            repository = GogsRepo.from_json(self._check_ok(response).json())
+
+    def delete_repository(self, user, repo):
+        return self.delete_repo(self.auth, user, repo)
+
+    def repository(self, user, repo):
+        return self.get_repo(self.auth, user, repo)
+
+    def repositories(self, user):
+        r = self._get('/user/repos', auth=self.auth)
+        repositories = self._check_ok(r).json()
+        repositories = [repo for repo in repositories if repo['owner']['username'] == user]
+        return repositories
+
 @register_target('gg', 'gogs')
 class GogsService(RepositoryService):
     fqdn = 'try.gogs.io'
-    #fqdn = 'http://127.0.0.1:3000'
-    gg = None
 
     def __init__(self, *args, **kwargs):
-        self.session = requests.Session()
-        RepositoryService.__init__(self, *args, **kwargs)
-        self.ensure_init()
-
-    def ensure_init(self):
-        if self.gg is not None:
-            return
         self.url_base, self.fqdn = self._url_parse(self.fqdn)
-        if 'insecure' not in self.config:
-            self.insecure = self.fqdn != 'try.gogs.io'
-        self.session.verify = not self.insecure
-        if 'server-cert' in self.config:
-            self.session.verify = self.config['server-cert']
-        self.default_private = self.config.get('default-private', 'false').lower() not in ('0','no','false')
-        self.ssh_url = self.config.get('ssh-url', None) or self.fqdn
-        if not self.repository:
-            config = git_config.GitConfigParser(os.path.join(os.environ['HOME'], '.gitconfig'), True)
-        else:
-            config = self.repository.config_reader()
-        proxies = {}
-        for scheme in 'http https'.split():
-            proxy = config.get_value(scheme, 'proxy', '')
-            if proxy:
-                proxies[scheme] = proxy
-        self.session.proxies.update(proxies)
-        self.gg = gogs_client.GogsApi(self.url_base, self.session)
-        #if ':' in self._privatekey:
-        #    self.auth = gogs_client.UsernamePassword(*self._privatekey.split(':',1))
-        #else:
-        self.auth = gogs_client.Token(self._privatekey)
+        self.gg = GogsClient(self.url_base)
+
+        super().__init__(*args, **kwargs)
+
+        self.gg.set_token(self._privatekey)
+        self.gg.set_default_private(self.default_create_private)
+        self.gg.setup_session(
+                self.session_certificate or not self.session_insecure,
+                self.session_proxy)
+
+    def connect(self):
+        try:
+            self.username = self.user  # Call to self.gg.authenticated_user()
+        except HTTPError as err:
+            if err.response is not None and err.response.status_code == 401:
+                if not self._privatekey:
+                    raise ConnectionError('Could not connect to GoGS. '
+                                          'Please configure .gitconfig '
+                                          'with your gogs private key.') from err
+                else:
+                    raise ConnectionError('Could not connect to GoGS. '
+                                          'Check your configuration and try again.') from err
+            else:
+                raise err
 
     @classmethod
     def _url_parse(cls, url):
@@ -75,70 +118,31 @@ def url_rw(self):
     @classmethod
     def get_auth_token(cls, login, password, prompt=None):
         import platform
-        name = 'git-repo2 token used on {}'.format(platform.node())
+        name = 'git-repo token used on {}'.format(platform.node())
         if '/' in login:
             url, login = login.rsplit('/', 1)
         else:
             url = input('URL [{}]> '.format(cls.fqdn)) or cls.fqdn
         url_base, fqdn = cls._url_parse(url)
-        gg = gogs_client.GogsApi(url_base)
-        auth = gogs_client.UsernamePassword(login, password)
+        gg = GogsApi(url_base)
+        auth = UsernamePassword(login, password)
         tokens = gg.get_tokens(auth, login)
         tokens = dict((token.name, token.token) for token in tokens)
         if name in tokens:
             return tokens[name]
-        if 'git-repo2 token' in tokens:
-            return tokens['git-repo2 token']
+        if 'git-repo token' in tokens:
+            return tokens['git-repo token']
         token = gg.create_token(auth, name, login)
         return token.token
 
     @property
     def user(self):
-        return self.gg.authenticated_user(self.auth).username
-
-    def orgs(self):
-        orgs = self.gg._check_ok(self.gg._get('/user/orgs', auth=self.auth)).json()
-        #return [gogs_client.GogsUser.from_json(org) for org in orgs]
-        return [org['username'] for org in orgs]
-
-    def connect(self):
-        self.ensure_init()
-        try:
-            if self.insecure:
-                try:
-                    try:
-                        urllib3 = requests.packages.urllib3
-                    except Exception:
-                        import urllib3
-                    urllib3.disable_warnings()
-                except ImportError:
-                    pass
-            self.username = self.user  # Call to self.gg.authenticated_user()
-        except requests.HTTPError as err:
-            if err.response is not None and err.response.status_code == 401:
-                if not self._privatekey:
-                    raise ConnectionError('Could not connect to GoGS. '
-                                          'Please configure .gitconfig '
-                                          'with your gogs private key.') from err
-                else:
-                    raise ConnectionError('Could not connect to GoGS. '
-                                          'Check your configuration and try again.') from err
-            else:
-                raise err
+        return self.gg.username
 
     def create(self, user, repo, add=False):
         try:
-            if user == self.username:
-                repository = self.gg.create_repo(self.auth, name=repo, private=self.default_private)
-            elif user in self.orgs():
-                data = dict(name=repo, private=self.default_private)
-                response = self.gg._post('/org/{}/repos'.format(user), auth=self.auth, data=data)
-                repository = gogs_client.GogsRepo.from_json(self.gg._check_ok(response).json())
-            else:
-                data = dict(name=repo, private=self.default_private)
-                response = self.gg._post('/admin/users/{}/repos'.format(user), auth=self.auth, data=data)
-                repository = gogs_client.GogsRepo.from_json(self.gg._check_ok(response).json())
-        except gogs_client.ApiFailure as err:
+            self.gg.create_repository(user, repo)
+        except ApiFailure as err:
             if err.status_code == 422:
                 raise ResourceExistsError("Project already exists.") from err
             else:
@@ -155,8 +159,8 @@ def delete(self, repo, user=None):
         if not user:
             user = self.username
         try:
-            self.gg.delete_repo(self.auth, user, repo)
-        except gogs_client.ApiFailure as err:
+            self.gg.delete_repository(user, repo)
+        except ApiFailure as err:
             if err.status_code == 404:
                 raise ResourceNotFoundError("Cannot delete: repository {}/{} does not exists.".format(user, repo)) from err
             elif err.status_code == 403:
@@ -190,9 +194,7 @@ def col_print(lines, indent=0, pad=2):
             for row in rows:
                 print(" "*indent + (" "*pad).join(line.ljust(col_width) for line in row))
 
-        r = self.gg._get('/user/repos', auth=self.auth)
-        repositories = self.gg._check_ok(r).json()
-        repositories = [repo for repo in repositories if repo['owner']['username'] == user]
+        repositories = self.gg.repositories(user)
         if user != self.username and not repositories and user not in self.orgs:
             raise ResourceNotFoundError("Unable to list namespace {} - only authenticated user and orgs available for listing.".format(user))
         if not _long:
@@ -227,8 +229,8 @@ def col_print(lines, indent=0, pad=2):
 
     def get_repository(self, user, repo):
         try:
-            return self.gg.get_repo(self.auth, user, repo)
-        except gogs_client.ApiFailure as err:
+            return self.gg.repository(user, repo)
+        except ApiFailure as err:
             if err.status_code == 404:
                 raise ResourceNotFoundError("Cannot get: repository {}/{} does not exists.".format(user, repo)) from err
             raise ResourceError("Unhandled error: {}".format(err)) from err

tests/integration/test_gogs.py

@@ -34,7 +34,7 @@ def get_service(self):
         # return gogs.GogsService(c=dict(fqdn=self.fqdn,__name__='gitrepo "gogs"',token=os.environ['PRIVATE_KEY_GOGS']))
 
     def get_requests_session(self):
-        return self.service.session
+        return self.service.gg.session
 
     def test_00_fork(self):
         pass