"""
$URL: svn+ssh://svn.mems-exchange.org/repos/trunk/qp/pub/test/utest_publish.py $
$Id: utest_publish.py 29552 2007-03-08 15:14:27Z dbinger $
"""
from cStringIO import StringIO
from qp.lib.spec import get_spec_problems
from qp.pub.common import clear_publisher, site_now
from qp.pub.hit import Hit
from qp.pub.publish import Publisher, DurusPublisher
from qp.pub.user import User, compute_digest
from sancho.utest import UTest

class Test(UTest):

    def _pre(self):
        clear_publisher()

    def _post(self):
        clear_publisher()

    def check_base(self):
        class Pub(Publisher):
            pass
        publisher = Pub()
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w'))
        publisher.process_hit(hit)

    def check_wsgi(self):
        env = dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w')
        env['wsgi.input'] = StringIO()
        class Pub(Publisher):
            def fill_response(self):
                self.hit.response.set_body('ok')
        publisher = Pub()
        status_headers = []
        def start_response(status, headers):
            status_headers.append(status)
            status_headers.append(headers)
        result = publisher(env, start_response)
        print status_headers[0]
        for key, value in status_headers[1]:
            print key, value
        for line in result:
            print line

    def check_not_found(self):
        class Pub(Publisher):
            def fill_response(self):
                self.not_found()
        publisher = Pub()
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w'))
        publisher.process_hit(hit)
        print hit.get_response().get_status_code()
        assert hit.get_response().get_status_code() == 404

    def check_permanent_redirect(self):
        class Pub(Publisher):
            def fill_response(self):
                self.redirect('', permanent=True)
        publisher = Pub()
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w'))
        publisher.process_hit(hit)
        print hit.get_response().get_status_code()
        assert hit.get_response().get_status_code() == 301

    def check_temporary_redirect(self):
        class Pub(Publisher):
            def fill_response(self):
                self.redirect('')
        publisher = Pub()
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w'))
        publisher.process_hit(hit)
        print hit.get_response().get_status_code()
        assert hit.get_response().get_status_code() == 302

    def check_retry(self):
        class Pub(Publisher):
            def fill_response(self):
                assert 0
            def is_terminal_exception(self):
                return False
        publisher = Pub()
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w'))
        publisher.process_hit(hit)

    def check_display_exceptions(self):
        class Pub(Publisher):
            def fill_response(self):
                assert 0
            def display_exceptions(self):
                return True
        publisher = Pub()
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
                             QUERY_STRING='how'))
        publisher.process_hit(hit)

    def check_hide_exceptions(self):
        class Pub(Publisher):
            def fill_response(self):
                assert 0
        publisher = Pub()
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
                             QUERY_STRING='how'))
        publisher.process_hit(hit)

    def check_ensure_signed_in_using_digest(self):
        publisher = DurusPublisher()
        null_user = publisher.get_users()[""]
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w'))
        def fill():
            publisher.ensure_signed_in_using_digest()
        def nothing():
            pass
        publisher.fill_response_using_root_directory = fill
        publisher.abort = nothing
        publisher.commit = nothing
        publisher.process_hit(hit)
        assert hit.get_response().get_header('Location').startswith('https')
        assert hit.get_response().get_status_code() == 302
        hit = Hit(None, dict(SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
                             HTTPS="On"))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401
        challenge = hit.get_response().get_header('WWW-Authenticate')
        realm = publisher.get_site().get_name()
        nonce = null_user.get_tokens().tokens[0]
        expected = ('Digest realm="%s", nonce="%s", opaque="0%s", stale=false, '
                    'algorithm=MD5, qop="auth"' % (realm, nonce, nonce))
        assert challenge == expected
        auth = "not_digest a b c d e"
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401
        auth = "Digest a b c d e"  # no username
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401
        auth = 'Digest username="db" b c d e' # no-nonce
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401

        db = User('db')
        publisher.get_users()['db'] = db
        auth = 'Digest username="db", nonce="%s", c, d, e,' % 'bogus'
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401

        nonce = null_user.get_tokens().new_token()
        auth = 'Digest username="db", nonce="%s", c, d, e,' % nonce
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401

        nonce = null_user.get_tokens().new_token()
        auth = 'Digest username="db", nonce="%s", realm="%s", d, e,' % (
            nonce, realm)
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401

        nonce = null_user.get_tokens().new_token()
        auth = ('Digest username="db", ' +
                ('response="%s", ' % ('4' * 32)) +
                ('nonce="%s", realm="%s", , e,' % (
            nonce, realm)))
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401

        db.set_password('s')
        nonce = null_user.get_tokens().new_token()
        auth = ('Digest username="db", ' +
                ('response="%s", ' % ('4' * 32)) +
                ('nonce="%s", realm="%s", , e,' % (
            nonce, realm)))
        hit = Hit(None, dict(
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 401

        user_digest = db.get_digester().get_digest(realm)
        nonce = null_user.get_tokens().new_token()
        response = compute_digest(
            user_digest, nonce, '01', 'bar', 'auth',
            compute_digest('GET', '/a'))
        auth = ('Digest username="db", uri="/a", ' +
                ('response="%s", ' % response) +
                ('cnonce="bar", nc=01, qop=auth, ') +
                ('nonce="%s", realm="%s" ' % (
            nonce, realm)))
        hit = Hit(None, dict(
            REQUEST_METHOD="GET",
            SCRIPT_NAME='', PATH_INFO='/a', SERVER_NAME='w',
            HTTPS="On", HTTP_AUTHORIZATION=auth))
        publisher.process_hit(hit)
        assert publisher.parse_log_hit_line(publisher.format_log_hit_line())
        assert hit.get_response().get_status_code() != 401
        line = (
        "2006-05-01 19:40:55 200 .004234 10.27.8.147 u'd' + 18809 https "
        "GET /test.css https://example.org/ok Mozilla/5.0_Macintosh")
        parsed = publisher.parse_log_hit_line(line)
        assert parsed, line
        assert parsed['method'] == 'GET'
        assert parsed['user'] == '+'
        assert parsed['duration'] == 0.004234

    def timezone(self):
        publisher = Publisher()
        time_zone = publisher.get_time_zone()
        assert site_now().tzinfo is time_zone

    def short(self):
        publisher = Publisher()
        hit = Hit(
            StringIO(),
            dict(REQUEST_METHOD="POST",
                CONTENT_TYPE="application/x-www-form-urlencoded",
                CONTENT_LENGTH=5))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 400

    def short_durus_publisher(self):
        publisher = DurusPublisher()
        hit = Hit(
            StringIO(),
            dict(REQUEST_METHOD="POST",
                CONTENT_TYPE="application/x-www-form-urlencoded",
                CONTENT_LENGTH=5))
        publisher.process_hit(hit)
        assert hit.get_response().get_status_code() == 400

    def test_hit(self):
        publisher = DurusPublisher()
        hit = Hit(
            StringIO(),
            dict(REQUEST_METHOD="POST",
                CONTENT_TYPE="application/x-www-form-urlencoded",
                CONTENT_LENGTH=5))
        publisher.set_hit(hit)
        assert get_spec_problems(hit) == []
        assert hit.get_session() is not None
        assert get_spec_problems(hit) == []

        publisher.process_hit(hit)
if __name__ == '__main__':
    Test()