Fix high cyclomatic complexities using Radon

coveralls/api.py

@@ -195,24 +195,24 @@ def load_config_from_ci_environment(self):
         # We load them first and allow more specific vars to overwrite
         self.load_config_from_generic_ci_environment()
 
-        if os.environ.get('APPVEYOR'):
-            name, job, number, pr = self.load_config_from_appveyor()
-        elif os.environ.get('BUILDKITE'):
-            name, job, number, pr = self.load_config_from_buildkite()
-        elif os.environ.get('CIRCLECI'):
-            name, job, number, pr = self.load_config_from_circle()
-        elif os.environ.get('GITHUB_ACTIONS'):
-            name, job, number, pr = self.load_config_from_github()
-        elif os.environ.get('JENKINS_HOME'):
-            name, job, number, pr = self.load_config_from_jenkins()
-        elif os.environ.get('TRAVIS'):
-            self._token_required = False
-            name, job, number, pr = self.load_config_from_travis()
-        elif os.environ.get('SEMAPHORE'):
-            name, job, number, pr = self.load_config_from_semaphore()
+        ci_handlers = {
+            'APPVEYOR': (self.load_config_from_appveyor, None),
+            'BUILDKITE': (self.load_config_from_buildkite, None),
+            'CIRCLECI': (self.load_config_from_circle, None),
+            'GITHUB_ACTIONS': (self.load_config_from_github, None),
+            'JENKINS_HOME': (self.load_config_from_jenkins, None),
+            'TRAVIS': (self.load_config_from_travis, lambda: setattr(self, '_token_required', False)),
+            'SEMAPHORE': (self.load_config_from_semaphore, None),
+        }
+
+        for env_var, (handler, pre_action) in ci_handlers.items():
+            if os.environ.get(env_var):
+                if pre_action:
+                    pre_action()
+                name, job, number, pr = handler()
+                break
         else:
             name, job, number, pr = self.load_config_from_unknown()
-
         self.config.setdefault('service_name', name)
         if job:
             self.config['service_job_id'] = job

coveralls/cli.py

@@ -47,62 +47,70 @@
 
 log = logging.getLogger('coveralls')
 
+def handle_debug(coverallz):
+    log.info('Testing coveralls-python...')
+    coverallz.wear(dry_run=True)
+
+def handle_output(coverallz, output_file):
+    log.info('Write coverage report to file...')
+    coverallz.save_report(output_file)
+
+def handle_submit(coverallz, submit_file):
+    with open(submit_file) as report_file:
+        coverallz.submit_report(report_file.read())
+
+def handle_finish(coverallz):
+    log.info('Finishing parallel jobs...')
+    coverallz.parallel_finish()
+    log.info('Done')
+
+def handle_default(coverallz):
+    log.info('Submitting coverage to coveralls.io...')
+    result = coverallz.wear()
+    log.info('Coverage submitted!')
+    log.debug(result)
+    if result:
+        log.info(result.get('message'))
+        log.info(result.get('url'))
+def setup_coveralls(options):
+    level = logging.DEBUG if options['--verbose'] else logging.INFO
+    log.addHandler(logging.StreamHandler())
+    log.setLevel(level)
 
+    token_required = not options['debug'] and not options['--output']
+
+    return Coveralls(
+        token_required,
+        config_file=options['--rcfile'],
+        service_name=options['--service'],
+        base_dir=options.get('--basedir') or '',
+        src_dir=options.get('--srcdir') or '',
+    )
 def main(argv=None):
-    # pylint: disable=too-complex
     version = importlib.metadata.version('coveralls')
     options = docopt.docopt(__doc__, argv=argv, version=version)
     if options['debug']:
         options['--verbose'] = True
 
-    level = logging.DEBUG if options['--verbose'] else logging.INFO
-    log.addHandler(logging.StreamHandler())
-    log.setLevel(level)
+    try:
+        coverallz = setup_coveralls(options)
 
-    token_required = not options['debug'] and not options['--output']
+        action_handlers = {
+            '--merge': lambda: coverallz.merge(options['--merge']),
+            'debug': lambda: handle_debug(coverallz),
+            '--output': lambda: handle_output(coverallz, options['--output']),
+            '--submit': lambda: handle_submit(coverallz, options['--submit']),
+            '--finish': lambda: handle_finish(coverallz),
+        }
 
-    try:
-        coverallz = Coveralls(
-            token_required,
-            config_file=options['--rcfile'],
-            service_name=options['--service'],
-            base_dir=options.get('--basedir') or '',
-            src_dir=options.get('--srcdir') or '',
-        )
-
-        if options['--merge']:
-            coverallz.merge(options['--merge'])
-
-        if options['debug']:
-            log.info('Testing coveralls-python...')
-            coverallz.wear(dry_run=True)
-            return
-
-        if options['--output']:
-            log.info('Write coverage report to file...')
-            coverallz.save_report(options['--output'])
-            return
-
-        if options['--submit']:
-            with open(options['--submit']) as report_file:
-                coverallz.submit_report(report_file.read())
-            return
-
-        if options['--finish']:
-            log.info('Finishing parallel jobs...')
-            coverallz.parallel_finish()
-            log.info('Done')
-            return
-
-        log.info('Submitting coverage to coveralls.io...')
-        result = coverallz.wear()
-
-        log.info('Coverage submitted!')
-        log.debug(result)
-        if result:
-            log.info(result.get('message'))
-            log.info(result.get('url'))
-    except KeyboardInterrupt:  # pragma: no cover
+        for action, handler in action_handlers.items():
+            if options.get(action):
+                handler()
+                return
+
+        handle_default(coverallz)
+
+    except KeyboardInterrupt:  
         log.info('Aborted')
     except Exception as e:
         log.exception('Error running coveralls: %s', e)

coveralls/git.py

@@ -32,32 +32,33 @@ def gitlog(fmt: str) -> str:
     )
 
 
+def get_github_branch() -> Optional[str]:
+    github_ref = os.environ.get('GITHUB_REF')
+    if github_ref and (github_ref.startswith('refs/heads/') or github_ref.startswith('refs/tags/')):
+        return github_ref.split('/', 2)[-1]
+    return os.environ.get('GITHUB_HEAD_REF')
+
+def get_ci_branch() -> Optional[str]:
+    return (
+        os.environ.get('APPVEYOR_REPO_BRANCH')
+        or os.environ.get('BUILDKITE_BRANCH')
+        or os.environ.get('CI_BRANCH')
+        or os.environ.get('CIRCLE_BRANCH')
+        or os.environ.get('GIT_BRANCH')
+        or os.environ.get('TRAVIS_BRANCH')
+        or os.environ.get('BRANCH_NAME')
+    )
+
+def get_git_branch() -> Optional[str]:
+    try:
+        return run_command('git', 'rev-parse', '--abbrev-ref', 'HEAD')
+    except Exception:
+        return None
+
 def git_branch() -> Optional[str]:
-    branch = None
     if os.environ.get('GITHUB_ACTIONS'):
-        github_ref = os.environ.get('GITHUB_REF')
-        if (
-            github_ref.startswith('refs/heads/')
-            or github_ref.startswith('refs/tags/')
-        ):
-            # E.g. in push events.
-            branch = github_ref.split('/', 2)[-1]
-        else:
-            # E.g. in pull_request events.
-            branch = os.environ.get('GITHUB_HEAD_REF')
-    else:
-        branch = (
-            os.environ.get('APPVEYOR_REPO_BRANCH')
-            or os.environ.get('BUILDKITE_BRANCH')
-            or os.environ.get('CI_BRANCH')
-            or os.environ.get('CIRCLE_BRANCH')
-            or os.environ.get('GIT_BRANCH')
-            or os.environ.get('TRAVIS_BRANCH')
-            or os.environ.get('BRANCH_NAME')
-            or run_command('git', 'rev-parse', '--abbrev-ref', 'HEAD')
-        )
-
-    return branch
+        return get_github_branch()
+    return get_ci_branch() or get_git_branch()
 
 
 def git_info() -> Dict[str, Dict[str, Any]]: