lsp.py

#!/usr/bin/env python3
# pragma pylint: disable=too-many-lines
# test line 1
from __future__ import annotations # See: https://github.com/PyCQA/pylint/issues/3320
import argparse
import fnmatch
import functools
import json
import os
import re
import subprocess
import sys
import traceback
from collections import namedtuple
from copy import deepcopy
from enum import Enum, auto
from itertools import islice
from pathlib import PurePath
from typing import Any, List, Optional, Tuple, Union, NewType

import colorama  # Enables the use of SGR & CUP terminal VT sequences on Windows.
from deepdiff import DeepDiff

if os.name == 'nt':
    # pragma pylint: disable=import-error
    import msvcrt
else:
    import tty
    # Turn off user input buffering so we get the input immediately,
    # not only after a line break
    if os.isatty(sys.stdin.fileno()):
        tty.setcbreak(sys.stdin.fileno())


# Type for the pure test name without .sol suffix or sub directory
TestName = NewType("TestName", str)

# Type for the test path, e.g.  subdir/mytest.sol
RelativeTestPath = NewType("RelativeTestPath", str)


def escape_string(text: str) -> str:
    """
    Trivially escapes given input string's \r \n and \\.
    """
    return text.translate(str.maketrans({
        "\r": r"\r",
        "\n": r"\n",
        "\\": r"\\"
    }))


def getCharFromStdin() -> str:
    """
    Gets a single character from stdin without line-buffering.
    """
    if os.name == 'nt':
        # pragma pylint: disable=import-error
        return msvcrt.getch().decode("utf-8")
    else:
        return sys.stdin.read(1)


"""
Named tuple that holds various regexes used to parse the test specification.
"""
TestRegexesTuple = namedtuple("TestRegexesTuple", [
    "sendRequest", # regex to find requests to be sent & tested
    "findQuotedTag", # regex to find tags wrapped in quotes
    "findTag", # regex to find tags
    "fileDiagnostics", # regex to find diagnostic expectations for a file
    "diagnostic" # regex to find a single diagnostic within the file expectations
])
"""
Instance of the named tuple holding the regexes
"""
TEST_REGEXES = TestRegexesTuple(
    re.compile(R'^// -> (?P<method>[\w\/]+) {'),
    re.compile(R'(?P<tag>"@\w+")'),
    re.compile(R'(?P<tag>@\w+)'),
    re.compile(R'// (?P<testname>\S+):([ ](?P<diagnostics>[\w @]*))?'),
    re.compile(R'(?P<tag>@\w+) (?P<code>\d\d\d\d)')
)

"""
Named tuple holding regexes to find tags in the solidity code
"""
TagRegexesTuple = namedtuple("TagRegexestuple", ["simpleRange", "multilineRange"])
TAG_REGEXES = TagRegexesTuple(
    re.compile(R"(?P<range>[\^]+) (?P<tag>@\w+)"),
    re.compile(R"\^(?P<delimiter>[()]{1,2}) (?P<tag>@\w+)$")
)

def split_path(path):
    """
    Return the test name and the subdir path of the given path.
    """
    sub_dir_separator = path.rfind("/")

    if sub_dir_separator == -1:
        return (path, None)

    return (path[sub_dir_separator+1:], path[:sub_dir_separator])

def count_index(lines, start=0):
    """
    Takes an iterable of lines and adds the current byte index so it's available
    when iterating or looping.
    """
    n = start
    for elem in lines:
        yield n, elem
        n += 1 + len(elem)

def tags_only(lines, start=0):
    """
    Filter the lines for tag comments and report the line number that the tags
    _refer_ to (which is not the line they are on!).
    """
    n = start
    numCommentLines = 0

    def hasTag(line):
        if line.find("// ") != -1:
            for _, regex in TAG_REGEXES._asdict().items():
                if regex.search(line[len("// "):]) is not None:
                    return True
        return False

    for line in lines:
        if hasTag(line):
            numCommentLines += 1
            yield n - numCommentLines, line
        else:
            numCommentLines = 0

        n += 1


def prepend_comments(sequence):
    """
    Prepends a comment indicator to each element
    """
    result = ""
    for line in sequence.splitlines(True):
        result = result + "// " + line
    return result


# {{{ JsonRpcProcess
class BadHeader(Exception):
    def __init__(self, msg: str):
        super().__init__("Bad header: " + msg)

class JsonRpcProcess:
    exe_path: str
    exe_args: List[str]
    process: subprocess.Popen
    trace_io: bool
    print_pid: bool

    def __init__(self, exe_path: str, exe_args: List[str], trace_io: bool = True, print_pid = False):
        self.exe_path = exe_path
        self.exe_args = exe_args
        self.trace_io = trace_io
        self.print_pid = print_pid

    def __enter__(self):
        self.process = subprocess.Popen(
            [self.exe_path, *self.exe_args],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        if self.print_pid:
            print(f"solc pid: {self.process.pid}. Attach with sudo gdb -p {self.process.pid}")

        return self

    def __exit__(self, exception_type, exception_value, traceback) -> None:
        self.process.kill()
        self.process.wait(timeout=2.0)

    def trace(self, topic: str, message: str) -> None:
        if self.trace_io:
            print(f"{SGR_TRACE}{topic}:{SGR_RESET} {message}")

    def receive_message(self) -> Union[None, dict]:
        # Note, we should make use of timeout to avoid infinite blocking if nothing is received.
        CONTENT_LENGTH_HEADER = "Content-Length: "
        CONTENT_TYPE_HEADER = "Content-Type: "
        if self.process.stdout is None:
            return None
        message_size = None
        while True:
            # read header
            line = self.process.stdout.readline()
            if len(line) == 0:
                # server quit
                return None
            line = line.decode("utf-8")
            if self.trace_io:
                print(f"Received header-line: {escape_string(line)}")
            if not line.endswith("\r\n"):
                raise BadHeader("missing newline")
            # Safely remove the "\r\n".
            line = line.rstrip("\r\n")
            if line == '':
                break # done with the headers
            if line.startswith(CONTENT_LENGTH_HEADER):
                line = line[len(CONTENT_LENGTH_HEADER):]
                if not line.isdigit():
                    raise BadHeader("size is not int")
                message_size = int(line)
            elif line.startswith(CONTENT_TYPE_HEADER):
                # nothing todo with type for now.
                pass
            else:
                raise BadHeader("unknown header")
        if message_size is None:
            raise BadHeader("missing size")
        rpc_message = self.process.stdout.read(message_size).decode("utf-8")
        json_object = json.loads(rpc_message)
        self.trace('receive_message', json.dumps(json_object, indent=4, sort_keys=True))
        return json_object

    def send_message(self, method_name: str, params: Optional[dict]) -> None:
        if self.process.stdin is None:
            return
        message = {
            'jsonrpc': '2.0',
            'method': method_name,
            'params': params
        }
        json_string = json.dumps(obj=message)
        rpc_message = f"Content-Length: {len(json_string)}\r\n\r\n{json_string}"
        self.trace(f'send_message ({method_name})', json.dumps(message, indent=4, sort_keys=True))
        self.process.stdin.write(rpc_message.encode("utf-8"))
        self.process.stdin.flush()

    def call_method(self, method_name: str, params: Optional[dict], expects_response: bool = True) -> Any:
        self.send_message(method_name, params)
        if not expects_response:
            return None
        return self.receive_message()

    def send_notification(self, name: str, params: Optional[dict] = None) -> None:
        self.send_message(name, params)

# }}}

SGR_RESET = '\033[m'
SGR_TRACE = '\033[1;36m'
SGR_NOTICE = '\033[1;35m'
SGR_TEST_BEGIN = '\033[1;33m'
SGR_ASSERT_BEGIN = '\033[1;34m'
SGR_STATUS_OKAY = '\033[1;32m'
SGR_STATUS_FAIL = '\033[1;31m'

class ExpectationFailed(Exception):
    class Part(Enum):
        Diagnostics = auto()
        Methods = auto()

    def __init__(self, reason: str, part):
        self.part = part
        super().__init__(reason)

class JSONExpectationFailed(ExpectationFailed):
    def __init__(self, actual, expected, part):
        self.actual = actual
        self.expected = expected

        expected_pretty = ""

        if expected is not None:
            expected_pretty = json.dumps(expected, sort_keys=True)

        diff = DeepDiff(actual, expected)

        super().__init__(
            f"\n\tExpected {expected_pretty}" + \
            f"\n\tbut got {json.dumps(actual, sort_keys=True)}.\n\t{diff}",
            part
        )


def create_cli_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Solidity LSP Test suite")
    parser.set_defaults(fail_fast=False)
    parser.add_argument(
        "-f", "--fail-fast",
        dest="fail_fast",
        action="store_true",
        help="Terminates the running tests on first failure."
    )
    parser.set_defaults(non_interactive=False)
    parser.add_argument(
        "-n", "--non-interactive",
        dest="non_interactive",
        action="store_true",
        help="Prevent interactive queries and just fail instead."
    )
    parser.set_defaults(print_solc_pid=False)
    parser.add_argument(
        "-p", "--print-solc-pid",
        dest="print_solc_pid",
        action="store_true",
        help="Print pid of each started solc for debugging purposes."
    )
    parser.set_defaults(trace_io=False)
    parser.add_argument(
        "-T", "--trace-io",
        dest="trace_io",
        action="store_true",
        help="Be more verbose by also printing assertions."
    )
    parser.set_defaults(print_assertions=False)
    parser.add_argument(
        "-v", "--print-assertions",
        dest="print_assertions",
        action="store_true",
        help="Be more verbose by also printing assertions."
    )
    parser.add_argument(
        "-t", "--test-pattern",
        dest="test_pattern",
        type=str,
        default="*",
        help="Filters all available tests by matching against this test pattern (using globbing)",
        nargs="?"
    )
    parser.add_argument(
        "solc_path",
        type=str,
        default="solc",
        help="Path to solc binary to test against",
        nargs="?"
    )
    parser.add_argument(
        "project_root_dir",
        type=str,
        default=f"{os.path.dirname(os.path.realpath(__file__))}/..",
        help="Path to Solidity project's root directory (must be fully qualified).",
        nargs="?"
    )
    return parser

class Counter:
    total: int = 0
    passed: int = 0
    failed: int = 0


# Returns the given marker with the end extended by 'amount'
def extendEnd(marker, amount=1):
    newMarker = deepcopy(marker)
    newMarker["end"]["character"] += amount
    return newMarker

class TestParserException(Exception):
    def __init__(self, incompleteResult, msg: str):
        self.result = incompleteResult
        super().__init__("Failed to parse test specification: " + msg)

class TestParser:
    """
    Parses test specifications.
    Usage example:

    parsed_testcases = TestParser(content).parse()

    # First diagnostics are yielded.
    # Type is "TestParser.Diagnostics"
    expected_diagnostics = next(parsed_testcases)

    ...
    # Now each request/response pair in the test definition
    # Type is "TestParser.RequestAndResponse"
    for testcase in self.parsed_testcases:
        ...
    """
    RequestAndResponse = namedtuple('RequestAndResponse',
        "method, request, response, responseBegin, responseEnd",
        defaults=(None, None, None, None)
    )
    Diagnostics = namedtuple('Diagnostics', 'tests start end has_header')
    Diagnostic = namedtuple('Diagnostic', 'marker code')

    TEST_START = "// ----"

    def __init__(self, content: str):
        self.content = content
        self.lines = None
        self.current_line_tuple = None

    def parse(self):
        """
        Starts parsing the test specifications.
        Will first yield with the diagnostics expectations as type 'Diagnostics'.
        After that, it will yield once for every Request/Response pair found in
        the file, each time as type 'RequestAndResponse'.

        """
        testDefStartIdx = self.content.rfind(f"\n{self.TEST_START}\n")

        if testDefStartIdx == -1:
            # Set start/end to end of file if there is no test section
            yield self.Diagnostics({}, len(self.content), len(self.content), False)
            return

        self.lines = islice(
            count_index(self.content[testDefStartIdx+1:].splitlines(), testDefStartIdx+1),
            1,
            None
        )
        self.next_line()

        yield self.parseDiagnostics()

        while not self.at_end():
            yield self.parseRequestAndResponse()
            self.next_line()


    def parseDiagnostics(self) -> TestParser.Diagnostics:
        """
        Parse diagnostic expectations specified in the file.
        Returns a named tuple instance of "Diagnostics"
        """
        diagnostics = { "tests": {}, "has_header": True }

        diagnostics["start"] = self.position()

        while not self.at_end():
            fileDiagMatch = TEST_REGEXES.fileDiagnostics.match(self.current_line())
            if fileDiagMatch is None:
                break

            testDiagnostics = []

            diagnostics_string = fileDiagMatch.group("diagnostics")
            if diagnostics_string is not None:
                for diagnosticMatch in TEST_REGEXES.diagnostic.finditer(diagnostics_string):
                    testDiagnostics.append(self.Diagnostic(
                        diagnosticMatch.group("tag"),
                        int(diagnosticMatch.group("code"))
                    ))

            diagnostics["tests"][fileDiagMatch.group("testname")] = testDiagnostics

            self.next_line()

        diagnostics["end"] = self.position()
        return self.Diagnostics(**diagnostics)


    def parseRequestAndResponse(self) -> TestParser.RequestAndResponse:
        RESPONSE_START = "// <- "
        REQUEST_END = "// }"
        COMMENT_PREFIX = "// "

        ret = {}
        start_character = None

        # Parse request header
        requestResult = TEST_REGEXES.sendRequest.match(self.current_line())
        if requestResult is None:
            raise TestParserException(ret, "Method for request not found on line " + self.current_line())

        ret["method"] = requestResult.group("method")
        ret["request"] = "{\n"

        self.next_line()

        # Search for request block end
        while not self.at_end():
            line = self.current_line()
            ret["request"] += line[len(COMMENT_PREFIX):] + "\n"

            self.next_line()

            if line.startswith(REQUEST_END):
                break

            # Reached end without finding request_end. Abort.
            if self.at_end():
                raise TestParserException(ret, "Request body not found")

        if self.at_end():
            return self.RequestAndResponse(**ret)

        # Parse response header
        if self.current_line().startswith(RESPONSE_START):
            start_character = self.current_line()[len(RESPONSE_START)]
            if start_character not in ("{", "["):
                raise TestParserException(ret, "Response header malformed")
            ret["response"] = self.current_line()[len(RESPONSE_START):] + "\n"
            ret["responseBegin"] = self.position()

        self.next_line()

        end_character = "}" if start_character == "{" else "]"

        # Search for request block end
        while not self.at_end():
            ret["response"] += self.current_line()[len(COMMENT_PREFIX):] + "\n"

            if self.current_line().startswith(f"// {end_character}"):
                ret["responseEnd"] = self.position() + len(self.current_line())
                break

            self.next_line()

            # Reached end without finding block_end. Abort.
            if self.at_end():
                raise TestParserException(ret, "Response footer not found")

        return self.RequestAndResponse(**ret)

    def next_line(self):
        self.current_line_tuple = next(self.lines, None)

    def current_line(self):
        return self.current_line_tuple[1]

    def position(self):
        """
        Returns current byte position
        """
        if self.current_line_tuple is None:
            return len(self.content)
        return self.current_line_tuple[0]

    def at_end(self):
        """
        Returns True if we exhausted the lines
        """
        return self.current_line_tuple is None

class FileLoadStrategy(Enum):
    Undefined = None
    ProjectDirectory = 'project-directory'
    DirectlyOpenedAndOnImport =  'directly-opened-and-on-import'

class FileTestRunner:
    """
    Runs all tests in a given file.
    It is required to call test_diagnostics() before calling test_methods().

    When a test fails, asks the user how to proceed.
    Offers automatic test expectation updates and rerunning of the tests.
    """

    class TestResult(Enum):
        SuccessOrIgnored = auto()
        Reparse = auto()

    def __init__(self, test_name, sub_dir, solc, suite):
        self.test_name = test_name
        self.sub_dir = sub_dir
        self.suite = suite
        self.solc = solc
        self.open_tests = []
        self.content = self.suite.get_test_file_contents(self.test_name, self.sub_dir)
        self.markers = self.suite.get_test_tags(self.test_name, self.sub_dir)
        self.parsed_testcases = None
        self.expected_diagnostics = None

    def test_diagnostics(self):
        """
        Test that the expected diagnostics match the actual diagnostics
        """
        try:
            self.parsed_testcases = TestParser(self.content).parse()

            # Process diagnostics first
            self.expected_diagnostics = next(self.parsed_testcases)
            assert isinstance(self.expected_diagnostics, TestParser.Diagnostics)
            if not self.expected_diagnostics.has_header:
                return

            expected_diagnostics_per_file = self.expected_diagnostics.tests

            # Add our own test diagnostics if they didn't exist
            if self.test_name not in expected_diagnostics_per_file:
                expected_diagnostics_per_file[self.test_name] = []

            published_diagnostics = \
                self.suite.open_file_and_wait_for_diagnostics(self.solc, self.test_name, self.sub_dir)

            for diagnostics in published_diagnostics:
                if not diagnostics["uri"].startswith(self.suite.project_root_uri + "/"):
                    raise RuntimeError(
                        f"'{self.test_name}.sol' imported file outside of test directory: '{diagnostics['uri']}'"
                    )
                self.open_tests.append(self.suite.normalizeUri(diagnostics["uri"]))

            self.suite.expect_equal(
                len(published_diagnostics),
                len(expected_diagnostics_per_file),
                description="Amount of reports does not match!")

            for diagnostics_per_file in published_diagnostics:
                testname, sub_dir = split_path(self.suite.normalizeUri(diagnostics_per_file['uri']))

                # Clear all processed expectations so we can check at the end
                # what's missing
                expected_diagnostics = expected_diagnostics_per_file.pop(testname, {})

                self.suite.expect_equal(
                    len(diagnostics_per_file["diagnostics"]),
                    len(expected_diagnostics),
                    description="Unexpected amount of diagnostics"
                )
                markers = self.suite.get_test_tags(testname, sub_dir)
                for actual_diagnostic in diagnostics_per_file["diagnostics"]:
                    expected_diagnostic = next((diagnostic for diagnostic in
                        expected_diagnostics if actual_diagnostic['range'] ==
                        markers[diagnostic.marker]), None)

                    if expected_diagnostic is None:
                        raise ExpectationFailed(
                            f"Unexpected diagnostic: {json.dumps(actual_diagnostic, indent=4, sort_keys=True)}",
                            ExpectationFailed.Part.Diagnostics
                        )

                    self.suite.expect_diagnostic(
                        actual_diagnostic,
                        code=expected_diagnostic.code,
                        marker=markers[expected_diagnostic.marker]
                    )

            if len(expected_diagnostics_per_file) > 0:
                raise ExpectationFailed(
                    f"Expected diagnostics but received none for {expected_diagnostics_per_file}",
                    ExpectationFailed.Part.Diagnostics
                )

        except Exception:
            self.close_all_open_files()
            raise

    def close_all_open_files(self):
        for testpath in self.open_tests:
            test, sub_dir = split_path(testpath)

            self.solc.send_message(
                'textDocument/didClose',
                { 'textDocument': { 'uri': self.suite.get_test_file_uri(test, sub_dir) }}
            )
            self.suite.wait_for_diagnostics(self.solc)

        self.open_tests.clear()

    def test_methods(self) -> bool:
        """
        Test all methods. Returns False if a reparsing is required, else True
        """
        try:
            # Now handle each request/response pair in the test definition
            for testcase in self.parsed_testcases:
                try:
                    self.run_testcase(testcase)
                except JSONExpectationFailed as e:
                    result = self.user_interaction_failed_method_test(testcase, e.actual, e.expected)

                    if result == self.TestResult.Reparse:
                        return False

            return True
        except TestParserException as e:
            print(e)
            print(e.result)
            raise
        finally:
            self.close_all_open_files()

    def user_interaction_failed_method_test(
        self,
        testcase: TestParser.RequestAndResponse,
        actual,
        expected
    ) -> TestResult:

        actual_pretty = self.suite.replace_ranges_with_tags(actual, self.sub_dir)

        if expected is None:
            print("Failed to parse expected response, received:\n" + actual)
        else:
            print("Expected:\n" + \
                self.suite.replace_ranges_with_tags(expected, self.sub_dir) + \
                "\nbut got:\n" + actual_pretty
            )

        if self.suite.non_interactive:
            return self.TestResult.SuccessOrIgnored

        while True:
            print("(u)pdate/(r)etry/(i)gnore?")
            user_response = getCharFromStdin()
            if user_response == "i":
                return self.TestResult.SuccessOrIgnored

            if user_response == "u":
                actual = actual["result"]
                self.content = self.content[:testcase.responseBegin] + \
                    prepend_comments(
                        "<- " + \
                        self.suite.replace_ranges_with_tags(actual, self.sub_dir)) + \
                    self.content[testcase.responseEnd:]

                with open(self.suite.get_test_file_path(\
                    self.test_name, self.sub_dir), \
                    mode="w", \
                    encoding="utf-8", \
                    newline='') as f:
                    f.write(self.content)
                return self.TestResult.Reparse
            if user_response == "r":
                return self.TestResult.Reparse

            print("Invalid response.")


    def run_testcase(self, testcase: TestParser.RequestAndResponse):
        """
        Runs the given testcase.
        """

        requestBodyJson = self.parse_json_with_tags(testcase.request, self.markers)
        # add textDocument/uri if missing
        if 'textDocument' not in requestBodyJson:
            requestBodyJson['textDocument'] = { 'uri': self.suite.get_test_file_uri(self.test_name, self.sub_dir) }

        actualResponseJson = self.solc.call_method(
            testcase.method,
            requestBodyJson,
            expects_response=testcase.response is not None
        )

        if testcase.response is None:
            return

        # simplify response
        if "result" in actualResponseJson:
            if isinstance(actualResponseJson["result"], list):
                for result in actualResponseJson["result"]:
                    if "uri" in result:
                        result["uri"] = result["uri"].replace(self.suite.project_root_uri + "/" + self.sub_dir + "/", "")

            elif isinstance(actualResponseJson["result"], dict):
                if "changes" in actualResponseJson["result"]:
                    changes = actualResponseJson["result"]["changes"]
                    for key in list(changes.keys()):
                        new_key = key.replace(self.suite.project_root_uri + "/", "")
                        changes[new_key] = changes[key]
                        del changes[key]

        if "jsonrpc" in actualResponseJson:
            actualResponseJson.pop("jsonrpc")

        try:
            expectedResponseJson = self.parse_json_with_tags(testcase.response, self.markers)
        except json.decoder.JSONDecodeError:
            expectedResponseJson = None

        expectedResponseJson = { "result": expectedResponseJson }

        self.suite.expect_equal(
            actualResponseJson,
            expectedResponseJson,
            f"Request failed: \n{testcase.request}",
            ExpectationFailed.Part.Methods
        )


    def parse_json_with_tags(self, content, markersFallback):
        """
        Replaces any tags with their actual content and parsers the result as
        json to return it.
        """
        split_by_tag = TEST_REGEXES.findTag.split(content)

        # add quotes so we can parse it as json
        contentReplaced = '"'.join(split_by_tag)
        contentJson = json.loads(contentReplaced)

        def replace_tag(data, markers):

            if isinstance(data, list):
                for el in data:
                    replace_tag(el, markers)
                return data

            if not isinstance(data, dict):
                return data

            def findMarker(desired_tag):
                if not isinstance(desired_tag, str):
                    return desired_tag

                for tag, tagRange in markers.items():
                    if tag == desired_tag:
                        return tagRange
                    elif tag.lower() == desired_tag.lower():
                        raise RuntimeError(f"Detected lower/upper case mismatch: Requested {desired_tag} but only found {tag}")

                raise RuntimeError(f"Marker {desired_tag} not found in file")


            # Check if we need markers from a specific file
            # Needs to be done before the loop or it might be called only after
            # we found "range" or "position"
            if "uri" in data:
                markers = self.suite.get_test_tags(data["uri"][:-len(".sol")], self.sub_dir)

            for key, val in data.items():
                if key == "range":
                    data[key] = findMarker(val)
                elif key == "position":
                    tag_range = findMarker(val)
                    if "start" in tag_range:
                        data[key] = tag_range["start"]
                elif key == "changes":
                    for path, list_of_changes in val.items():
                        test_name, file_sub_dir = split_path(path)
                        markers = self.suite.get_test_tags(test_name[:-len(".sol")], file_sub_dir)
                        for change in list_of_changes:
                            if "range" in change:
                                change["range"] = findMarker(change["range"])
                elif isinstance(val, dict):
                    replace_tag(val, markers)
                elif isinstance(val, list):
                    for el in val:
                        replace_tag(el, markers)
            return data

        return replace_tag(contentJson, markersFallback)


class SolidityLSPTestSuite: # {{{
    test_counter = Counter()
    assertion_counter = Counter()
    print_assertions: bool = False
    trace_io: bool = False
    fail_fast: bool = False
    test_pattern: str

    def __init__(self):
        colorama.init()
        args = create_cli_parser().parse_args()
        self.solc_path = args.solc_path
        self.project_root_dir = os.path.realpath(args.project_root_dir) + "/test/libsolidity/lsp"
        self.project_root_uri = PurePath(self.project_root_dir).as_uri()
        self.print_assertions = args.print_assertions
        self.trace_io = args.trace_io
        self.test_pattern = args.test_pattern
        self.fail_fast = args.fail_fast
        self.non_interactive = args.non_interactive
        self.print_solc_pid = args.print_solc_pid

        print(f"{SGR_NOTICE}test pattern: {self.test_pattern}{SGR_RESET}")

    def main(self) -> int:
        """
        Runs all test cases.
        Returns 0 on success and the number of failing assertions (capped to 127) otherwise.
        """
        all_tests = sorted([
            str(name)[5:]
            for name in dir(SolidityLSPTestSuite)
            if callable(getattr(SolidityLSPTestSuite, name)) and name.startswith("test_")
        ])
        filtered_tests = fnmatch.filter(all_tests, self.test_pattern)
        if filtered_tests.count("generic") == 0:
            filtered_tests.append("generic")

        for method_name in filtered_tests:
            test_fn = getattr(self, 'test_' + method_name)
            title: str = test_fn.__name__[5:]
            print(f"{SGR_TEST_BEGIN}Testing {title} ...{SGR_RESET}")
            try:
                with JsonRpcProcess(
                    self.solc_path,
                    ["--experimental", "--lsp"],
                    trace_io=self.trace_io,
                    print_pid=self.print_solc_pid
                ) as solc:
                    test_fn(solc)
                    self.test_counter.passed += 1
            except ExpectationFailed:
                self.test_counter.failed += 1
                print(traceback.format_exc())
                if self.fail_fast:
                    break
            except Exception as e: # pragma pylint: disable=broad-except
                self.test_counter.failed += 1
                print(f"Unhandled exception {e.__class__.__name__} caught: {e}")
                print(traceback.format_exc())
                if self.fail_fast:
                    break

        print(
            f"\n{SGR_NOTICE}Summary:{SGR_RESET}\n\n"
            f"  Test cases: {self.test_counter.passed} passed, {self.test_counter.failed} failed\n"
            f"  Assertions: {self.assertion_counter.passed} passed, {self.assertion_counter.failed} failed\n"
        )

        return min(max(self.test_counter.failed, self.assertion_counter.failed), 127)

    def setup_lsp(
        self,
        lsp: JsonRpcProcess,
        expose_project_root=True,
        file_load_strategy: FileLoadStrategy=FileLoadStrategy.DirectlyOpenedAndOnImport,
        custom_include_paths: list[str] = None,
        project_root_subdir=None
    ):
        """
        Prepares the solc LSP server by calling `initialize`,
        and `initialized` methods.
        """
        project_root_uri_with_maybe_subdir = self.project_root_uri
        if project_root_subdir is not None:
            project_root_uri_with_maybe_subdir = self.project_root_uri + '/' + project_root_subdir
        params = {
            'processId': None,
            'rootUri': project_root_uri_with_maybe_subdir,
            # Enable traces to receive the amount of expected diagnostics before
            # actually receiving them.
            'trace': 'messages',
            'capabilities': {
                'textDocument': {
                    'publishDiagnostics': {'relatedInformation': True}
                },
                'workspace': {
                    'applyEdit': True,
                    'configuration': True,
                    'didChangeConfiguration': {'dynamicRegistration': True},
                    'workspaceEdit': {'documentChanges': True},
                    'workspaceFolders': True
                }
            }
        }

        if file_load_strategy != FileLoadStrategy.Undefined:
            params['initializationOptions'] = {}
            params['initializationOptions']['file-load-strategy'] = file_load_strategy.value

        if custom_include_paths is not None and len(custom_include_paths) != 0:
            if params['initializationOptions'] is None:
                params['initializationOptions'] = {}
            params['initializationOptions']['include-paths'] = custom_include_paths

        if not expose_project_root:
            params['rootUri'] = None

        lsp.call_method('initialize', params)
        lsp.send_notification('initialized')

    # {{{ helpers
    def get_test_file_path(self, test_case_name, sub_dir=None):
        if sub_dir:
            return f"{self.project_root_dir}/{sub_dir}/{test_case_name}.sol"
        return f"{self.project_root_dir}/{test_case_name}.sol"

    def get_test_file_uri(self, test_case_name, sub_dir=None):
        return PurePath(self.get_test_file_path(test_case_name, sub_dir)).as_uri()

    def get_test_file_contents(self, test_case_name, sub_dir=None):
        """
        Reads the file contents from disc for a given test case.
        The `test_case_name` will be the basename of the file
        in the test path (test/libsolidity/lsp/{sub_dir}).
        """
        with open(self.get_test_file_path(test_case_name, sub_dir), mode="r", encoding="utf-8", newline='') as f:
            return f.read().replace("\r\n", "\n")

    def require_params_for_method(self, method_name: str, message: dict) -> Any:
        """
        Ensures the given RPC message does contain the
        field 'method' with the given method name,
        and then returns its passed params.
        An exception is raised on expectation failures.
        """
        assert message is not None
        if 'error' in message.keys():
            code = message['error']["code"]
            text = message['error']['message']
            raise RuntimeError(f"Error {code} received. {text}")
        if 'method' not in message.keys():
            raise RuntimeError("No method received but something else.")
        self.expect_equal(message['method'], method_name, description="Ensure expected method name")
        return message['params']

    def wait_for_diagnostics(self, solc: JsonRpcProcess) -> List[dict]:
        """
        Return all published diagnostic reports sorted by file URI.