from __future__ import absolute_import, division, print_function, \
    with_statement

import logging
import os.path
import time

import tornado.escape
import tornado.gen
import tornado.ioloop
from tornado.test.util import unittest
from tornado.testing import AsyncHTTPTestCase, gen_test
import tornado.web

from pilbox.app import PilboxApplication
from pilbox.errors import SignatureError, ClientError, HostError, \
    BackgroundError, DimensionsError, FilterError, FormatError, ModeError, \
    PositionError, QualityError, UrlError, ImageFormatError, FetchError
from pilbox.signature import sign
from pilbox.test import image_test

try:
    from urllib import urlencode
except ImportError:
    from urllib.parse import urlencode

try:
    import cv
except ImportError:
    cv = None


logger = logging.getLogger("tornado.application")


class _AppAsyncMixin(object):
    def fetch_error(self, code, *args, **kwargs):
        response = self.fetch(*args, **kwargs)
        self.assertEqual(response.code, code)
        self.assertEqual(response.headers.get("Content-Type", None),
                         "application/json")
        return tornado.escape.json_decode(response.body)

    def fetch_success(self, *args, **kwargs):
        response = self.fetch(*args, **kwargs)
        self.assertEqual(response.code, 200)
        return response

    def get_image_resize_cases(self):
        cases = image_test.get_image_resize_cases()
        m = dict(background="bg", filter="filter", format="fmt",
                 position="pos", quality="q")
        for i, case in enumerate(cases):
            path = "/test/data/%s" % os.path.basename(case["source_path"])
            cases[i]["source_query_params"] = dict(
                url=self.get_url(path),
                w=case["width"] or "",
                h=case["height"] or "",
                mode=case["mode"])
            for k in m.keys():
                if k in case:
                    cases[i]["source_query_params"][m.get(k)] = case[k]
            if case.get("format") in ["jpeg", "jpg"]:
                cases[i]["content_type"] = "image/jpeg"
            elif case.get("format") == "png":
                cases[i]["content_type"] = "image/png"
            elif case.get("format") == "webp":
                cases[i]["content_type"] = "image/webp"
            else:
                cases[i]["content_type"] = None
        return cases


class _PilboxTestApplication(PilboxApplication):
    def get_handlers(self):
        path = os.path.join(os.path.dirname(__file__), "data")
        handlers = [(r"/test/data/test-delayed.jpg", _DelayedHandler),
                    (r"/test/data/(.*)",
                     tornado.web.StaticFileHandler,
                     {"path": path})]
        handlers.extend(super(_PilboxTestApplication, self).get_handlers())
        return handlers


class _DelayedHandler(tornado.web.RequestHandler):

    @tornado.web.asynchronous
    @tornado.gen.engine
    def get(self):
        delay = time.time() + float(self.get_argument("delay", 0.0))
        yield tornado.gen.Task(
            tornado.ioloop.IOLoop.instance().add_timeout, delay)
        self.finish()


class AppTest(AsyncHTTPTestCase, _AppAsyncMixin):
    def get_app(self):
        return _PilboxTestApplication()

    def test_missing_url(self):
        qs = urlencode(dict(w=1, h=1))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), UrlError.get_code())

    def test_missing_dimensions(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), DimensionsError.get_code())

    def test_invalid_width(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w="a", h=1))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), DimensionsError.get_code())

    def test_invalid_height(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h="a"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), DimensionsError.get_code())

    def test_invalid_mode(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1, mode="foo"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), ModeError.get_code())

    def test_invalid_hexadecimal_background(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1,
                            mode="fill", bg="r"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), BackgroundError.get_code())

    def test_invalid_long_background(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1,
                            mode="fill", bg="0f0f0f0f0"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), BackgroundError.get_code())

    def test_invalid_position(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1, pos="foo"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), PositionError.get_code())

    def test_invalid_filter(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1, filter="bar"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), FilterError.get_code())

    def test_invalid_format(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1, fmt="foo"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), FormatError.get_code())

    def test_invalid_integer_quality(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1, q="a"))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), QualityError.get_code())

    def test_outofbounds_quality(self):
        qs = urlencode(dict(url="http://foo.co/x.jpg", w=1, h=1, q=200))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), QualityError.get_code())

    def test_unsupported_image_format(self):
        path = "/test/data/test-bad-format.gif"
        qs = urlencode(dict(url=self.get_url(path), w=1, h=1))
        resp = self.fetch_error(415, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), ImageFormatError.get_code())

    def test_not_found(self):
        path = "/test/data/test-not-found.jpg"
        qs = urlencode(dict(url=self.get_url(path), w=1, h=1))
        resp = self.fetch_error(404, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), FetchError.get_code())

    def test_not_connect(self):
        qs = urlencode(dict(url="http://a.com/a.jpg", w=1, h=1))
        resp = self.fetch_error(404, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), FetchError.get_code())

    def test_invalid_protocol(self):
        path = os.path.join(os.path.dirname(__file__), "data", "test1.jpg")
        qs = urlencode(dict(url="file://%s" % path, w=1, h=1))
        resp = self.fetch_error(400, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), UrlError.get_code())

    def test_valid(self):
        cases = self.get_image_resize_cases()
        for case in cases:
            if case.get("mode") == "crop" and case.get("position") == "face":
                continue
            self._assert_expected_resize(case)

    @unittest.skipIf(cv is None, "OpenCV is not installed")
    def test_valid_face(self):
        cases = self.get_image_resize_cases()
        for case in cases:
            if case.get("mode") == "crop" and case.get("position") == "face":
                self._assert_expected_resize(case)

    def _assert_expected_resize(self, case):
        qs = urlencode(case["source_query_params"])
        resp = self.fetch_success("/?%s" % qs)
        msg = "/?%s does not match %s" \
            % (qs, case["expected_path"])
        if case["content_type"]:
            self.assertEqual(resp.headers.get("Content-Type", None),
                             case["content_type"])
        with open(case["expected_path"], "rb") as expected:
            self.assertEqual(resp.buffer.read(), expected.read(), msg)


class AppRestrictedTest(AsyncHTTPTestCase, _AppAsyncMixin):
    KEY = "abcdef"
    NAME = "abc"

    def get_app(self):
        return _PilboxTestApplication(
            client_name=self.NAME,
            client_key=self.KEY,
            allowed_hosts=["foo.co", "bar.io", "localhost"])

    def test_missing_client_name(self):
        params = dict(url="http://foo.co/x.jpg", w=1, h=1)
        qs = sign(self.KEY, urlencode(params))
        resp = self.fetch_error(403, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), ClientError.get_code())

    def test_bad_client_name(self):
        params = dict(url="http://foo.co/x.jpg", w=1, h=1, client="123")
        qs = sign(self.KEY, urlencode(params))
        resp = self.fetch_error(403, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), ClientError.get_code())

    def test_missing_signature(self):
        params = dict(url="http://foo.co/x.jpg", w=1, h=1, client=self.NAME)
        qs = urlencode(params)
        resp = self.fetch_error(403, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), SignatureError.get_code())

    def test_bad_signature(self):
        params = dict(url="http://foo.co/x.jpg", w=1, h=1,
                      client=self.NAME, sig="abc123")
        qs = urlencode(params)
        resp = self.fetch_error(403, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), SignatureError.get_code())

    def test_bad_host(self):
        params = dict(url="http://bar.co/x.jpg", w=1, h=1, client=self.NAME)
        qs = sign(self.KEY, urlencode(params))
        resp = self.fetch_error(403, "/?%s" % qs)
        self.assertEqual(resp.get("error_code"), HostError.get_code())

    def test_valid(self):
        cases = self.get_image_resize_cases()
        for case in cases:
            if case.get("mode") == "crop" and case.get("position") == "face":
                continue
            params = case["source_query_params"]
            params["client"] = self.NAME
            qs = sign(self.KEY, urlencode(params))
            resp = self.fetch_success("/?%s" % qs)
            msg = "/?%s does not match %s" \
                % (qs, case["expected_path"])
            with open(case["expected_path"], "rb") as expected:
                self.assertEqual(resp.buffer.read(), expected.read(), msg)


class AppSlowTest(AsyncHTTPTestCase, _AppAsyncMixin):
    def get_app(self):
        return _PilboxTestApplication(timeout=0.5)

    def test_timeout(self):
        url = self.get_url("/test/data/test-delayed.jpg?delay=1.0")
        qs = urlencode(dict(url=url, w=1, h=1))
        resp = self.fetch_error(404, "/?%s" %qs)
        self.assertEqual(resp.get("error_code"), FetchError.get_code())

[((799, 839), 'logging.getLogger', 'logging.getLogger', (['"""tornado.application"""'], {}), "('tornado.application')\n", (816, 839), False, 'import logging\n'), ((7495, 7549), 'tornado.test.util.unittest.skipIf', 'unittest.skipIf', (['(cv is None)', '"""OpenCV is not installed"""'], {}), "(cv is None, 'OpenCV is not installed')\n", (7510, 7549), False, 'from tornado.test.util import unittest\n'), ((1404, 1439), 'pilbox.test.image_test.get_image_resize_cases', 'image_test.get_image_resize_cases', ([], {}), '()\n', (1437, 1439), False, 'from pilbox.test import image_test\n'), ((7842, 7880), 'urllib.parse.urlencode', 'urlencode', (["case['source_query_params']"], {}), "(case['source_query_params'])\n", (7851, 7880), False, 'from urllib.parse import urlencode\n'), ((9264, 9281), 'urllib.parse.urlencode', 'urlencode', (['params'], {}), '(params)\n', (9273, 9281), False, 'from urllib.parse import urlencode\n'), ((9569, 9586), 'urllib.parse.urlencode', 'urlencode', (['params'], {}), '(params)\n', (9578, 9586), False, 'from urllib.parse import urlencode\n'), ((2994, 3005), 'time.time', 'time.time', ([], {}), '()\n', (3003, 3005), False, 'import time\n'), ((3452, 3471), 'pilbox.errors.UrlError.get_code', 'UrlError.get_code', ([], {}), '()\n', (3469, 3471), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((3668, 3694), 'pilbox.errors.DimensionsError.get_code', 'DimensionsError.get_code', ([], {}), '()\n', (3692, 3694), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((3898, 3924), 'pilbox.errors.DimensionsError.get_code', 'DimensionsError.get_code', ([], {}), '()\n', (3922, 3924), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((4129, 4155), 'pilbox.errors.DimensionsError.get_code', 'DimensionsError.get_code', ([], {}), '()\n', (4153, 4155), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((4368, 4388), 'pilbox.errors.ModeError.get_code', 'ModeError.get_code', ([], {}), '()\n', (4386, 4388), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((4656, 4682), 'pilbox.errors.BackgroundError.get_code', 'BackgroundError.get_code', ([], {}), '()\n', (4680, 4682), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((4951, 4977), 'pilbox.errors.BackgroundError.get_code', 'BackgroundError.get_code', ([], {}), '()\n', (4975, 4977), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((5193, 5217), 'pilbox.errors.PositionError.get_code', 'PositionError.get_code', ([], {}), '()\n', (5215, 5217), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((5434, 5456), 'pilbox.errors.FilterError.get_code', 'FilterError.get_code', ([], {}), '()\n', (5454, 5456), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((5670, 5692), 'pilbox.errors.FormatError.get_code', 'FormatError.get_code', ([], {}), '()\n', (5690, 5692), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((5911, 5934), 'pilbox.errors.QualityError.get_code', 'QualityError.get_code', ([], {}), '()\n', (5932, 5934), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((6149, 6172), 'pilbox.errors.QualityError.get_code', 'QualityError.get_code', ([], {}), '()\n', (6170, 6172), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((6430, 6457), 'pilbox.errors.ImageFormatError.get_code', 'ImageFormatError.get_code', ([], {}), '()\n', (6455, 6457), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((6699, 6720), 'pilbox.errors.FetchError.get_code', 'FetchError.get_code', ([], {}), '()\n', (6718, 6720), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((6919, 6940), 'pilbox.errors.FetchError.get_code', 'FetchError.get_code', ([], {}), '()\n', (6938, 6940), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((7218, 7237), 'pilbox.errors.UrlError.get_code', 'UrlError.get_code', ([], {}), '()\n', (7235, 7237), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((8713, 8730), 'urllib.parse.urlencode', 'urlencode', (['params'], {}), '(params)\n', (8722, 8730), False, 'from urllib.parse import urlencode\n'), ((8831, 8853), 'pilbox.errors.ClientError.get_code', 'ClientError.get_code', ([], {}), '()\n', (8851, 8853), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((8993, 9010), 'urllib.parse.urlencode', 'urlencode', (['params'], {}), '(params)\n', (9002, 9010), False, 'from urllib.parse import urlencode\n'), ((9111, 9133), 'pilbox.errors.ClientError.get_code', 'ClientError.get_code', ([], {}), '()\n', (9131, 9133), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((9381, 9406), 'pilbox.errors.SignatureError.get_code', 'SignatureError.get_code', ([], {}), '()\n', (9404, 9406), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((9686, 9711), 'pilbox.errors.SignatureError.get_code', 'SignatureError.get_code', ([], {}), '()\n', (9709, 9711), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((9848, 9865), 'urllib.parse.urlencode', 'urlencode', (['params'], {}), '(params)\n', (9857, 9865), False, 'from urllib.parse import urlencode\n'), ((9966, 9986), 'pilbox.errors.HostError.get_code', 'HostError.get_code', ([], {}), '()\n', (9984, 9986), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((10986, 11007), 'pilbox.errors.FetchError.get_code', 'FetchError.get_code', ([], {}), '()\n', (11005, 11007), False, 'from pilbox.errors import SignatureError, ClientError, HostError, BackgroundError, DimensionsError, FilterError, FormatError, ModeError, PositionError, QualityError, UrlError, ImageFormatError, FetchError\n'), ((10313, 10330), 'urllib.parse.urlencode', 'urlencode', (['params'], {}), '(params)\n', (10322, 10330), False, 'from urllib.parse import urlencode\n')]