<commit_before>#!/usr/bin/env python
# Copyright (c) 2015 Cloudera, Inc. 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
#
# 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.

import pytest
import os
import time
import threading
import socket
import uuid
import urllib2
import json

from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from thrift.server.TServer import TServer
from Types.ttypes import TNetworkAddress

import StatestoreService.StatestoreSubscriber as Subscriber
import StatestoreService.StatestoreService as Statestore
from StatestoreService.StatestoreSubscriber import TUpdateStateResponse
from StatestoreService.StatestoreSubscriber import TTopicRegistration
from ErrorCodes.ttypes import TErrorCode
from Status.ttypes import TStatus

# Tests for the statestore. The StatestoreSubscriber class is a skeleton implementation of
# a Python-based statestore subscriber with additional hooks to allow testing. Each
# StatestoreSubscriber runs its own server so that the statestore may contact it.
#
# All tests in this file may be run in parallel. They assume that a statestore instance is
# already running, and is configured with out-of-the-box defaults (as is the case in our
# usual test environment) which govern failure-detector timeouts etc.
#
# These tests do not yet provide sufficient coverage.
#    If no topic entries, do the first and second subscribers always get a callback?
#    Adding topic entries to non-existant topic
#    Test for from_version and to_version behavior
#    Test with many concurrent subscribers
#    Test that only the subscribed-to topics are sent
#    Test that topic deletions take effect correctly.

def get_statestore_subscribers(host='localhost', port=25010):
  response = urllib2.urlopen("http://{0}:{1}/subscribers?json".format(host, port))
  page = response.read()
  return json.loads(page)

STATUS_OK = TStatus(TErrorCode.OK)
DEFAULT_UPDATE_STATE_RESPONSE = TUpdateStateResponse(status=STATUS_OK, topic_updates=[],
                                                     skipped=False)

class WildcardServerSocket(TSocket.TSocketBase, TTransport.TServerTransportBase):
  """Specialised server socket that binds to a random port at construction"""
  def __init__(self, host=None, port=0):
    self.host = host
    self.handle = None
    self.handle = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.handle.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    self.handle.bind(('localhost', 0))
    _, self.port = self.handle.getsockname()

  def listen(self):
    self.handle.listen(128)

  def accept(self):
    client, addr = self.handle.accept()
    result = TSocket.TSocket()
    result.setHandle(client)
    return result


class KillableThreadedServer(TServer):
  """Based on TServer.TThreadedServer, this server may be shutdown (by calling
  shutdown()), after which no new connections may be made. Most of the implementation is
  directly copied from Thrift."""
  def __init__(self, *args, **kwargs):
    TServer.__init__(self, *args)
    self.daemon = kwargs.get("daemon", False)
    self.is_shutdown = False
    self.transports = set()

  def shutdown(self):
    self.is_shutdown = True
    self.serverTransport.close()
    for t in self.transports:
      t.close()
    self.wait_until_down()

  def wait_until_up(self, num_tries=10):
    for i in xrange(num_tries):
      cnxn = TSocket.TSocket('localhost', self.port)
      try:
        cnxn.open()
        return
      except Exception, e:
        if i == num_tries - 1: raise
      time.sleep(0.1)

  def wait_until_down(self, num_tries=10):
    for i in xrange(num_tries):
      cnxn = TSocket.TSocket('localhost', self.port)
      try:
        cnxn.open()
        time.sleep(0.1)
      except Exception, e:
        return
    raise Exception("Server did not stop")

  def serve(self):
    self.port = self.serverTransport.port
    self.serverTransport.listen()
    while not self.is_shutdown:
      client = self.serverTransport.accept()
      # Since accept() can take a while, check again if the server is shutdown to avoid
      # starting an unnecessary thread.
      if self.is_shutdown: return
      t = threading.Thread(target=self.handle, args=(client,))
      t.setDaemon(self.daemon)
      t.start()
<commit_msg>Fix race condition in test_statestore changes<commit_after>#!/usr/bin/env python
# Copyright (c) 2015 Cloudera, Inc. 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
#
# 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.

import pytest
import os
import time
import threading
import socket
import uuid
import urllib2
import json

from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from thrift.server.TServer import TServer
from Types.ttypes import TNetworkAddress

import StatestoreService.StatestoreSubscriber as Subscriber
import StatestoreService.StatestoreService as Statestore
from StatestoreService.StatestoreSubscriber import TUpdateStateResponse
from StatestoreService.StatestoreSubscriber import TTopicRegistration
from ErrorCodes.ttypes import TErrorCode
from Status.ttypes import TStatus

# Tests for the statestore. The StatestoreSubscriber class is a skeleton implementation of
# a Python-based statestore subscriber with additional hooks to allow testing. Each
# StatestoreSubscriber runs its own server so that the statestore may contact it.
#
# All tests in this file may be run in parallel. They assume that a statestore instance is
# already running, and is configured with out-of-the-box defaults (as is the case in our
# usual test environment) which govern failure-detector timeouts etc.
#
# These tests do not yet provide sufficient coverage.
#    If no topic entries, do the first and second subscribers always get a callback?
#    Adding topic entries to non-existant topic
#    Test for from_version and to_version behavior
#    Test with many concurrent subscribers
#    Test that only the subscribed-to topics are sent
#    Test that topic deletions take effect correctly.

def get_statestore_subscribers(host='localhost', port=25010):
  response = urllib2.urlopen("http://{0}:{1}/subscribers?json".format(host, port))
  page = response.read()
  return json.loads(page)

STATUS_OK = TStatus(TErrorCode.OK)
DEFAULT_UPDATE_STATE_RESPONSE = TUpdateStateResponse(status=STATUS_OK, topic_updates=[],
                                                     skipped=False)

class WildcardServerSocket(TSocket.TSocketBase, TTransport.TServerTransportBase):
  """Specialised server socket that binds to a random port at construction"""
  def __init__(self, host=None, port=0):
    self.host = host
    self.handle = None
    self.handle = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.handle.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    self.handle.bind(('localhost', 0))
    _, self.port = self.handle.getsockname()

  def listen(self):
    self.handle.listen(128)

  def accept(self):
    client, addr = self.handle.accept()
    result = TSocket.TSocket()
    result.setHandle(client)
    return result


class KillableThreadedServer(TServer):
  """Based on TServer.TThreadedServer, this server may be shutdown (by calling
  shutdown()), after which no new connections may be made. Most of the implementation is
  directly copied from Thrift."""
  def __init__(self, *args, **kwargs):
    TServer.__init__(self, *args)
    self.daemon = kwargs.get("daemon", False)
    self.is_shutdown = False
    self.transports = set()
    self.port = self.serverTransport.port

  def shutdown(self):
    self.is_shutdown = True
    self.serverTransport.close()
    for t in self.transports:
      t.close()
    self.wait_until_down()

  def wait_until_up(self, num_tries=10):
    for i in xrange(num_tries):
      cnxn = TSocket.TSocket('localhost', self.port)
      try:
        cnxn.open()
        return
      except Exception, e:
        if i == num_tries - 1: raise
      time.sleep(0.1)

  def wait_until_down(self, num_tries=10):
    for i in xrange(num_tries):
      cnxn = TSocket.TSocket('localhost', self.port)
      try:
        cnxn.open()
        time.sleep(0.1)
      except Exception, e:
        return
    raise Exception("Server did not stop")

  def serve(self):
    self.serverTransport.listen()
    while not self.is_shutdown:
      client = self.serverTransport.accept()
      # Since accept() can take a while, check again if the server is shutdown to avoid
      # starting an unnecessary thread.
      if self.is_shutdown: return
      t = threading.Thread(target=self.handle, args=(client,))
      t.setDaemon(self.daemon)
      t.start()

{'fraction_non_alphanumeric': 0.08092365300603287, 'fraction_numerical': 0.006656958602038693, 'mean_word_length': 3.8316582914572863, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7469113', 'n_tokens_mistral': 2798, 'n_tokens_neox': 2735, 'n_words': 949}