<commit_before>#!/usr/bin/python
#  This program is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 2 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU Library General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program; if not, write to the Free Software
#  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
# TestServer.py
# Copyright (C) 2012 <NAME>

import json
import sys
import os
import mimetypes
import textwrap
import urlparse
import inspect
import ResponderTest
import TestDefinitions
import TestRunner
import traceback
from wsgiref.simple_server import make_server
from ola import PidStore
from DMXSender import DMXSender
from ola.ClientWrapper import ClientWrapper
from TestState import TestState
from ola.UID import UID
from optparse import OptionParser, OptionGroup, OptionValueError


__author__ = '<EMAIL> (<NAME>)'


settings = {
  'PORT': 9999,
}

status = {
  '200': '200 OK',
  '403': '403 Forbidden',
  '404': '404 Not Found',
  '500': '500 Internal Server Error',
}

paths = {
  '/RunTests': 'run_tests',
  '/GetDevices': 'get_devices',
  '/GetUnivInfo': 'get_univ_info',
}


"""
  An instance of this class is created to serve every request.
"""
class TestServerApplication(object):
  def __init__(self, environ, start_response):
    self.environ = environ
    self.start = start_response
    self.get_params = {}
    self.response = {}
    self.output = None
    self.headers = []
    self.is_static_request = False
    try:
      self.wrapper = ClientWrapper()
      self.__request_handler()
    except:
      self.status = status['500']
      self.__set_response_status(False)
      self.__set_response_message('Error creating connection with olad. Is it running?')
      print traceback.print_exc()

  def __set_response_status(self, bool_value):
    if type(bool_value) == bool:
      self.response.update({'status': bool_value})

  def __set_response_message(self, message):
    if type(message) == str:
      self.response.update({'message': message})

  def __request_handler(self):
    self.request = self.environ['PATH_INFO']
     
    if self.request.startswith('/static/'):
      self.is_static_request = True
      self.status = status['200']
    elif self.request not in paths.keys():
      self.status = status['404']
    else:
      self.status = status['200']
      self.get_params = urlparse.parse_qs(self.environ['QUERY_STRING'])
      for param in self.get_params:
        self.get_params[param] = self.get_params[param][0]

    return self.__response_handler()

  def __response_handler(self):
    if self.status == status['404']:
      self.__set_response_status(False)
      self.__set_response_message('Invalid request!')

    elif self.status == status['200']:
      if self.is_static_request:
        """
          Remove the first '/' (Makes it easy to partition)
          static/foo/bar partitions to ('static', '/', 'foo/bar')
        """
        resource = self.request[1:]
        resource = resource.partition('/')[2]
        self.__static_content_handler(resource)
      else:
        try:
          self.__getattribute__(paths[self.request])(self.get_params)
        except AttributeError:
          self.status = status['500']
          self.__response_handler()
          print traceback.print_exc()

    elif self.status == status['500']:
      self.__set_response_status(False)
      self.__set_response_message('Error 500: Internal failure')

  def __static_content_handler(self, resource):
    filename = os.path.abspath(os.path.join(settings['www_dir'], resource))

    if not filename.startswith(settings['www_dir']):
      self.status = status['403']
      self.output = 'OLA TestServer: 403: Access denied!'
    elif not os.path.exists(filename) or not os.path.isfile(filename):
      self.status = status['404']
      self.output = 'OLA TestServer: 404: File does not exist!'
    elif not os.access(filename, os.R_OK):
      self.status = status['403']
      self.output = 'OLA TestServer: 403: You do not have permission to access this file.'
    else:
      mimetype, encoding = mimetypes.guess_type(filename)
      if mimetype:
        self.headers.append(('Content-type', mimetype))
      if encoding:
        self.headers.append(('Content-encoding', encoding))

      stats = os.stat(filename)
      self.headers.append(('Content-length', str(stats.st_size)))

      self.output = open(filename, 'rb').read()

  def __get_universes(self):
    global univs
    def format_univ_info(state, universes):
      global univs
      if state.Succeeded():
        univs = [univ.__dict__ for univ in universes]

      self.wrapper.Stop()

    self.wrapper.Client().FetchUniverses(format_univ_info)
    self.wrapper.Run()
    self.wrapper.Reset()
    return univs

  def get_univ_info(self, params):
    universes = self.__get_universes()
    self.__set_response_status(True)
    self.response.update({'universes': universes})

  def run_tests(self, params):
    test_filter = None

    defaults = {
                'u': 0,
                'uid': None,
                'w': 0,
                'f': 0,
                'c': 10,
                't': None,
    }
    defaults.update(params)

    if defaults['uid'] is None:
      self.__set_response_status(False)
      self.__set_response_message('Missing parameter: uid')
      return

    universe = int(defaults['u'])
    if not universe in [univ['_id'] for univ in self.__get_universes()]:
      self.__set_response_status(False)
      self.__set_response_message('Universe %d doesn\'t exist' % (universe))
      return<commit_msg>Added GetTestDefs<commit_after>#!/usr/bin/python
#  This program is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 2 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU Library General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program; if not, write to the Free Software
#  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
# TestServer.py
# Copyright (C) 2012 <NAME>

import json
import sys
import os
import mimetypes
import textwrap
import urlparse
import inspect
import ResponderTest
import TestDefinitions
import TestRunner
import traceback
from wsgiref.simple_server import make_server
from ola import PidStore
from DMXSender import DMXSender
from ola.ClientWrapper import ClientWrapper
from TestState import TestState
from ola.UID import UID
from optparse import OptionParser, OptionGroup, OptionValueError


__author__ = '<EMAIL> (<NAME>)'


settings = {
  'PORT': 9999,
}

status = {
  '200': '200 OK',
  '403': '403 Forbidden',
  '404': '404 Not Found',
  '500': '500 Internal Server Error',
}

paths = {
  '/RunTests': 'run_tests',
  '/GetDevices': 'get_devices',
  '/GetUnivInfo': 'get_univ_info',
  '/GetTestDefs': 'get_test_definitions',
}


"""
  An instance of this class is created to serve every request.
"""
class TestServerApplication(object):
  def __init__(self, environ, start_response):
    self.environ = environ
    self.start = start_response
    self.get_params = {}
    self.response = {}
    self.output = None
    self.headers = []
    self.is_static_request = False
    try:
      self.wrapper = ClientWrapper()
      self.__request_handler()
    except:
      self.status = status['500']
      self.__set_response_status(False)
      self.__set_response_message('Error creating connection with olad. Is it running?')
      print traceback.print_exc()

  def __set_response_status(self, bool_value):
    if type(bool_value) == bool:
      self.response.update({'status': bool_value})

  def __set_response_message(self, message):
    if type(message) == str:
      self.response.update({'message': message})

  def __request_handler(self):
    self.request = self.environ['PATH_INFO']
     
    if self.request.startswith('/static/'):
      self.is_static_request = True
      self.status = status['200']
    elif self.request not in paths.keys():
      self.status = status['404']
    else:
      self.status = status['200']
      self.get_params = urlparse.parse_qs(self.environ['QUERY_STRING'])
      for param in self.get_params:
        self.get_params[param] = self.get_params[param][0]

    return self.__response_handler()

  def __response_handler(self):
    if self.status == status['404']:
      self.__set_response_status(False)
      self.__set_response_message('Invalid request!')

    elif self.status == status['200']:
      if self.is_static_request:
        """
          Remove the first '/' (Makes it easy to partition)
          static/foo/bar partitions to ('static', '/', 'foo/bar')
        """
        resource = self.request[1:]
        resource = resource.partition('/')[2]
        self.__static_content_handler(resource)
      else:
        try:
          self.__getattribute__(paths[self.request])(self.get_params)
        except AttributeError:
          self.status = status['500']
          self.__response_handler()
          print traceback.print_exc()

    elif self.status == status['500']:
      self.__set_response_status(False)
      self.__set_response_message('Error 500: Internal failure')

  def __static_content_handler(self, resource):
    filename = os.path.abspath(os.path.join(settings['www_dir'], resource))

    if not filename.startswith(settings['www_dir']):
      self.status = status['403']
      self.output = 'OLA TestServer: 403: Access denied!'
    elif not os.path.exists(filename) or not os.path.isfile(filename):
      self.status = status['404']
      self.output = 'OLA TestServer: 404: File does not exist!'
    elif not os.access(filename, os.R_OK):
      self.status = status['403']
      self.output = 'OLA TestServer: 403: You do not have permission to access this file.'
    else:
      mimetype, encoding = mimetypes.guess_type(filename)
      if mimetype:
        self.headers.append(('Content-type', mimetype))
      if encoding:
        self.headers.append(('Content-encoding', encoding))

      stats = os.stat(filename)
      self.headers.append(('Content-length', str(stats.st_size)))

      self.output = open(filename, 'rb').read()

  def __get_universes(self):
    global univs
    def format_univ_info(state, universes):
      global univs
      if state.Succeeded():
        univs = [univ.__dict__ for univ in universes]

      self.wrapper.Stop()

    self.wrapper.Client().FetchUniverses(format_univ_info)
    self.wrapper.Run()
    self.wrapper.Reset()
    return univs

  def get_univ_info(self, params):
    universes = self.__get_universes()
    self.__set_response_status(True)
    self.response.update({'universes': universes})

  def get_test_definitions(self, params):
    self.__set_response_status(True)
    tests_by_category = self.__get_test_defs_by_catg()
    self.response.update({'test_defs': tests_by_category})

  def __get_test_defs_by_catg(self):

    tests = []
    for symbol in dir(TestDefinitions):
      obj = getattr(TestDefinitions, symbol)
      if not inspect.isclass(obj):
        continue
      if (obj == ResponderTest.ResponderTestFixture or
          obj == ResponderTest.OptionalParameterTestFixture):
        continue
      if issubclass(obj, ResponderTest.ResponderTestFixture):
        tests.append(obj)

    tests_by_category = {}
    for test in tests:
      tests_by_category.setdefault(test.CATEGORY.__str__(), []) \
                                  .append(test.__name__)
    return tests_by_category

  def run_tests(self, params):
    test_filter = None

    defaults = {
                'u': 0,
                'uid': None,
                'w': 0,
                'f': 0,
                'c': 10,
                't': None,
    }
    defaults.update(params)

    if defaults['uid'] is None:
      self.__set_response_status(False)
      self.__set_response_message('Missing parameter: uid')
      return

    universe = int(defaults['u'])
    if not universe in [univ['_id'] for univ in self.__get_universes()]:
      self.__set_response_status(False)
      self.__set_response_message('Universe %d doesn\'t exist' % (universe))
      return
{'fraction_non_alphanumeric': 0.11107649166536844, 'fraction_numerical': 0.015578750584203147, 'mean_word_length': 3.173927178153446, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1968534', 'n_tokens_mistral': 4034, 'n_tokens_neox': 3821, 'n_words': 1024}