#!/usr/bin/env python
"""
open/DurusWorks/qp/http/test/utest_response.py

"""
from durus.utils import BytesIO, as_bytes, join_bytes
from qp.fill.static import FileStream
from qp.http.response import HTTPResponse, Stream
from qp.lib.spec import unicode_string
from sancho.utest import raises
from sancho.utest import UTest
from tempfile import NamedTemporaryFile
import os
import string
import sys

ascii_lowercase = as_bytes('abcdefghijklmnopqrstuvwxyz')

class Test(UTest):

    def a(self):
        response = HTTPResponse()
        response.set_content_type('text/html', 'utf-8')
        response.set_status(200)
        response.set_status(200, 'Excellent')
        assert response.get_status_code() == 200
        assert response.get_status() == (200, 'Excellent')
        response.set_status(479)
        try:
            response.set_status(600)
            assert 0
        except ValueError:
            pass
        try:
            response.set_status('200')
            assert 0
        except TypeError:
            pass
        response.set_expires(None)
        response.set_expires(seconds=200)
        response.set_header('foo', 1)
        assert response.get_header('foo') == 1
        response.set_cookie('a', 23)
        response.expire_cookie('a')
        response.set_body('ok')
        response.set_compress(False)
        response.set_body('ok' * 1000)

    def b(self):
        response = HTTPResponse()
        assert response.get_mime_type() == 'text/html'
        response.set_content_type('text/plain', 'utf-8')
        assert response.get_mime_type() == 'text/plain', response.get_mime_type()    
        assert (200, 'OK') == response.status
        assert response.get_content_length() == None
        response.set_body('ab')
        assert response.get_content_length() == 2
        response.set_body(Stream(BytesIO(as_bytes('a')), 1))
        assert response.get_content_length() == 1
        response.set_cookie('cookie', 1)
        headers = response.generate_headers()
        headers_dict = dict(headers)
        del headers_dict[as_bytes('Date')]
        assert len(headers_dict) == 5
        print(headers_dict)
        for k, v in [('Content-Type', 'text/plain; charset=utf-8'),
                     ('Expires', '-1'),
                     ('Content-Length', '1'),
                     ('Set-Cookie', 'cookie="1"')]:
            assert headers_dict[as_bytes(k)] == as_bytes(v), (k, v)
        response.set_expires(None)
        headers = response.generate_headers()
        response.set_expires(days=1)
        headers = response.generate_headers()
        response.set_cookie('a', 'x"', secure=1, expires=1,
                            b=None, a='all"ok"')
        response.set_header('bogus', 'yes')
        headers = response.generate_headers()

    def d(self):
        response = HTTPResponse()
        response.set_cookie('a', 'x')
        response.set_body('ok')        
        headers = response.generate_headers()
        headers_dict = dict(headers)
        del headers_dict[as_bytes('Date')]
        assert len(headers_dict) == 5
        print(headers_dict)
        for k, v in [('Content-Type', 'text/html; charset=utf-8'),
                     ('Expires', '-1'),
                     ('Content-Length', '2'),
                     ('Set-Cookie', 'a="x"')]:
            assert headers_dict[as_bytes(k)] == as_bytes(v), (k, v)

    def check_write(self):
        stream = Stream(BytesIO(as_bytes('a')), 1)
        assert join_bytes(stream) == as_bytes('a')
        response = HTTPResponse()
        response.buffered=False
        out = BytesIO()
        response.write(out, include_status=False)
        response.write(out, include_status=True)
        response.write(out, include_body=False)
        response.set_body('a')
        response.write(out)
        response.set_body(stream)
        response.write(out)
        response.set_body(Stream(BytesIO(as_bytes('a')*1000), 1000))
        response.write(out)
        response.set_body(unicode_string('0') * 1000)
        response.write(out)
        response.set_range((0, 3))
        out = BytesIO()
        response.write(out)
        assert response.get_content_length() == 4
        response.set_range((None, 2))
        out = BytesIO()
        response.write(out)
        assert response.get_content_length() == 2, out.getvalue()        
        response.set_range((1, None))
        out = BytesIO()
        response.write(out)
        assert response.get_content_length() == 999                
        

    def c(self):
        response = HTTPResponse()
        response.set_cookie('a', '"ok', path="/", secure=True)
        response.set_cookie('b', None)
        response.set_header('ok', 'this')
        list(response.generate_headers())
        assert response.cache == 0
        response.set_expires(1)
        list(response.generate_headers())
        response.cache = None
        list(response.generate_headers())

    def _eat_stream(self, stream):
        chunks = []
        for chunk in stream:
            chunks.append(as_bytes(chunk))
        return as_bytes("").join(chunks)

    def stream1(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable)
        result = self._eat_stream(stream)
        assert result == ascii_lowercase, (result, ascii_lowercase)

    def stream2(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable)
        stream.set_range((0, None))
        raises(ValueError, self._eat_stream, stream)

    def stream3(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        stream.set_range((0, None))
        result = self._eat_stream(stream)
        assert result == ascii_lowercase

    def stream4(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        stream.set_range((5, None))
        result = self._eat_stream(stream)
        print("\n\n", result , "\n\n")
        assert result == ascii_lowercase[5:]

    def stream5(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable)
        stream.set_range((None, len(ascii_lowercase)))
        raises(ValueError, self._eat_stream, stream)

    def stream6(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        stream.set_range((None, len(ascii_lowercase)))
        result = self._eat_stream(stream)
        assert result == ascii_lowercase

    def stream7(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        stream.set_range((None, 5))
        result = self._eat_stream(stream)
        print("\n\n", result , "\n\n")
        assert result == ascii_lowercase[-5:]

    def stream7(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        stream.set_range((4, 5))
        result = self._eat_stream(stream)
        print("\n\n", result , "\n\n")
        assert result == ascii_lowercase[4:6]

    def stream7(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        stream.set_range((4, 20))
        result = self._eat_stream(stream)
        print("\n\n", result , "\n\n")
        assert result == ascii_lowercase[4:21]

    def stream8(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        response = HTTPResponse()
        response.set_range((4, 15))
        response.set_body(stream)
        headers = response.generate_headers()
        headers_dict = dict(headers)
        del headers_dict[as_bytes('Date')]
        print(headers_dict)
        for k, v in [
            ('Content-Type', 'text/html; charset=utf-8'),
            ('Expires', '-1'),
            ('Content-Length', '12'),
            ('Content-Range', 'bytes 4-15/26')]:
            assert headers_dict[as_bytes(k)] == as_bytes(v), (k, v)
        assert response.get_status()[0] == 206


    def stream9(self):
        iterable = [ "abcde", "fg", "hijklmn", "o", "pqrst", "uvwxyz" ]
        stream = Stream(iterable, len(ascii_lowercase))
        response = HTTPResponse()
        response.set_range((4, None))
        response.set_body(stream)
        headers = response.generate_headers()
        headers_dict = dict(headers)
        del headers_dict[as_bytes('Date')]
        print(headers_dict)        
        for k, v in [
            ('Content-Type', 'text/html; charset=utf-8'),
            ('Expires', '-1'),
            ('Content-Length', '22'),
            ('Content-Range', 'bytes 4-25/26')]:
            assert headers_dict[as_bytes(k)] == as_bytes(v), (k, v)
        assert response.get_status()[0] == 206

    def stream10(self):
        temp_file = NamedTemporaryFile()
        temp_file.write(ascii_lowercase)
        temp_file.flush()
        temp_file.seek(0)
        stat = os.stat(temp_file.name)
        stream = FileStream(temp_file, stat.st_size)
        response = HTTPResponse()
        response.set_range((4, None))
        response.set_body(stream)
        headers = response.generate_headers()
        headers_dict = dict(headers)
        del headers_dict[as_bytes('Date')]
        print(headers_dict)
        for k, v in [
            ('Content-Type', 'text/html; charset=utf-8'),
            ('Expires', '-1'),
            ('Content-Length', '22'),
            ('Content-Range', 'bytes 4-25/26')]:
            assert headers_dict[as_bytes(k)] == as_bytes(v), (k, v)
        assert response.get_status()[0] == 206

    def stream11(self):
        temp_file = NamedTemporaryFile()
        temp_file.write(ascii_lowercase)
        temp_file.flush()
        temp_file.seek(0)
        stat = os.stat(temp_file.name)
        stream = FileStream(temp_file, stat.st_size)
        response = HTTPResponse()
        response.set_range((4, 15))
        response.set_body(stream)
        headers = response.generate_headers()
        headers_dict = dict(headers)
        del headers_dict[as_bytes('Date')]
        print(headers_dict)
        for k, v in [
            ('Content-Type', 'text/html; charset=utf-8'),
            ('Expires', '-1'),
            ('Content-Length', '12'),
            ('Content-Range', 'bytes 4-15/26')]:
            assert headers_dict[as_bytes(k)] == as_bytes(v), (k, v)
        assert response.get_status()[0] == 206

    def stream12(self):
        temp_file = NamedTemporaryFile()
        temp_file.write(ascii_lowercase)
        temp_file.flush()
        temp_file.seek(0)
        stat = os.stat(temp_file.name)
        stream = FileStream(temp_file, stat.st_size)
        response = HTTPResponse()
        response.set_range((4, 30))
        response.set_body(stream)
        headers = response.generate_headers()
        headers_dict = dict(headers)
        del headers_dict[as_bytes('Date')]
        print(headers_dict)
        for k, v in [
            ('Content-Type', 'text/html; charset=utf-8'),
            ('Expires', '-1'),
            ('Content-Length', '22'),
            ('Content-Range', 'bytes 4-25/26')]:
            assert headers_dict[as_bytes(k)] == as_bytes(v), (k, v)
        assert response.get_status()[0] == 206

if __name__ == '__main__':
    Test()