Merge pull request #17 from lewagon/git-repos

add git and gh helper classes

Author: The-Pavel

.env.sample

@@ -0,0 +1 @@
+GH_API_DELETE_TOKEN=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

.github/workflows/run-tests.yml

@@ -23,5 +23,7 @@ jobs:
           pip install -e .
 
       - name: Install package and test
+        env:
+          GH_API_DELETE_TOKEN: ${{ secrets.AUTOBOT_QA_ADMIN_TOKEN }}
         run: |
           make pytest

.gitignore

@@ -3,3 +3,4 @@ __pycache__
 *.egg-info
 .DS_Store
 dist
+.env

CHANGELOG.md

@@ -1,5 +1,11 @@
 
-# 0.2.9 (2022-11-25)
+# 0.2.10 (2022-11-25)
+
+### Added
+
+- Adds `GitRepo`, `GhOrg`, `GhOrgSecret` and `GhRepo` helper classes
+- Adds tests for the `GhRepo` class
+- Adds `subblack`, `subred`, `subgreen`, `subyellow`, `subblue`, `submagenta`, `subcyan`, and `subwhite` helpers for `colorama`
 
 ### Added
 

Makefile

@@ -5,3 +5,6 @@ pylint:
 
 pytest:
 	PYTHONDONTWRITEBYTECODE=1 pytest -v --color=yes
+
+pytest_output:
+	PYTHONDONTWRITEBYTECODE=1 pytest -v --color=yes -s

doc/drafts/gh_repo.py

@@ -0,0 +1,128 @@
+
+import requests
+
+from wagon_common.helpers.output import red
+
+
+class GhRepo:
+    """
+    helper class for gh api calls
+    """
+
+    base_url = "https://api.github.com"
+    default_path = f"/repos/{self.owner}/{self.repo}"
+
+    def __init__(self, name, token=None, is_org=True, verbose=False):
+        """
+        required gh token scopes:
+        - repo: push commits TBC
+        - admin:org: create repos TBC
+        - workflow: push commits containing `.github/workflows/*.yml` files
+        - delete_repo: delete `lewagon-test` and `lewagon-qa` repositories
+        """
+
+        self.name, self.owner, self.repo = self.__identify(name)
+        self.token = token
+        self.is_org = is_org
+        self.verbose = verbose
+
+    def __identify(self, name):
+        """
+        identify owner and repo from provided name
+        (gh cli nomenclatura)
+        """
+
+        parts = name.split("/", maxsplit=1)
+
+        if len(parts) == 2:
+            owner = parts[0]
+            repo = parts[1]
+        else:
+            owner = "lewagon"
+            repo = name
+            name = f"{owner}/{repo}"
+
+        return name, owner, repo
+
+    def __call(self, path=None, verb="get", headers={}, params={}, context="", status_code=200, decode_response=True):
+        """
+        resolve api call
+        """
+
+        # set path
+        if path is None:
+            path = self.default_path
+
+        # resolve http verb call method
+        call_method = dict(
+            get=requests.get,
+            put=requests.put,
+            patch=requests.patch,
+            post=requests.post,
+            delete=requests.delete)[verb]
+
+        # add auth
+        headers["Authorization"] = f"token {self.token}"
+
+        # list repo params
+        response = call_method(self.base_url + path,
+                               headers=headers,
+                               json=params)
+
+        if response.status_code != status_code:
+
+            red("\nGH api error 🤕",
+                f"\n- context {context}"
+                + f"\n- path: {path}"
+                + f"\n- verb: {verb}"
+                + f"\n- params: {params}"
+                + f"\n- expected status code: {status_code}"
+                + f"\n- status code: {response.status_code}"
+                + f"\n- response: {response.content}")
+
+            raise ValueError("GH api error")
+
+        if decode_response:
+            return response.json()
+
+    def create(self, params={}):
+        """
+        create repo
+        """
+
+        params["org"] = self.owner
+        params["name"] = self.repo
+
+        return self.__call(path=f"/orgs/{self.owner}/repos",
+                           verb="post",
+                           params=params,
+                           status_code=201)
+
+    def get(self):
+        """
+        get repo
+        """
+
+        return self.__call()
+
+    def update(self, params):
+        """
+        update repo
+        """
+
+        return self.__call(verb="patch", params=params)
+
+    def delete(self, dry_run=True):
+        """
+        delete repo
+        """
+
+        # protect production repositories
+        if self.owner.lower() not in ["lewagon-test", "lewagon-qa"]:
+            raise NameError("cannot delete repo in production organisation")
+
+        if dry_run:
+            return {}
+
+        # delete repo
+        return self.__call(verb="delete", status_code=204, decode_response=False)

requirements.txt

@@ -1,3 +1,5 @@
 colorama
 pylint
 pytest
+python-dotenv
+requests

setup.py

@@ -5,7 +5,7 @@
     requirements = [c.strip() for c in f.readlines()]
 
 setup(name="wagon_common",
-      version="0.2.9",
+      version="0.2.10",
       description="Le Wagon common packages",
       url="https://github.com/lewagon/python-utilities/",
       author="Sébastien Saunier",

tests/gh/__init__.py

@@ -0,0 +1,117 @@
+
+from wagon_common.gh.gh_repo import GhRepo
+
+import os
+
+import pytest
+
+from dotenv import load_dotenv, find_dotenv
+
+
+class TestGhRepo():
+
+    @pytest.fixture
+    def token(self):
+        """
+        fetch gh token to perform the tests
+        """
+
+        # Arrange
+        load_dotenv(find_dotenv())
+        token = os.environ.get("GH_API_DELETE_TOKEN")
+
+        # Act & Assert
+        yield token
+
+        # Cleanup
+
+    def test_gh_repo_naming(self):
+
+        # Arrange
+
+        # Act
+        repo = GhRepo("lewagon-test/data-solutions", token=None)
+
+        lw_repo = GhRepo("myriad", token=None)
+
+        # Assert
+        assert repo.owner == "lewagon-test"
+        assert repo.repo == "data-solutions"
+        assert repo.name == "lewagon-test/data-solutions"
+
+        assert repo.ssh_url == f"git@github.com:{repo.owner}/{repo.repo}.git"
+        assert repo.https_url == f"https://None@github.com/{repo.owner}/{repo.repo}.git"
+
+        assert lw_repo.owner == "lewagon"
+        assert lw_repo.repo == "myriad"
+        assert lw_repo.name == "lewagon/myriad"
+
+        # Cleanup
+
+    def test_gh_repo_delete_prod(self, token):
+        """
+        verify that a production repo cannot be deleted
+        """
+
+        exception_catched = False
+
+        try:
+            GhRepo("lewagon/data-solutions", token=token).delete()
+        except NameError as e:
+            exception_catched = True
+            assert "cannot delete repo in" in str(e)
+
+        assert exception_catched
+
+    def test_gh_repo_delete_test_qa(self, token):
+        """
+        verify that a test or QA repo can be deleted
+        """
+
+        # test gh repo can be deleted
+        exception_catched = False
+
+        try:
+            GhRepo("lewagon-test/data-solutions", token=token).delete()
+        except NameError:
+            exception_catched = True
+
+        assert not exception_catched
+
+        # QA gh repo can be deleted
+        exception_catched = False
+
+        try:
+            GhRepo("lewagon-qa/data-solutions", token=token).delete()
+        except NameError:
+            exception_catched = True
+
+        assert not exception_catched
+
+    def test_gh_repo_crud(self, token):
+        """
+        verifies repo crud
+        """
+
+        # Arrange
+
+        # Act
+        repo = GhRepo("lewagon-qa/automated-test", token=token)
+        create_response = repo.create()
+        get_response = repo.get()
+        update_response = repo.update(dict(description="automated update test"))
+        repo.delete(dry_run=False)
+
+        # Assert
+        assert "node_id" in create_response
+        assert "node_id" in get_response
+        assert "node_id" in update_response
+
+        create_node_id = create_response["node_id"]
+        get_node_id = get_response["node_id"]
+        update_node_id = update_response["node_id"]
+
+        assert create_node_id == get_node_id
+        assert create_node_id == update_node_id
+
+        # Cleanup