<commit_before># -*- coding: utf-8 -*-
import unittest
import json
import base64
import urllib
from functools import wraps
from cStringIO import StringIO

import pytest
import requests
from PIL import Image, ImageChops
from splash import defaults
from splash.utils import truncated
from splash.tests.utils import NON_EXISTING_RESOLVABLE, SplashServer


def https_only(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            if self.__class__.https_supported:
                func(self, *args, **kwargs)
        except AttributeError:
            func(self, *args, **kwargs)
    return wrapper


def skip_proxy(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            if self.__class__.proxy_test is False:
                func(self, *args, **kwargs)
        except AttributeError:
            func(self, *args, **kwargs)
    return wrapper


class DirectRequestHandler(object):

    endpoint = "render.html"

    def __init__(self, ts):
        self.ts = ts

    @property
    def host(self):
        return "localhost:%s" % self.ts.splashserver.portnum

    def request(self, query, endpoint=None, headers=None):
        url, params = self._url_and_params(endpoint, query)
        return requests.get(url, params=params, headers=headers)

    def post(self, query, endpoint=None, payload=None, headers=None):
        url, params = self._url_and_params(endpoint, query)
        return requests.post(url, params=params, data=payload, headers=headers)

    def _url_and_params(self, endpoint, query):
        endpoint = endpoint if endpoint is not None else self.endpoint
        if isinstance(query, dict):
            url = "http://%s/%s" % (self.host, endpoint)
            params = query
        else:
            url = "http://%s/%s?%s" % (self.host, endpoint, query)
            params = None
        return url, params


@pytest.mark.usefixtures("class_ts")
class BaseRenderTest(unittest.TestCase):

    endpoint = "render.html"
    request_handler = DirectRequestHandler
    use_gzip = False

    def setUp(self):
        if self.use_gzip:
            try:
                from twisted.web.server import GzipEncoderFactory
            except ImportError:
                pytest.skip("Gzip support is not available in old Twisted")

    def mockurl(self, path, host='localhost'):
        return self.ts.mockserver.url(path, self.use_gzip, host=host)

    def _get_handler(self):
        handler = self.request_handler(self.ts)
        handler.endpoint = self.endpoint
        return handler

    def request(self, query, endpoint=None, headers=None, **kwargs):
        return self._get_handler().request(query, endpoint, headers, **kwargs)

    def post(self, query, endpoint=None, payload=None, headers=None, **kwargs):
        return self._get_handler().post(query, endpoint, payload, headers, **kwargs)

    def assertStatusCode(self, response, code):
        msg = (response.status_code, truncated(response.content, 1000))
        self.assertEqual(response.status_code, code, msg)

    def assertPng(self, response, width=None, height=None):
        self.assertStatusCode(response, 200)
        self.assertEqual(response.headers["content-type"], "image/png")
        img = Image.open(StringIO(response.content))
        self.assertEqual(img.format, "PNG")
        if width is not None:
            self.assertEqual(img.size[0], width)
        if height is not None:
            self.assertEqual(img.size[1], height)
        return img

    COLOR_NAMES = ('red', 'green', 'blue', 'alpha')

    def assertBoxColor(self, response, box, etalon):
        assert len(etalon) == 4  # RGBA components
        img = Image.open(StringIO(response.content))
        extrema = img.crop(box).getextrema()
        for (color_name, (min_val, max_val)) in zip(self.COLOR_NAMES, extrema):
            self.assertEqual(
                min_val, max_val,
                "Selected region does not have same values in %s component:"
                "%s -> %s" % (color_name, min_val, max_val))
        color = tuple(e[0] for e in extrema)
        self.assertEqual(color, etalon,
                         "Region color (%s) doesn't match the etalon (%s)" %
                         (color, etalon))


class Base(object):
    # a hack to skip running of a base RenderTest

    class RenderTest(BaseRenderTest):

        @unittest.skipIf(NON_EXISTING_RESOLVABLE, "non existing hosts are resolvable")
        def test_render_error(self):
            r = self.request({"url": "http://non-existent-host/"})
            self.assertStatusCode(r, 502)

        def test_timeout(self):
            r = self.request({"url": self.mockurl("delay?n=10"), "timeout": "0.5"})
            self.assertStatusCode(r, 504)

        def test_timeout_out_of_range(self):
            r = self.request({"url": self.mockurl("delay?n=10"), "timeout": "999"})
            self.assertStatusCode(r, 400)

        @skip_proxy
        def test_missing_url(self):
            r = self.request({})
            self.assertStatusCode(r, 400)
            self.assertTrue("url" in r.text)

        def test_jsalert(self):
            r = self.request({"url": self.mockurl("jsalert"), "timeout": "3"})
            self.assertStatusCode(r, 200)

        def test_jsconfirm(self):
            r = self.request({"url": self.mockurl("jsconfirm"), "timeout": "3"})
            self.assertStatusCode(r, 200)

        def test_iframes(self):
            r = self.request({"url": self.mockurl("iframes"), "timeout": "3"})
            self.assertStatusCode(r, 200)

        def test_wait(self):
            r1 = self.request({"url": self.mockurl("jsinterval")})
            r2 = self.request({"url": self.mockurl("jsinterval")})
            r3 = self.request({"url": self.mockurl("jsinterval"), "wait": "0.2"})
            self.assertStatusCode(r1, 200)
            self.assertStatusCode(r2, 200)
            self.assertStatusCode(r3, 200)
            self.assertEqual(r1.content, r2.content)
            self.assertNotEqual(r1.content, r3.content)


class RenderHtmlTest(Base.RenderTest):

    endpoint = "render.html"

    def test_jsrender(self):
        self._test_jsrender(self.mockurl("jsrender"))

    @https_only
    def test_jsrender_https(self):<commit_msg>added a small test for invalid wait argument<commit_after># -*- coding: utf-8 -*-
import unittest
import json
import base64
import urllib
from functools import wraps
from cStringIO import StringIO

import pytest
import requests
from PIL import Image, ImageChops
from splash import defaults
from splash.utils import truncated
from splash.tests.utils import NON_EXISTING_RESOLVABLE, SplashServer


def https_only(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            if self.__class__.https_supported:
                func(self, *args, **kwargs)
        except AttributeError:
            func(self, *args, **kwargs)
    return wrapper


def skip_proxy(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            if self.__class__.proxy_test is False:
                func(self, *args, **kwargs)
        except AttributeError:
            func(self, *args, **kwargs)
    return wrapper


class DirectRequestHandler(object):

    endpoint = "render.html"

    def __init__(self, ts):
        self.ts = ts

    @property
    def host(self):
        return "localhost:%s" % self.ts.splashserver.portnum

    def request(self, query, endpoint=None, headers=None):
        url, params = self._url_and_params(endpoint, query)
        return requests.get(url, params=params, headers=headers)

    def post(self, query, endpoint=None, payload=None, headers=None):
        url, params = self._url_and_params(endpoint, query)
        return requests.post(url, params=params, data=payload, headers=headers)

    def _url_and_params(self, endpoint, query):
        endpoint = endpoint if endpoint is not None else self.endpoint
        if isinstance(query, dict):
            url = "http://%s/%s" % (self.host, endpoint)
            params = query
        else:
            url = "http://%s/%s?%s" % (self.host, endpoint, query)
            params = None
        return url, params


@pytest.mark.usefixtures("class_ts")
class BaseRenderTest(unittest.TestCase):

    endpoint = "render.html"
    request_handler = DirectRequestHandler
    use_gzip = False

    def setUp(self):
        if self.use_gzip:
            try:
                from twisted.web.server import GzipEncoderFactory
            except ImportError:
                pytest.skip("Gzip support is not available in old Twisted")

    def mockurl(self, path, host='localhost'):
        return self.ts.mockserver.url(path, self.use_gzip, host=host)

    def _get_handler(self):
        handler = self.request_handler(self.ts)
        handler.endpoint = self.endpoint
        return handler

    def request(self, query, endpoint=None, headers=None, **kwargs):
        return self._get_handler().request(query, endpoint, headers, **kwargs)

    def post(self, query, endpoint=None, payload=None, headers=None, **kwargs):
        return self._get_handler().post(query, endpoint, payload, headers, **kwargs)

    def assertStatusCode(self, response, code):
        msg = (response.status_code, truncated(response.content, 1000))
        self.assertEqual(response.status_code, code, msg)

    def assertPng(self, response, width=None, height=None):
        self.assertStatusCode(response, 200)
        self.assertEqual(response.headers["content-type"], "image/png")
        img = Image.open(StringIO(response.content))
        self.assertEqual(img.format, "PNG")
        if width is not None:
            self.assertEqual(img.size[0], width)
        if height is not None:
            self.assertEqual(img.size[1], height)
        return img

    COLOR_NAMES = ('red', 'green', 'blue', 'alpha')

    def assertBoxColor(self, response, box, etalon):
        assert len(etalon) == 4  # RGBA components
        img = Image.open(StringIO(response.content))
        extrema = img.crop(box).getextrema()
        for (color_name, (min_val, max_val)) in zip(self.COLOR_NAMES, extrema):
            self.assertEqual(
                min_val, max_val,
                "Selected region does not have same values in %s component:"
                "%s -> %s" % (color_name, min_val, max_val))
        color = tuple(e[0] for e in extrema)
        self.assertEqual(color, etalon,
                         "Region color (%s) doesn't match the etalon (%s)" %
                         (color, etalon))


class Base(object):
    # a hack to skip running of a base RenderTest

    class RenderTest(BaseRenderTest):

        @unittest.skipIf(NON_EXISTING_RESOLVABLE, "non existing hosts are resolvable")
        def test_render_error(self):
            r = self.request({"url": "http://non-existent-host/"})
            self.assertStatusCode(r, 502)

        def test_timeout(self):
            r = self.request({"url": self.mockurl("delay?n=10"), "timeout": "0.5"})
            self.assertStatusCode(r, 504)

        def test_timeout_out_of_range(self):
            r = self.request({"url": self.mockurl("delay?n=10"), "timeout": "999"})
            self.assertStatusCode(r, 400)

        @skip_proxy
        def test_missing_url(self):
            r = self.request({})
            self.assertStatusCode(r, 400)
            self.assertTrue("url" in r.text)

        def test_jsalert(self):
            r = self.request({"url": self.mockurl("jsalert"), "timeout": "3"})
            self.assertStatusCode(r, 200)

        def test_jsconfirm(self):
            r = self.request({"url": self.mockurl("jsconfirm"), "timeout": "3"})
            self.assertStatusCode(r, 200)

        def test_iframes(self):
            r = self.request({"url": self.mockurl("iframes"), "timeout": "3"})
            self.assertStatusCode(r, 200)

        def test_wait(self):
            r1 = self.request({"url": self.mockurl("jsinterval")})
            r2 = self.request({"url": self.mockurl("jsinterval")})
            r3 = self.request({"url": self.mockurl("jsinterval"), "wait": "0.2"})
            self.assertStatusCode(r1, 200)
            self.assertStatusCode(r2, 200)
            self.assertStatusCode(r3, 200)
            self.assertEqual(r1.content, r2.content)
            self.assertNotEqual(r1.content, r3.content)

        def test_invalid_wait(self):
            for wait in ['foo', '11', '11.0']:
                r = self.request({'url': self.mockurl("jsrender"),
                                  'wait': wait})
                self.assertStatusCode(r, 400)


class RenderHtmlTest(Base.RenderTest):

    endpoint = "render.html"

    def test_jsrender(self):
        self._test_jsrender(self.mockurl("jsrender"))

    @https_only
    def test_jsrender_https(self):
{'fraction_non_alphanumeric': 0.12314771486507566, 'fraction_numerical': 0.011230697239120261, 'mean_word_length': 2.5609552901971675, 'pattern_counts': {'":': 30, '<': 3, '<?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': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6931050', 'n_tokens_mistral': 3935, 'n_tokens_neox': 3766, 'n_words': 907}