#!/usr/bin/env python
# -*- coding: utf-8 -*-
# $Id: RequestTestBase.py 9623 2016-04-08 05:16:08Z Eric $
#
# Copyright (c) 2016 Nuwa Information Co., Ltd, All Rights Reserved.
#
# Licensed under the Proprietary License,
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at our web site.
#
# See the License for the specific language governing permissions and
# limitations under the License.
#
# $Author: Eric $
# $Date: 2016-04-08 13:16:08 +0800 (Fri, 08 Apr 2016) $
# $Revision: 9623 $
"""
Test Request general implementation.
"""

#import sys
import os
import subprocess

import unittest

import threading
import time

from Queue import Queue
q = Queue()

class DoThread(threading.Thread):
    """
    Thread for test.
    """

    def __init__(self, url, testcase):
        self.url = url
        self.testcase = testcase
        self.result = None
        threading.Thread.__init__ (self)

    def run(self):
        q.get()
        try:
            self.result = self.testcase.do(self.url)
        finally: # We must always make task_done or it might hang forever.
            q.task_done()

default = 'trident'

class _RequestTest(unittest.TestCase):
    """
    Test case for Request.
    """

    TEST_HOST_URL = 'http://127.0.0.1:8000'

    # Test URLs.
    URLS = [
        '%s/' % TEST_HOST_URL,
        '%s/ext-2.0/examples/window/hello.html' % TEST_HOST_URL,
        '%s/ext-2.0/examples/layout/accordion.html' % TEST_HOST_URL,
        '%s/ext-2.0/examples/layout/anchor.html' % TEST_HOST_URL,
        '%s/ext-2.0/examples/layout/table.html' % TEST_HOST_URL,
    ]
    
    def setUp(self):            
        file = os.path.abspath(os.path.join(os.path.dirname( __file__ ),
                                            'StartCherrypyServer.py'))
        
        self.server = subprocess.Popen(['python', file],
                                       stdin=subprocess.PIPE,
                                       stdout=subprocess.PIPE,
                                       stderr=subprocess.PIPE, shell=True)
        
    def tearDown(self):
        self.server.terminate()
        self.server.kill()
    
    def do(self, _url, impl=default, options=None):
        """
        Do fetch.

        @param _url URL.
        @param impl Implementation.
        @param options Options.
        @return Result.
        """
        from Iuno import browser

        t = threading.currentThread()
        print t, hex(id(t)), str(time.time()), _url
        #try:
        if True:
            if options:
                r = browser.fetch(_url, impl, options)
            else:
                r = browser.fetch(_url, impl)
            print t, hex(id(t)), str(time.time()), _url, 'RETURNED'
            return r
        #except:
            #print 'Exception happened:', sys.exc_info()

    def openAssertFile(self, url):
        """
        Open assert file by URL.

        @param url URL.
        @return File object.
        """
        import hashlib

        f = os.path.join(self.ASSERT_DATA_DIR,
                         '%s.html' % hashlib.md5(url).hexdigest())
        return open(f, 'rb')

    def _writeDebugResult(self, f, src):
        """
        Write the result to debug purpose file.

        @param f Opened assert file object.
        @param src HTML source.
        """
        # Debug purpose.
        parts = list(os.path.split(f.name))
        parts[-1] = '_%s' % parts[-1]
        with open(os.path.join(*parts), 'wb') as f2:
            f2.write(src)

    def _nz(self, html):
        """
        Normalize got HTML source (strip \r, \n).

        @param html HTML source.
        @return Normalized source.
        """
        return html

    def testSingleThread(self):
        """
        Test with single thread case (simplest case).
        """
        if not hasattr(self, 'ASSERT_DATA_DIR'):
            return

        _url = self.TEST_HOST_URL + '/'
        result = self.do(_url)

        self.assertEqual(result.url, _url)
        #self.assertEqual(result.lastModified, '11/20/2004 07:16:26')
        self.assertEqual(result.content,
            "<HTML><HEAD></HEAD>\r\n<BODY>\r\n<H1>It works!</H1></BODY></HTML>")

    def testSingleThreadMultiTimes(self):
        """
        Test with single thread and call it multiple times.
        """
        if not hasattr(self, 'ASSERT_DATA_DIR'):
            return

        for u in self.URLS:
            result = self.do(u)
            self.assertEqual(result.url, u)
            with self.openAssertFile(u) as f:
                self.assertEqual(self._nz(result.content), self._nz(f.read()))
                self._writeDebugResult(f, result.content)

    def doMultithread(self):
        """
        Do fetch with multithread.

        @return Results.
        """
        global q

        results = []

        threads = []
        for u in self.URLS:
            q.put(u)
            t = DoThread(u, self)
            threads.append(t)
            t.start()

        q.join()

        q = Queue()

        for t in threads:
            result = []
            result.append((t.result.url, t.url))
            with self.openAssertFile(t.url) as f:
                result.append((self._nz(t.result.content),
                               self._nz(f.read())))
                results.append(result)

                self._writeDebugResult(f, t.result.content)

        return results

    def testMultiThread(self):
        """
        Test with multithread.
        """
        if not hasattr(self, 'ASSERT_DATA_DIR'):
            return

        results = self.doMultithread()
        for r in results:
            for actual, expect in r:
                self.assertEqual(actual, expect)

    def testMultiThreadMultiTimes(self):
        """
        Test with multithread for multiple times.
        """
        if not hasattr(self, 'ASSERT_DATA_DIR'):
            return

        for x in range(5):
            results = self.doMultithread()
            for r in results:
                for actual, expect in r:
                    self.assertEqual(actual, expect)

    def testRSS(self):
        """
        Test RSS links.
        """
        if not hasattr(self, 'ASSERT_DATA_DIR'):
            return

        import feedparser

        _url = 'http://bw.businessweekly.com.tw/feedsec.php?feedid=0'
        result = self.do(_url)

        self.assertEqual(result.url, _url)
        self.assertEqual(result.snapshot, "")

        rss = feedparser.parse(result.content)
        self.assertTrue(rss.feed != {})

        # This is very strange that the feedparser.parse get different XML from
        # server with browser.
        #rss2 = feedparser.parse(_url)
        #self.assertEqual(rss.feed, rss2.feed)
        # This might changed between first and second fetch.
        # But we assume that will not happen too frequently.
        #self.assertEqual(rss, rss2)

        _url = 'http://tw.news.yahoo.com/rss/'
        result = self.do(_url)

        self.assertEqual(result.url, _url)
        self.assertEqual(result.snapshot, "")

        rss = feedparser.parse(result.content)
        self.assertTrue(rss.feed != {})

    def _testPost(self):
        """
        Test send POST request.
        """
        result = self.do('http://httpbin.org/post', options={
            'post': {
                'abc': 123,
                'def': '+-*/',
            },
        })

        self.assertTrue('"abc": "123"' in result.content)
        self.assertTrue('"def": "+-*/"' in result.content)

    def _testCookies(self):
        """
        Test send request with cookies.
        """
        result = self.do('http://httpbin.org/cookies', options={
            'cookies': {
                'abc': 123,
                'def': '+-*/',
            }
        })

        self.assertTrue('"abc": "123"' in result.content)
        self.assertTrue('"def": "+-*/"' in result.content)

    def _testScript(self):
        """
        Test send request with javascript.
        """
        result = self.do('http://naga.servehttp.com/', options={
            'script': "document.write('testScript');",
        })

        self.assertTrue('testScript' in result.content)

if __name__ == '__main__':
    unittest.main()
    