done

Author: Ilanlido

logger.py

@@ -0,0 +1,4 @@
+import logging.config
+
+logging.config.fileConfig('logging.conf')
+logger = logging.getLogger("disaster_recovery")

logging.conf

@@ -0,0 +1,27 @@
+[loggers]
+keys=root, disaster_recovery
+
+[handlers]
+keys=consoleHandler
+
+[formatters]
+keys=simpleFormatter
+
+[logger_disaster_recovery]
+level=DEBUG
+handlers=consoleHandler
+qualname=disaster_recovery
+propagate=0
+
+[handler_consoleHandler]
+class=StreamHandler
+level=DEBUG
+formatter=simpleFormatter
+args=(sys.stdout,)
+
+[formatter_simpleFormatter]
+format=%(asctime)s - %(levelname)s - %(message)s
+
+[logger_root]
+level=ERROR
+handlers=consoleHandler

main.py

@@ -1,17 +1,14 @@
 from typing import Dict
-
-from PyInquirer import prompt, Separator
-import logging
+from PyInquirer import prompt
 from consts import providers, contexts, config_file
+from logger import logger
 from menus.menu_base import MenuBase
 from menus.release_block_pr_menu import ReleaseBlockPrMenu
 
 release_block_pr_menu = ReleaseBlockPrMenu()
 menus = [release_block_pr_menu]
 menu_value_to_class: Dict[str, MenuBase] = {menu.get_menu(): menu for menu in menus}
 
-logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
-logger = logging.getLogger(__name__)
 
 main_menu = [
     {
@@ -50,5 +47,7 @@
     menu = menu_value_to_class.get(answer.get('menu'))
     if menu:
         menu.handle(answer)
+        logger.info("Completed successfully")
     else:
         logger.error(f"invalid menu {answer.get('menu')}")
+

menus/release_block_pr_menu.py

@@ -1,34 +1,31 @@
 import os
-from typing import List, Dict
+from typing import List, Dict, Type
 import pydantic
-import logging
 from consts import github
+from logger import logger
 from menus.menu_base import MenuBase
 from providers.github_handler import GithubHandler
-from providers.provider_handler import ProviderHandler
-import PyInquirer
 from release_block_pr_config import ReleaseBlockPrConfig
 
-logger = logging.getLogger(__name__)
 
 release_block = "Release Blocked PR's"
 
 
 class ReleaseBlockPrMenu(MenuBase):
-    handlers: Dict[str, ProviderHandler] = {
-        github: GithubHandler()
+    handlers: Dict[str, Type[GithubHandler]] = {
+        github: GithubHandler
     }
 
     def handle(self, answers):
         if answers.get("provider") in self.handlers:
-            handler = self.handlers[answers["provider"]]
+            handler_class = self.handlers[answers["provider"]]
             repositories_to_unblock: List[ReleaseBlockPrConfig] = pydantic.parse_file_as(List[ReleaseBlockPrConfig],
                                                                                          answers["config_file"])
             for tokens_to_repositories in repositories_to_unblock:
+                handler = handler_class(tokens_to_repositories.token)
                 for repository in tokens_to_repositories.repositories:
-                    handler.release_branch_protection(tokens_to_repositories.token, repository.organization_name,
-                                                      repository.organization_name, repository.branch,
-                                                      answers["contexts"])
+                    handler.release_branch_protection(repository.organization_name, repository.repository_name,
+                                                      repository.branch, answers["contexts"])
         else:
             logger.error(f"Invalid provider selected: {answers.get('provider')}")
 

providers/github_handler.py

@@ -1,11 +1,30 @@
-import logging
-from typing import List
+from typing import List, Set
 
+from logger import logger
 from providers.provider_handler import ProviderHandler
-
-logger = logging.getLogger(__name__)
+from github import Github
 
 
 class GithubHandler(ProviderHandler):
-    def release_branch_protection(self, token, organization_name, repository_name, branch, contexts: List[str]):
-        logger.info(f"Releasing branch protection on repository {repository_name}")
+    def __init__(self, token):
+        super().__init__(token)
+        self.github: Github = Github(token)
+
+    def release_branch_protection(self, organization_name, repository_name, branch, contexts: List[str]):
+        full_repo_name = f"{organization_name}/{repository_name}"
+        if len(contexts) > 0:
+            try:
+                repo = self.github.get_repo(full_repo_name, lazy=True)
+                branch = repo.get_branch(branch)
+                required_status_checks = branch.get_required_status_checks()
+                current_contexts = set(required_status_checks.contexts)
+                logger.info(f"Current required status checks {current_contexts}")
+                contexts_to_keep = current_contexts.difference(contexts)
+                if contexts_to_keep != current_contexts:
+                    logger.info(
+                        f"Updating required status checks repository {full_repo_name} branch {branch.name} to contexts {contexts_to_keep}")
+                    branch.edit_required_status_checks(contexts=list(contexts_to_keep))
+                else:
+                    logger.info(f"Skipping {full_repo_name} contexts to remove are not required")
+            except:
+                logger.exception(f"Failed to update required status checks on repository {full_repo_name}")

providers/provider_handler.py

@@ -1,8 +1,12 @@
 from abc import ABC, abstractmethod
-from typing import List
+from typing import List, Set
 
 
 class ProviderHandler(ABC):
+
+    def __init__(self, token):
+        self.token = token
+
     @abstractmethod
-    def release_branch_protection(self, token, organization_name, repository_name, branch, contexts: List[str]):
+    def release_branch_protection(self, organization_name, repository_name, branch, contexts: List[str]):
         pass

requirements.txt

@@ -0,0 +1,3 @@
+pydantic==1.10.2
+PyGithub==1.57
+PyInquirer==1.0.3