<commit_before># Python Tools for Visual Studio
# Copyright(c) Microsoft Corporation
# All rights reserved.
# 
# Licensed under the Apache License, Version 2.0 (the License); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at http://www.apache.org/licenses/LICENSE-2.0
# 
# THIS CODE IS PROVIDED ON AN  *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS
# OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY
# IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE,
# MERCHANTABLITY OR NON-INFRINGEMENT.
# 
# See the Apache Version 2.0 License for specific language governing
# permissions and limitations under the License.

from __future__ import with_statement

__author__ = "<NAME> <<EMAIL>>"
__version__ = "3.0.0.0"

import sys
import json
import unittest
import socket
import traceback
from types import CodeType, FunctionType
try:
    import thread
except:
    import _thread as thread

if sys.version_info[0] < 3:
    if sys.version_info[:2] < (2, 6):
        from codecs import open
    else:
        from io import open

import visualstudio_py_ipcjson as _vsipc

class _TestOutput(object):
    """file like object which redirects output to the repl window."""
    errors = 'strict'

    def __init__(self, old_out, is_stdout):
        self.is_stdout = is_stdout
        self.old_out = old_out
        if sys.version >= '3.' and hasattr(old_out, 'buffer'):
            self.buffer = _TestOutputBuffer(old_out.buffer, is_stdout)

    def flush(self):
        if self.old_out:
            self.old_out.flush()
    
    def writelines(self, lines):
        for line in lines:
            self.write(line)
    
    @property
    def encoding(self):
        return 'utf8'

    def write(self, value):
        _channel.send_event('stdout' if self.is_stdout else 'stderr', content=value)
        if self.old_out:
            self.old_out.write(value)
    
    def isatty(self):
        return True

    def next(self):
        pass
    
    @property
    def name(self):
        if self.is_stdout:
            return "<stdout>"
        else:
            return "<stderr>"

    def __getattr__(self, name):
        return getattr(self.old_out, name)

class _TestOutputBuffer(object):
    def __init__(self, old_buffer, is_stdout):
        self.buffer = old_buffer
        self.is_stdout = is_stdout

    def write(self, data):
        _channel.send_event('stdout' if self.is_stdout else 'stderr', content=data)
        self.buffer.write(data)

    def flush(self): 
        self.buffer.flush()

    def truncate(self, pos = None):
        return self.buffer.truncate(pos)

    def tell(self):
        return self.buffer.tell()

    def seek(self, pos, whence = 0):
        return self.buffer.seek(pos, whence)

class _IpcChannel(_vsipc.SocketIO, _vsipc.IpcChannel):
    def __init__(self, socket):
        super(_IpcChannel, self).__init__(socket=socket)

_channel = None


class VsTestResult(unittest.TextTestResult):
    def startTest(self, test):
        super(VsTestResult, self).startTest(test)
        if _channel is not None:
            _channel.send_event(
                name='start', 
                test = test.test_id
            )

    def addError(self, test, err):
        super(VsTestResult, self).addError(test, err)<commit_msg>Undo sharing of IPC JSON code in test launcher, it caused a regression in TestExecutorTests.TestMultiprocessing.<commit_after># Python Tools for Visual Studio
# Copyright(c) Microsoft Corporation
# All rights reserved.
# 
# Licensed under the Apache License, Version 2.0 (the License); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at http://www.apache.org/licenses/LICENSE-2.0
# 
# THIS CODE IS PROVIDED ON AN  *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS
# OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY
# IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE,
# MERCHANTABLITY OR NON-INFRINGEMENT.
# 
# See the Apache Version 2.0 License for specific language governing
# permissions and limitations under the License.

from __future__ import with_statement

__author__ = "Microsoft Corporation <<EMAIL>>"
__version__ = "3.0.0.0"

import sys
import json
import unittest
import socket
import traceback
from types import CodeType, FunctionType
try:
    import thread
except:
    import _thread as thread

if sys.version_info[0] < 3:
    if sys.version_info[:2] < (2, 6):
        from codecs import open
    else:
        from io import open


class _TestOutput(object):
    """file like object which redirects output to the repl window."""
    errors = 'strict'

    def __init__(self, old_out, is_stdout):
        self.is_stdout = is_stdout
        self.old_out = old_out
        if sys.version >= '3.' and hasattr(old_out, 'buffer'):
            self.buffer = _TestOutputBuffer(old_out.buffer, is_stdout)

    def flush(self):
        if self.old_out:
            self.old_out.flush()
    
    def writelines(self, lines):
        for line in lines:
            self.write(line)
    
    @property
    def encoding(self):
        return 'utf8'

    def write(self, value):
        _channel.send_event('stdout' if self.is_stdout else 'stderr', content=value)
        if self.old_out:
            self.old_out.write(value)
    
    def isatty(self):
        return True

    def next(self):
        pass
    
    @property
    def name(self):
        if self.is_stdout:
            return "<stdout>"
        else:
            return "<stderr>"

    def __getattr__(self, name):
        return getattr(self.old_out, name)

class _TestOutputBuffer(object):
    def __init__(self, old_buffer, is_stdout):
        self.buffer = old_buffer
        self.is_stdout = is_stdout

    def write(self, data):
        _channel.send_event('stdout' if self.is_stdout else 'stderr', content=data)
        self.buffer.write(data)

    def flush(self): 
        self.buffer.flush()

    def truncate(self, pos = None):
        return self.buffer.truncate(pos)

    def tell(self):
        return self.buffer.tell()

    def seek(self, pos, whence = 0):
        return self.buffer.seek(pos, whence)

class _IpcChannel(object):
    def __init__(self, socket):
        self.socket = socket
        self.seq = 0
        self.lock = thread.allocate_lock()

    def close(self):
        self.socket.close()

    def send_event(self, name, **args):
        with self.lock:
            body = {'type': 'event', 'seq': self.seq, 'event':name, 'body':args}
            self.seq += 1
            content = json.dumps(body).encode('utf8')
            headers = ('Content-Length: %d\n\n' % (len(content), )).encode('utf8')
            self.socket.send(headers)
            self.socket.send(content)

_channel = None


class VsTestResult(unittest.TextTestResult):
    def startTest(self, test):
        super(VsTestResult, self).startTest(test)
        if _channel is not None:
            _channel.send_event(
                name='start', 
                test = test.test_id
            )

    def addError(self, test, err):
        super(VsTestResult, self).addError(test, err)
{'fraction_non_alphanumeric': 0.09683684794672585, 'fraction_numerical': 0.005549389567147614, 'mean_word_length': 2.6668362156663274, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6644576', 'n_tokens_mistral': 2194, 'n_tokens_neox': 2124, 'n_words': 654}