<commit_before># Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import unittest

from telemetry import decorators
from telemetry.core.backends.chrome_inspector import inspector_websocket
from telemetry.core.backends.chrome_inspector import websocket
from telemetry.unittest_util import simple_mock


class FakeSocket(object):
  """A fake websocket that allows test to send random data."""
  def __init__(self, mock_timer):
    self._mock_timer = mock_timer
    self._responses = []
    self._timeout = None

  def AddResponse(self, response, time):
    if self._responses:
      assert self._responses[-1][1] < time, (
          'Current response is scheduled earlier than previous response.')
    self._responses.append((response, time))

  def recv(self):
    if not self._responses:
      raise Exception('No more recorded responses.')

    response, time = self._responses.pop(0)
    current_time = self._mock_timer.time()
    if self._timeout is not None and time - current_time > self._timeout:
      self._mock_timer.SetTime(current_time + self._timeout + 1)
      raise websocket.WebSocketTimeoutException()

    self._mock_timer.SetTime(time)
    return response

  def settimeout(self, timeout):
    self._timeout = timeout


def _DoNothingHandler(_elapsed_time):
  pass


class InspectorWebsocketUnittest(unittest.TestCase):

  def setUp(self):
    self._mock_timer = simple_mock.MockTimer(inspector_websocket)

  def tearDown(self):
    self._mock_timer.Restore()

  @decorators.Disabled('chromeos')  # crbug.com/483212
  def testDispatchNotification(self):
    inspector = inspector_websocket.InspectorWebsocket()
    fake_socket = FakeSocket(self._mock_timer)
    # pylint: disable=protected-access
    inspector._socket = fake_socket

    results = []
    def OnTestEvent(result):
      results.append(result)

    inspector.RegisterDomain('Test', OnTestEvent)
    fake_socket.AddResponse('{"method": "Test.foo"}', 5)
    inspector.DispatchNotifications()
    self.assertEqual(1, len(results))
    self.assertEqual('Test.foo', results[0]['method'])<commit_msg>Disable testDispatchNotification.testDispatchNotification on Mac<commit_after># Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import unittest

from telemetry import decorators
from telemetry.core.backends.chrome_inspector import inspector_websocket
from telemetry.core.backends.chrome_inspector import websocket
from telemetry.unittest_util import simple_mock


class FakeSocket(object):
  """A fake websocket that allows test to send random data."""
  def __init__(self, mock_timer):
    self._mock_timer = mock_timer
    self._responses = []
    self._timeout = None

  def AddResponse(self, response, time):
    if self._responses:
      assert self._responses[-1][1] < time, (
          'Current response is scheduled earlier than previous response.')
    self._responses.append((response, time))

  def recv(self):
    if not self._responses:
      raise Exception('No more recorded responses.')

    response, time = self._responses.pop(0)
    current_time = self._mock_timer.time()
    if self._timeout is not None and time - current_time > self._timeout:
      self._mock_timer.SetTime(current_time + self._timeout + 1)
      raise websocket.WebSocketTimeoutException()

    self._mock_timer.SetTime(time)
    return response

  def settimeout(self, timeout):
    self._timeout = timeout


def _DoNothingHandler(_elapsed_time):
  pass


class InspectorWebsocketUnittest(unittest.TestCase):

  def setUp(self):
    self._mock_timer = simple_mock.MockTimer(inspector_websocket)

  def tearDown(self):
    self._mock_timer.Restore()

  @decorators.Disabled('chromeos', 'mac')  # crbug.com/483212, crbug.com/498950
  def testDispatchNotification(self):
    inspector = inspector_websocket.InspectorWebsocket()
    fake_socket = FakeSocket(self._mock_timer)
    # pylint: disable=protected-access
    inspector._socket = fake_socket

    results = []
    def OnTestEvent(result):
      results.append(result)

    inspector.RegisterDomain('Test', OnTestEvent)
    fake_socket.AddResponse('{"method": "Test.foo"}', 5)
    inspector.DispatchNotifications()
    self.assertEqual(1, len(results))
    self.assertEqual('Test.foo', results[0]['method'])
{'fraction_non_alphanumeric': 0.09631791647956893, 'fraction_numerical': 0.00898069151324652, 'mean_word_length': 4.272189349112426, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '7154312', 'n_tokens_mistral': 1347, 'n_tokens_neox': 1281, 'n_words': 324}