teamcity/flake8_v3_plugin.py

import re
from io import BytesIO

from flake8.formatting import base

from teamcity.messages import TeamcityServiceMessages
from teamcity import __version__, is_running_under_teamcity


class TeamcityReport(base.BaseFormatter):
    name = 'teamcity-messages'
    version = __version__

    @staticmethod
    def _add_option(parser, name, *args, **kwargs):
        if all(option.long_option_name != name for option in parser.options):
            parser.add_option(name, *args, **kwargs)

    @classmethod
    def add_options(cls, parser):
        cls._add_option(parser,
                        '--teamcity',
                        default=is_running_under_teamcity(),
                        help="Force output of JetBrains TeamCity service messages")
        cls._add_option(parser,
                        '--no-teamcity',
                        default=False,
                        help="Disable output of JetBrains TeamCity service messages (even under TeamCity build)")

    @classmethod
    def parse_options(cls, options):
        if not options.no_teamcity:
            if options.teamcity or is_running_under_teamcity():
                options.format = 'teamcity-messages'

    def format(self, error):
        normalized_filename = error.filename.replace("\\", "/")
        position = '%s:%d:%d' % (
            normalized_filename, error.line_number, error.column_number)
        error_message = '%s %s' % (error.code, error.text)
        test_name = 'pep8: %s: %s' % (normalized_filename, error_message)

        line = error.physical_line
        offset = error.column_number
        details = [
            position,
            line.rstrip(),
            re.sub(r'\S', ' ', line[:offset]) + '^',
        ]
        details = '\n'.join(details)

        bytesio = BytesIO()
        messages = TeamcityServiceMessages(output=bytesio)

        messages.testStarted(test_name)
        messages.testFailed(test_name, error_message, details)
        messages.testFinished(test_name)

        return bytesio.getvalue().decode('UTF-8')