<commit_before># Copyright 2012 Pinterest.com
# -*- coding: utf-8 -*-
#
# 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
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from builtins import bytes as newbytes

import collections
import errno
import functools
import json
import os
import mock
import re
import socket
import unittest

import pytest

from pymemcache.client.base import PooledClient, Client, normalize_server_spec
from pymemcache.exceptions import (
    MemcacheClientError,
    MemcacheServerError,
    MemcacheUnknownCommandError,
    MemcacheUnknownError,
    MemcacheIllegalInputError
)

from pymemcache import pool
from pymemcache.test.utils import MockMemcacheClient


class MockSocket(object):
    def __init__(self, recv_bufs, connect_failure=None, close_failure=None):
        self.recv_bufs = collections.deque(recv_bufs)
        self.send_bufs = []
        self.closed = False
        self.timeouts = []
        self.connect_failure = connect_failure
        self.close_failure = close_failure
        self.connections = []
        self.socket_options = []

    @property
    def family(self):
        # TODO: Use ipaddress module when dropping support for Python < 3.3
        ipv6_re = re.compile(r'^[0-9a-f:]+$')
        is_ipv6 = any(ipv6_re.match(c[0]) for c in self.connections)
        return socket.AF_INET6 if is_ipv6 else socket.AF_INET

    def sendall(self, value):
        self.send_bufs.append(value)

    def close(self):
        if isinstance(self.close_failure, Exception):
            raise self.close_failure
        self.closed = True

    def recv(self, size):
        value = self.recv_bufs.popleft()
        if isinstance(value, Exception):
            raise value
        return value

    def settimeout(self, timeout):
        self.timeouts.append(timeout)

    def connect(self, server):
        if isinstance(self.connect_failure, Exception):
            raise self.connect_failure
        self.connections.append(server)

    def setsockopt(self, level, option, value):
        self.socket_options.append((level, option, value))


class MockUnixSocketServer(object):
    def __init__(self, socket_path):
        if os.path.exists(socket_path):
            os.remove(socket_path)
        self.socket_path = socket_path
        self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

    def __enter__(self):
        self.socket.bind(self.socket_path)
        self.socket.listen(1)
        return self.socket

    def __exit__(self, *args):
        self.socket.close()
        os.remove(self.socket_path)


class MockSocketModule(object):
    def __init__(self, connect_failure=None, close_failure=None):
        self.connect_failure = connect_failure
        self.close_failure = close_failure
        self.sockets = []

    def socket(self, family, type, proto=0, fileno=None):
        socket = MockSocket(
            [],
            connect_failure=self.connect_failure,
            close_failure=self.close_failure)
        self.sockets.append(socket)
        return socket

    def __getattr__(self, name):
        return getattr(socket, name)


class CustomizedClient(Client):

    def _extract_value(self, expect_cas, line, buf, remapped_keys,
                       prefixed_keys):
        return b'key', b'value', b'END\r\n'


@pytest.mark.unit()
class ClientTestMixin(object):
    def make_client(self, mock_socket_values, **kwargs):
        client = Client(None, **kwargs)
        # mock out client._connect() rather than hard-settting client.sock to
        # ensure methods are checking whether self.sock is None before
        # attempting to use it
        sock = MockSocket(list(mock_socket_values))
        client._connect = mock.Mock(side_effect=functools.partial(
            setattr, client, "sock", sock))<commit_msg>Provide a mock implementation of socket.getaddrinfo (#327)<commit_after># Copyright 2012 Pinterest.com
# -*- coding: utf-8 -*-
#
# 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
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from builtins import bytes as newbytes

import collections
import errno
import functools
import json
import os
import mock
import re
import socket
import unittest

import pytest

from pymemcache.client.base import PooledClient, Client, normalize_server_spec
from pymemcache.exceptions import (
    MemcacheClientError,
    MemcacheServerError,
    MemcacheUnknownCommandError,
    MemcacheUnknownError,
    MemcacheIllegalInputError
)

from pymemcache import pool
from pymemcache.test.utils import MockMemcacheClient


# TODO: Use ipaddress module when dropping support for Python < 3.3
def is_ipv6(address):
    return re.match(r'^[0-9a-f:]+$', address)


class MockSocket(object):
    def __init__(self, recv_bufs, connect_failure=None, close_failure=None):
        self.recv_bufs = collections.deque(recv_bufs)
        self.send_bufs = []
        self.closed = False
        self.timeouts = []
        self.connect_failure = connect_failure
        self.close_failure = close_failure
        self.connections = []
        self.socket_options = []

    @property
    def family(self):
        any_ipv6 = any(is_ipv6(c[0]) for c in self.connections)
        return socket.AF_INET6 if any_ipv6 else socket.AF_INET

    def sendall(self, value):
        self.send_bufs.append(value)

    def close(self):
        if isinstance(self.close_failure, Exception):
            raise self.close_failure
        self.closed = True

    def recv(self, size):
        value = self.recv_bufs.popleft()
        if isinstance(value, Exception):
            raise value
        return value

    def settimeout(self, timeout):
        self.timeouts.append(timeout)

    def connect(self, server):
        if isinstance(self.connect_failure, Exception):
            raise self.connect_failure
        self.connections.append(server)

    def setsockopt(self, level, option, value):
        self.socket_options.append((level, option, value))


class MockUnixSocketServer(object):
    def __init__(self, socket_path):
        if os.path.exists(socket_path):
            os.remove(socket_path)
        self.socket_path = socket_path
        self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

    def __enter__(self):
        self.socket.bind(self.socket_path)
        self.socket.listen(1)
        return self.socket

    def __exit__(self, *args):
        self.socket.close()
        os.remove(self.socket_path)


class MockSocketModule(object):
    def __init__(self, connect_failure=None, close_failure=None):
        self.connect_failure = connect_failure
        self.close_failure = close_failure
        self.sockets = []

    def socket(self, family, type, proto=0, fileno=None):
        socket = MockSocket(
            [],
            connect_failure=self.connect_failure,
            close_failure=self.close_failure)
        self.sockets.append(socket)
        return socket

    def getaddrinfo(self, host, port, family=0, type=0, proto=0, flags=0):
        family = family or (
            socket.AF_INET6 if is_ipv6(host) else socket.AF_INET
        )
        type = type or socket.SOCK_STREAM
        proto = proto or socket.IPPROTO_TCP
        sockaddr = (
            ('::1', 11211, 0, 0)
            if family == socket.AF_INET6
            else ('127.0.0.1', 11211)
        )
        return [(family, type, proto, '', sockaddr)]

    def __getattr__(self, name):
        return getattr(socket, name)


class CustomizedClient(Client):

    def _extract_value(self, expect_cas, line, buf, remapped_keys,
                       prefixed_keys):
        return b'key', b'value', b'END\r\n'


@pytest.mark.unit()
class ClientTestMixin(object):
    def make_client(self, mock_socket_values, **kwargs):
        client = Client(None, **kwargs)
        # mock out client._connect() rather than hard-settting client.sock to
        # ensure methods are checking whether self.sock is None before
        # attempting to use it
        sock = MockSocket(list(mock_socket_values))
        client._connect = mock.Mock(side_effect=functools.partial(
            setattr, client, "sock", sock))
{'fraction_non_alphanumeric': 0.08955720785706359, 'fraction_numerical': 0.007879258683830874, 'mean_word_length': 2.9491673970201577, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 1, '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': '4015516', 'n_tokens_mistral': 2650, 'n_tokens_neox': 2578, 'n_words': 708}