#
 #
 # Licensed under the Apache License, Version 2.0 (the "License"); you
 # may not use this file except in compliance with the License.  You
 # may obtain a copy of the License at
 #
 #      http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 # implied.  See the License for the specific language governing
 # permissions and limitations under the License.
 #
 #
 #

# mod_python tests

from mod_python.python22 import *

from mod_python import apache
import sys
import unittest
import re
import time
import os
import io

PY2 = sys.version[0] == '2'

# This is used for mod_python.publisher security tests
_SECRET_PASSWORD = 'root'
__ANSWER = 42

class SimpleTestCase(unittest.TestCase):

    def __init__(self, methodName, req):
        unittest.TestCase.__init__(self, methodName)
        self.req = req

    def test_apache_log_error(self):

        s = self.req.server
        c = self.req.connection

        apache.log_error("Testing apache.log_error():", apache.APLOG_INFO, s)

        apache.log_error("xEMERGx", apache.APLOG_EMERG, s)
        apache.log_error("xALERTx", apache.APLOG_ALERT, s)
        apache.log_error("xCRITx", apache.APLOG_CRIT, s)
        apache.log_error("xERRx", apache.APLOG_ERR, s)
        apache.log_error("xWARNINGx", apache.APLOG_WARNING, s)
        apache.log_error("xNOTICEx", apache.APLOG_NOTICE, s)
        apache.log_error("xINFOx", apache.APLOG_INFO, s)
        apache.log_error("xDEBUGx", apache.APLOG_DEBUG, s)

        s.log_error("xEMERGx", apache.APLOG_EMERG)
        s.log_error("xALERTx", apache.APLOG_ALERT)
        s.log_error("xCRITx", apache.APLOG_CRIT)
        s.log_error("xERRx", apache.APLOG_ERR)
        s.log_error("xWARNINGx", apache.APLOG_WARNING)
        s.log_error("xNOTICEx", apache.APLOG_NOTICE)
        s.log_error("xINFOx", apache.APLOG_INFO)
        s.log_error("xDEBUGx", apache.APLOG_DEBUG)

        c.log_error("xEMERGx", apache.APLOG_EMERG)
        c.log_error("xALERTx", apache.APLOG_ALERT)
        c.log_error("xCRITx", apache.APLOG_CRIT)
        c.log_error("xERRx", apache.APLOG_ERR)
        c.log_error("xWARNINGx", apache.APLOG_WARNING)
        c.log_error("xNOTICEx", apache.APLOG_NOTICE)
        c.log_error("xINFOx", apache.APLOG_INFO)
        c.log_error("xDEBUGx", apache.APLOG_DEBUG)

        # see what's in the log now
        f = open("%s/logs/error_log" % apache.server_root())
        # for some reason re doesn't like \n, why?
        log = "".join(map(str.strip, f.readlines()))
        f.close()

        if not re.search("xEMERGx.*xALERTx.*xCRITx.*xERRx.*xWARNINGx.*xNOTICEx.*xINFOx.*xDEBUGx.*xEMERGx.*xALERTx.*xCRITx.*xERRx.*xWARNINGx.*xNOTICEx.*xINFOx.*xDEBUGx.*xEMERGx.*xALERTx.*xCRITx.*xERRx.*xWARNINGx.*xNOTICEx.*xINFOx.*xDEBUGx", log):
            self.fail("Could not find test messages in error_log")


    def test_apache_table(self):

        log = self.req.log_error

        log("Testing table object.")

        # tests borrowed from Python test suite for dict
        _test_table()

        # inheritance
        log("  inheritance")
        class mytable(apache.table):
            def __str__(self):
                return "str() from mytable"
        mt = mytable({'a':'b'})

        # add()
        log("  table.add()")
        a = apache.table({'a':'b'})
        a.add('a', 'c')
        if a['a'] != ['b', 'c']:
            self.fail('table.add() broken: a["a"] is %s' % repr(a["a"]))

        log("Table test DONE.")

    def test_req_add_common_vars(self):

        self.req.log_error("Testing req.add_common_vars().")

        a = len(self.req.subprocess_env)
        self.req.add_common_vars()
        b = len(self.req.subprocess_env)
        if a >= b:
            self.fail("req.subprocess_env() is same size before and after")

    def test_req_add_cgi_vars(self):

        self.req.log_error("Testing req.add_cgi_vars().")

        a = len(self.req.subprocess_env)
        self.req.add_cgi_vars()
        b = len(self.req.subprocess_env)
        if a >= b:
            self.fail("req.subprocess_env() is same size before and after")

    def test_req_members(self):

        # just run through request members making sure
        # they make sense

        req = self.req
        log = req.log_error

        log("Examining request memebers:")

        log("    req.connection: %s" % repr(req.connection))
        s = str(type(req.connection))
        if s not in ("<class 'mp_conn'>", "<type 'mp_conn'>"):
            self.fail("strange req.connection type %s" % repr(s))

        log("    req.server: '%s'" % repr(req.server))
        s = str(type(req.server))
        if s not in ("<class 'mp_server'>", "<type 'mp_server'>"):
            self.fail("strange req.server type %s" % repr(s))

        for x in ((req.next, "next"),
                  (req.prev, "prev"),
                  (req.main, "main")):
            val, name = x
            log("    req.%s: '%s'" % (name, repr(val)))
            if val:
                self.fail("strange, req.%s should be None, not %s" % (name, repr(val)))

        log("    req.the_request: '%s'" % req.the_request)
        if not re.match(r"GET /.* HTTP/1\.", req.the_request):
            self.fail("strange req.the_request %s" % repr(req.the_request))

        for x in ((req.assbackwards, "assbackwards"),
                  (req.proxyreq, "proxyreq"),
                  (req.header_only, "header_only")):
            val, name = x
            log("    req.%s: %s" % (name, repr(val)))
            if val:
                self.fail("%s should be 0" % name)

        log("    req.protocol: %s" % repr(req.protocol))
        if not req.protocol == req.the_request.split()[-1]:
            self.fail("req.protocol doesn't match req.the_request")

        log("    req.proto_num: %s" % repr(req.proto_num))
        if req.proto_num != 1000 + int(req.protocol[-1]):
            self.fail("req.proto_num doesn't match req.protocol")

        log("    req.hostname: %s" % repr(req.hostname))
        if req.hostname != "test_internal":
            self.fail("req.hostname isn't 'test_internal'")

        log("    req.request_time: %s" % repr(req.request_time))
        if (time.time() - req.request_time) > 10:
            self.fail("req.request_time suggests request started more than 10 secs ago")

        log("    req.status_line: %s" % repr(req.status_line))
        if req.status_line:
            self.fail("req.status_line should be None at this point")

        log("    req.status: %s" % repr(req.status))
        if req.status != 200:
            self.fail("req.status should be 200")
        req.status = req.status # make sure its writable

        log("    req.method: %s" % repr(req.method))
        if req.method != "GET":
            self.fail("req.method should be 'GET'")

        log("    req.method_number: %s" % repr(req.method_number))
        if req.method_number != 0:
            self.fail("req.method_number should be 0")

        log("    req.allowed: %s" % repr(req.allowed))
        if req.allowed != 0:
            self.fail("req.allowed should be 0")

        log("    req.allowed_xmethods: %s" % repr(req.allowed_xmethods))
        if req.allowed_xmethods != ():
            self.fail("req.allowed_xmethods should be an empty tuple")

        log("    req.allowed_methods: %s" % repr(req.allowed_methods))
        if req.allowed_methods != ():
            self.fail("req.allowed_methods should be an empty tuple")

        log("    req.sent_bodyct: %s" % repr(req.sent_bodyct))
        if req.sent_bodyct != 0:
            self.fail("req.sent_bodyct should be 0")

        log("    req.bytes_sent: %s" % repr(req.bytes_sent))
        save = req.bytes_sent
        log("       writing 4 bytes...")
        req.write("1234")
        log("       req.bytes_sent: %s" % repr(req.bytes_sent))
        if req.bytes_sent - save != 4:
            self.fail("req.bytes_sent should have incremented by 4, but didn't")

        log("    req.mtime: %s" % repr(req.mtime))
        if req.mtime != 0:
            self.fail("req.mtime should be 0")

        log("    req.chunked: %s" % repr(req.chunked))
        if req.chunked != 1:
            self.fail("req.chunked should be 1")

        log("    req.range: %s" % repr(req.range))
        if req.range:
            self.fail("req.range should be None")

        log("    req.clength: %s" % repr(req.clength))
        log("        calling req.set_content_length(15)...")
        req.set_content_length(15)
        log("        req.clength: %s" % repr(req.clength))
        if req.clength != 15:
            self.fail("req.clength should be 15")

        log("    req.remaining: %s" % repr(req.remaining))
        if req.remaining != 0:
            self.fail("req.remaining should be 0")

        log("    req.read_length: %s" % repr(req.read_length))
        if req.read_length != 0:
            self.fail("req.read_length should be 0")

        log("    req.read_body: %s" % repr(req.read_body))
        if req.read_body != 0:
            self.fail("req.read_body should be 0")

        log("    req.read_chunked: %s" % repr(req.read_chunked))
        if req.read_chunked != 0:
            self.fail("req.read_chunked should be 0")

        log("    req.expecting_100: %s" % repr(req.expecting_100))
        if req.expecting_100 != 0:
            self.fail("req.expecting_100 should be 0")

        log("    req.headers_in: %s" % repr(req.headers_in))
        if req.headers_in["Host"][:13].lower() != "test_internal":
            self.fail("The 'Host' header should begin with 'test_internal'")

        log("    req.headers_out: %s" % repr(req.headers_out))
        if (("content-length" not in req.headers_out) or
            req.headers_out["content-length"] != "15"):
            self.fail("req.headers_out['content-length'] should be 15")

        log("    req.subprocess_env: %s" % repr(req.subprocess_env))
        if req.subprocess_env["SERVER_SOFTWARE"].find("Python") == -1:
            self.fail("req.subprocess_env['SERVER_SOFTWARE'] should contain 'Python'")

        log("    req.notes: %s" % repr(req.notes))
        log("        doing req.notes['testing'] = '123' ...")
        req.notes['testing'] = '123'
        log("    req.notes: %s" % repr(req.notes))
        if req.notes["testing"] != '123':
            self.fail("req.notes['testing'] should be '123'")

        log("    req.phase: %s" % repr(req.phase))
        if req.phase != "PythonHandler":
            self.fail("req.phase should be 'PythonHandler'")

        log("    req.interpreter: %s" % repr(req.interpreter))
        if req.interpreter != apache.interpreter:
            self.fail("req.interpreter should be same as apache.interpreter" % repr(apache.interpreter))
        if req.interpreter != req.server.server_hostname:
            self.fail("req.interpreter should be same as req.server.server_hostname: %s" % repr(req.server.server_hostname))

        log("    req.content_type: %s" % repr(req.content_type))
        log("        doing req.content_type = 'test/123' ...")
        req.content_type = 'test/123'
        log("        req.content_type: %s" % repr(req.content_type))
        if req.content_type != 'test/123' or not req._content_type_set:
            self.fail("req.content_type should be 'test/123' and req._content_type_set 1")

        log("    req.handler: %s" % repr(req.handler))
        if req.handler != "mod_python":
            self.fail("req.handler should be 'mod_python'")

        log("    req.content_encoding: %s" % repr(req.content_encoding))
        if req.content_encoding:
            self.fail("req.content_encoding should be None")

        log("    req.content_languages: %s" % repr(req.content_languages))
        if req.content_languages != ():
            self.fail("req.content_languages should be an empty tuple")

        log("    req.vlist_validator: %s" % repr(req.vlist_validator))
        if req.vlist_validator:
            self.fail("req.vlist_validator should be None")

        log("    req.user: %s" % repr(req.user))
        if req.user:
            self.fail("req.user should be None")

        log("    req.ap_auth_type: %s" % repr(req.ap_auth_type))
        if req.ap_auth_type:
            self.fail("req.ap_auth_type should be None")

        log("    req.no_cache: %s" % repr(req.no_cache))
        if req.no_cache != 0:
            self.fail("req.no_cache should be 0")

        log("    req.no_local_copy: %s" % repr(req.no_local_copy))
        if req.no_local_copy != 0:
            self.fail("req.no_local_copy should be 0")

        log("    req.unparsed_uri: %s" % repr(req.unparsed_uri))
        if req.unparsed_uri != "/tests.py":
            self.fail("req.unparsed_uri should be '/tests.py'")

        log("    req.uri: %s" % repr(req.uri))
        if req.uri != "/tests.py":
            self.fail("req.uri should be '/tests.py'")

        log("    req.filename: %s" % repr(req.filename))
        if req.filename != req.document_root() + req.uri:
            self.fail("req.filename should be req.document_root() + req.uri, but it isn't")

        log("    req.canonical_filename: %s" % repr(req.canonical_filename))
        if not req.canonical_filename:
            self.fail("req.canonical_filename should not be blank")

        log("    req.path_info: %s" % repr(req.path_info))
        if req.path_info != '':
            self.fail("req.path_info should be ''")

        log("    req.args: %s" % repr(req.args))
        if req.args:
            self.fail("req.args should be None")

        log("    req.finfo: %s" % repr(req.finfo))
        if req.finfo[apache.FINFO_FNAME] and (req.finfo[apache.FINFO_FNAME] != req.canonical_filename):
            self.fail("req.finfo[apache.FINFO_FNAME] should be the (canonical) filename")

        log("    req.parsed_uri: %s" % repr(req.parsed_uri))
        if req.parsed_uri[apache.URI_PATH] != '/tests.py':
            self.fail("req.parsed_uri[apache.URI_PATH] should be '/tests.py'")

        log("    req.used_path_info: %s" % repr(req.used_path_info))
        if req.used_path_info != 2:
            self.fail("req.used_path_info should be 2") # XXX really? :-)

        log("    req.eos_sent: %s" % repr(req.eos_sent))
        if req.eos_sent:
            self.fail("req.eos_sent says we sent EOS, but we didn't")

        if apache.MODULE_MAGIC_NUMBER_MAJOR > 20111130:

            try:
                import socket
                localip = socket.gethostbyname("localhost")
            except:
                localip = "127.0.0.1"

            log("    req.useragent_ip: %s" % repr(req.useragent_ip))
            if not req.useragent_ip in ("127.0.0.1", localip):
                self.fail("req.useragent_ip should be '127.0.0.1'")

            log("    req.useragent_addr: %s" % repr(req.useragent_addr))
            if not req.useragent_addr[0] in ("127.0.0.1", "0.0.0.0", localip):
                self.fail("req.useragent_addr[0] should be '127.0.0.1' or '0.0.0.0'")


    def test_req_get_config(self):

        req = self.req
        log = req.log_error

        log("req.get_config(): %s" % repr(req.get_config()))
        if req.get_config()["PythonDebug"] != "1":
            self.fail("get_config return should show PythonDebug 1")

        log("req.get_options(): %s" % repr(req.get_options()))
        if req.get_options() != apache.table({"testing":"123"}):
            self.fail("get_options() should contain 'testing':'123', contains %s"%list(req.get_options().items()))

    def test_req_get_remote_host(self):

        # simulating this test for real is too complex...
        req = self.req
        log = req.log_error
        log("req.get_get_remote_host(): %s" % repr(req.get_remote_host(apache.REMOTE_HOST)))
        log("req.get_get_remote_host(): %s" % repr(req.get_remote_host()))
        if (req.get_remote_host(apache.REMOTE_HOST) != None) or \
           (req.get_remote_host() != "127.0.0.1"):
            self.fail("remote host test failed: %s" % req.get_remote_host())

    def test_server_members(self):

        req = self.req
        log = req.log_error
        server = req.server

        log("Examining server memebers:")

        log("    server.defn_name: %s" % repr(server.defn_name))
        if server.defn_name[-9:] != "test.conf":
            self.fail("server.defn_name does not end in 'test.conf'")

        log("    server.defn_line_number: %s" % repr(server.defn_line_number))
        if server.defn_line_number == 0:
            self.fail("server.defn_line_number should not be 0")

        log("    server.server_admin: %s" % repr(server.server_admin))
        if server.server_admin != "serveradmin@somewhere.com":
            self.fail("server.server_admin must be 'serveradmin@somewhere.com'")

        log("    server.server_hostname: %s" % repr(server.server_hostname))
        if server.server_hostname != "test_internal":
            self.fail("server.server_hostname must be 'test_internal'")

        log("    server.port: %s" % repr(server.port))
        # hmm it really is 0...
        #if server.port == 0:
        #    self.fail("server.port should not be 0")

        log("    server.error_fname: %s" % repr(server.error_fname))
        if server.error_fname != "logs/error_log":
            self.fail("server.error_fname should be 'logs/error_log'")

        log("    server.loglevel: %s" % repr(server.loglevel))
        if server.loglevel != 7:
            self.fail("server.loglevel should be 7")

        log("    server.is_virtual: %s" % repr(server.is_virtual))
        if server.is_virtual != 1:
            self.fail("server.is_virtual should be 1")

        log("    server.timeout: %s" % repr(server.timeout))
        if not server.timeout in (5.0, 60.0):
            self.fail("server.timeout should be 5.0 or 60.0")

        log("    server.keep_alive_timeout: %s" % repr(server.keep_alive_timeout))
        if server.keep_alive_timeout != 15.0:
            self.fail("server.keep_alive_timeout should be 15.0")

        log("    server.keep_alive_max: %s" % repr(server.keep_alive_max))
        if server.keep_alive_max != 100:
            self.fail("server.keep_alive_max should be 100")

        log("    server.keep_alive: %s" % repr(server.keep_alive))
        if server.keep_alive != 1:
            self.fail("server.keep_alive should be 1")

        log("    server.path: %s" % repr(server.path))
        if server.path != "some/path":
            self.fail("server.path should be 'some/path'")

        log("    server.pathlen: %s" % repr(server.pathlen))
        if server.pathlen != len('some/path'):
            self.fail("server.pathlen should be %d" % len('some/path'))

        log("    server.limit_req_line: %s" % repr(server.limit_req_line))
        if server.limit_req_line != 8190:
            self.fail("server.limit_req_line should be 8190")

        log("    server.limit_req_fieldsize: %s" % repr(server.limit_req_fieldsize))
        if server.limit_req_fieldsize != 8190:
            self.fail("server.limit_req_fieldsize should be 8190")

        log("    server.limit_req_fields: %s" % repr(server.limit_req_fields))
        if server.limit_req_fields != 100:
            self.fail("server.limit_req_fields should be 100")

        log("    server.names: %s" % repr(server.names))
        if server.names != ():
            self.fail("server.names should be an empty tuple")

        log("    server.wild_names: %s" % repr(server.wild_names))
        if server.wild_names != ():
            self.fail("server.wild_names should be an empty tuple")


    def test_connection_members(self):

        req = self.req
        log = req.log_error
        conn = req.connection

        try:
            import socket
            localip = socket.gethostbyname("localhost")
        except:
            localip = "127.0.0.1"

        log("Examining connection memebers:")

        log("    connection.base_server: %s" % repr(conn.base_server))
        if type(conn.base_server) is not type(req.server):
            self.fail("conn.base_server should be same type as req.server")

        log("    connection.local_addr: %s" % repr(conn.local_addr))
        if not conn.local_addr[0] in ("127.0.0.1", "0.0.0.0", localip):
            self.fail("conn.local_addr[0] should be '127.0.0.1' or '0.0.0.0'")

        if apache.MODULE_MAGIC_NUMBER_MAJOR > 20111130:

            log("    connection.client_addr: %s" % repr(conn.client_addr))
            if not conn.client_addr[0] in ("127.0.0.1", "0.0.0.0", localip):
                self.fail("conn.client_addr[0] should be '127.0.0.1' or '0.0.0.0'")

            log("    connection.client_ip: %s" % repr(conn.client_ip))
            if not conn.client_ip in ("127.0.0.1", localip):
                self.fail("conn.client_ip should be '127.0.0.1'")

        else:

            log("    connection.remote_addr: %s" % repr(conn.remote_addr))
            if not conn.remote_addr[0] in ("127.0.0.1", "0.0.0.0", localip):
                self.fail("conn.remote_addr[0] should be '127.0.0.1' or '0.0.0.0'")

            log("    connection.remote_ip: %s" % repr(conn.remote_ip))
            if not conn.remote_ip in ("127.0.0.1", localip):
                self.fail("conn.remote_ip should be '127.0.0.1'")

        log("    connection.remote_host: %s" % repr(conn.remote_host))
        if conn.remote_host is not None:
            self.fail("conn.remote_host should be None")

        log("    connection.remote_logname: %s" % repr(conn.remote_logname))
        if conn.remote_logname is not None:
            self.fail("conn.remote_logname should be None")

        log("    connection.aborted: %s" % repr(conn.aborted))
        if conn.aborted != 0:
            self.fail("conn.aborted should be 0")

        log("    connection.keepalive: %s" % repr(conn.keepalive))
        if conn.keepalive != 2:
            self.fail("conn.keepalive should be 2")

        log("    connection.double_reverse: %s" % repr(conn.double_reverse))
        if conn.double_reverse != 0:
            self.fail("conn.double_reverse should be 0")

        log("    connection.keepalives: %s" % repr(conn.keepalives))
        if conn.keepalives != 1:
            self.fail("conn.keepalives should be 1")

        log("    connection.local_ip: %s" % repr(conn.local_ip))
        if not conn.local_ip in ("127.0.0.1", localip):
            self.fail("conn.local_ip should be '127.0.0.1'")

        log("    connection.local_host: %s" % repr(conn.local_host))
        if conn.local_host is not None:
            self.fail("conn.local_host should be None")

        log("    connection.id: %s" % repr(conn.id))
        if conn.id > 10000:
            self.fail("conn.id probably should not be this high?")

        log("    connection.notes: %s" % repr(conn.notes))
        if repr(conn.notes) != '{}':
            self.fail("conn.notes should be {}")

def make_suite(req):

    mpTestSuite = unittest.TestSuite()
    mpTestSuite.addTest(SimpleTestCase("test_apache_log_error", req))
    mpTestSuite.addTest(SimpleTestCase("test_apache_table", req))
    # NB: add_common_vars must be before cgi_vars
    mpTestSuite.addTest(SimpleTestCase("test_req_add_common_vars", req))
    mpTestSuite.addTest(SimpleTestCase("test_req_add_cgi_vars", req))
    mpTestSuite.addTest(SimpleTestCase("test_req_members", req))
    mpTestSuite.addTest(SimpleTestCase("test_req_get_config", req))
    mpTestSuite.addTest(SimpleTestCase("test_req_get_remote_host", req))
    mpTestSuite.addTest(SimpleTestCase("test_server_members", req))
    mpTestSuite.addTest(SimpleTestCase("test_connection_members", req))
    return mpTestSuite


def handler(req):

    if PY2:
        out = io.BytesIO()
    else:
        out = io.StringIO()

    tr = unittest.TextTestRunner(out)
    result = tr.run(make_suite(req))

    req.log_error(out.getvalue())

    if result.wasSuccessful():
        req.write("test ok")
    else:
        req.write("test failed")

    return apache.OK

def req_add_handler(req):

    req.secret_message = "foo"
    req.add_handler("PythonHandler", "tests::simple_handler")

    return apache.OK

def simple_handler(req):
    # for req_add_handler()
    if (req.secret_message == "foo"):
        req.write("test ok")

    return apache.OK

def req_add_bad_handler(req):
    # bad_handler does not exist so adding it should
    # should raise an AttributeError exception

    req.log_error("req_add_bad_handler " + req.hlist.handler)
    req.add_handler("PythonHandler", "tests::bad_handler")
    req.log_error("req_add_bad_handler " + req.hlist.handler)
    req.write("test ok")

    return apache.OK

def req_add_empty_handler_string(req):
    # Adding an empty string as a handler should should
    # should raise an exception

    req.log_error("req_add_empty_handler_string")
    req.add_handler("PythonHandler", "")
    req.write("no exception")

    return apache.OK

def req_add_handler_empty_phase(req):
    req.log_error("req_add_handler_empty_phase")
    req.log_error("phase=%s" % req.phase)
    req.log_error("interpreter=%s" % req.interpreter)
    req.log_error("directory=%s" % req.hlist.directory)
    if req.phase != "PythonHandler":
        directory = os.path.dirname(__file__)
        req.add_handler("PythonHandler", "tests::req_add_handler_empty_phase", directory)
    else:
        req.write("test ok")

    return apache.OK

def accesshandler_add_handler_to_empty_hl(req):
    # Prior to version 3.2.6, adding a python handler
    # to and empty handler list would cause a segfault

    req.secret_message = "foo"
    req.log_error("accesshandler_add_handler_to_empty_hl")
    req.add_handler("PythonHandler", "tests::simple_handler")

    return apache.OK

def test_req_add_handler_directory(req):
    # dir1 will not have a trailing slash and on Win32
    # will use back slashes and not forward slashes.
    dir1 = os.path.dirname(__file__)
    if req.phase == "PythonFixupHandler":
        req.add_handler("PythonHandler", "tests::test_req_add_handler_directory", dir1)
    else:
	# dir2 should only use forward slashes and
	# should have a trailing forward slash added by
	# call to req.add_handler(). When dir1 and dir2
	# are normalised for current operating system,
        # they should be equivalent.
        dir2 = req.hlist.directory
        if dir2[-1] != '/' or dir2.count('\\') != 0:
            req.write('test failed')
        else:
            dir1 = os.path.normpath(dir1)
            dir2 = os.path.normpath(dir2)
            if dir2 != dir1:
                req.write('test failed')
            else:
                req.write('test ok')

    return apache.OK


def req_allow_methods(req):

    req.allow_methods(["PYTHONIZE"])
    return apache.HTTP_METHOD_NOT_ALLOWED

def req_get_basic_auth_pw(req):

    LATIN1_SPAM = 'sp\xe1m'
    LATIN1_EGGS = '\xe9ggs'

    pw = req.get_basic_auth_pw()
    if (req.user == "spam" and pw == "eggs" or
        req.user == LATIN1_SPAM and pw == LATIN1_EGGS):
        req.write("test ok")
    else:
        req.write("test failed, user %s, pw %s" % (repr(req.user), repr(pw)))

    return apache.OK

def req_unauthorized(req):

    pw = req.get_basic_auth_pw()
    if req.user == "spam" and pw == "eggs":
        req.write("test ok")
        return apache.OK

    return apache.HTTP_UNAUTHORIZED

def req_auth_type(req):

    auth_type = req.auth_type()
    if auth_type != "dummy":
        req.log_error("auth_type check failed")
        req.write("test failed, req.auth_type() returned: %s" % repr(auth_type))
        return apache.DONE
    auth_name = req.auth_name()
    if auth_name != "blah":
        req.log_error("auth_name check failed")
        req.write("test failed, req.auth_name() returned: %s" % repr(auth_name))
        return apache.DONE

    if req.phase == "PythonAuthenHandler":

        req.user = "dummy"
        req.ap_auth_type = req.auth_type()

    elif req.phase != "PythonAuthzHandler":

        req.write("test ok")

    return apache.OK

def req_requires(req):

    if req.requires() == ('valid-user',):
        req.write("test ok")
        return apache.DONE

    req.write("test failed")
    return apache.DONE

def req_document_root(req):

    req.write(req.document_root())
    return apache.OK

def req_internal_redirect(req):

    req.internal_redirect("/test.int")

    return apache.OK

def req_internal_redirect_int(req):
    # used by req_internal_redirect

    req.prev.write("test ")
    req.write("ok")

    return apache.OK

def req_construct_url(req):

    url = req.construct_url("/index.html")

    if not re.match("^http://test_req_construct_url:[0-9]+/index.html$",url):
        req.write("test failed")
    else:
        req.write("test ok")

    return apache.OK

def req_read(req):

    s = req.read()
    req.write(s)

    return apache.OK

def req_readline(req):

    s = req.readline()
    while s:
        req.write(s)
        s = req.readline()

    return apache.OK

def req_readlines(req):


    if 'SizeHint' in req.headers_in:
        lines = req.readlines(int(req.headers_in['SizeHint']))
    else:
        lines = req.readlines()

    req.write(b"".join(lines))

    return apache.OK

def req_discard_request_body(req):

    s = req.read(10)
    if s != b'1234567890':
        req.log_error('read() #1 returned %s' % repr(s))
        req.write('test failed')
        return apache.OK

    status = req.discard_request_body()
    if status != apache.OK:
        req.log_error('discard_request_body() returned %d' % status)
        return status

    s = req.read()
    if s:
        req.log_error('read() #2 returned %s' % repr(s))
        req.write('test failed')
        return apache.OK

    req.write('test ok')

    return apache.OK

def req_register_cleanup(req):

    req.cleanup_data = "req_register_cleanup test ok"
    req.register_cleanup(cleanup, req)
    req.write("registered cleanup that will write to log")

    return apache.OK

def cleanup(data):
    # for req_register_cleanup above

    data.log_error(data.cleanup_data)

def server_cleanup(data):
    # for srv_register_cleanup and apache_register_cleanup below

    apache.log_error(data)

def req_headers_out(req):

    req.headers_out["X-Test-Header"] = "test ok"
    req.write("test ok")

    return apache.OK

def req_headers_out_access(req):

    return apache.OK

def req_sendfile(req):

    import tempfile
    fname  = tempfile.mktemp("txt")
    f = open(fname, "w")
    f.write("  test ok  ");
    f.close()

    req.sendfile(fname, 2, 7)

    # os.remove(fname)
    return apache.OK

def req_sendfile2(req):

    import tempfile
    fname  = tempfile.mktemp("txt")
    f = open(fname, "w")
    f.write("0123456789"*100);
    f.close()

    req.sendfile(fname)

    # os.remove(fname)
    return apache.OK

def req_sendfile3(req):
    """Check if sendfile handles symlinks properly.
       This is only valid on posix systems.
    """

    import tempfile
    # note mktemp is deprecated in python 2.3. Should use mkstemp instead.
    fname  = tempfile.mktemp("txt")
    f = open(fname, "w")
    f.write("0123456789"*100);
    f.close()
    fname_symlink =  '%s.lnk' % fname
    os.symlink(fname, fname_symlink)
    req.sendfile(fname_symlink)
    os.remove(fname_symlink)
    os.remove(fname)
    return apache.OK

def req_handler(req):
    if req.phase == "PythonFixupHandler":
        req.handler = "mod_python"
        req.handler = None
        req.handler = "mod_python"
        req.add_handler("PythonHandler","tests::req_handler")
        return apache.OK
    elif req.phase == "PythonHandler":
        req.write('test ok')
        return apache.OK
    else:
        req.write('test failed')
        return apache.OK

def req_no_cache(req):
    req.no_cache = 1
    req.write('test ok')
    return apache.OK

def req_update_mtime(req):
    assert(req.mtime == 0.0)
    req.update_mtime(100.0)
    assert(req.mtime == 100.0)
    req.set_etag()
    req.set_last_modified()
    req.write('test ok')
    return apache.OK

def util_redirect(req):
    from mod_python import util
    if req.main:
        # Sub request for ErrorDocument.
        req.write("test failed")
        return apache.DONE
    else:
        if req.phase == "PythonFixupHandler":
            util.redirect(req,location="/dummy",text="test ok")
        else:
            req.write('test failed')
            return apache.OK

def req_server_get_config(req):

    if req.server.get_config().get("PythonDebug", "0") != "1" or \
            req.get_config().get("PythonDebug", "0") != "0":
        req.write('test failed')
    else:
        req.write('test ok')

    return apache.OK

def req_server_get_options(req):

    try:
        server_options = apache.main_server.get_options()
        assert(server_options.get("global","0") == "0")
        assert(server_options.get("override","0") == "0")

        server_options = req.connection.base_server.get_options()
        assert(server_options.get("global","0") == "0")
        assert(server_options.get("override","0") == "0")

        server_options = req.server.get_options()
        assert(server_options["global"] == "1")
        assert(server_options["override"] == "1")

        request_options = req.get_options()
        assert(request_options["global"] == "1")
        assert(request_options["override"] == "2")
        assert(request_options["local"] == "1")
    except:
        req.write('test failed')
    else:
        req.write('test ok')

    return apache.OK

def fileupload(req):
    from mod_python import util

    fields = util.FieldStorage(req)
    f = fields.getfirst('testfile')

    if PY2:
        import md5
        req.write(md5.new(f.file.read()).hexdigest())
    else:
        from hashlib import md5
        req.write(md5(f.file.read()).hexdigest())

    return apache.OK

def srv_register_cleanup(req):

    req.server.register_cleanup(req, server_cleanup, "srv_register_cleanup test ok")
    req.write("registered server cleanup that will write to log")

    return apache.OK

def apache_register_cleanup(req):

    apache.register_cleanup(server_cleanup, "apache_register_cleanup test ok")
    req.write("registered server cleanup that will write to log")

    return apache.OK

def apache_exists_config_define(req):
    if apache.exists_config_define('FOOBAR'):
        req.write('FOOBAR')
    else:
        req.write('NO_FOOBAR')
    return apache.OK

def util_fieldstorage(req):

    from mod_python import util
    req.write(repr(util.FieldStorage(req).list))
    return apache.OK

def postreadrequest(req):

    req.log_error('postreadrequest')

    req.add_common_vars()

    req.subprocess_env['TEST1'] = "'"
    req.subprocess_env['TEST2'] = '"'

    req.log_error('subprocess_env = %s' % req.subprocess_env)
    req.log_error('subprocess_env.values() = %s' % list(req.subprocess_env.values()))

    for value in req.subprocess_env.values():
        req.log_error('VALUE = %s' % value)

    for item in req.subprocess_env.items():
        req.log_error('ITEM = %s' % (item,))

    req.log_error('SCRIPT_FILENAME = %s' % req.subprocess_env.get('SCRIPT_FILENAME'))
    req.log_error('SCRIPT_FILENAME = %s' % req.subprocess_env['SCRIPT_FILENAME'])

    req.write("test ok")

    return apache.DONE


def trans(req):

    req.filename = req.document_root()+"/tests.py"

    return apache.OK

def import_test(req):

    import sys, os
    directory = os.path.dirname(__file__)
    assert([os.path.normpath(d) for d in sys.path].count(directory) == 1)
    if "dummymodule" in sys.modules:
        if "dummymodule::function" not in apache.main_server.get_options():
            req.log_error("dummymodule::function not executed")
            req.write("test failed")
        else:
            req.write("test ok")
    else:
        req.log_error("dummymodule not found in sys.modules")
        req.write("test failed")

    return apache.OK

def outputfilter(fltr):

    assert(not hasattr(fltr, "non_existent"))

    s = fltr.read()
    while s:
        fltr.write(s.upper())
        s = fltr.read()

    if s is None:
        fltr.close()

    return apache.OK

def simplehandler(req):

    if req.phase != "PythonHandler":
        req.write("test failed")
        return apache.OK

    req.write("test ok")
    return apache.OK

def req_add_output_filter(req):

    req.add_output_filter("MP_TEST_FILTER")

    req.write("test ok")

    return apache.OK

def req_register_output_filter(req):

    req.register_output_filter("MP_TEST_FILTER","tests::outputfilter")

    req.add_output_filter("MP_TEST_FILTER")

    req.write("test ok")

    return apache.OK

def connectionhandler(conn):

    # read whatever
    s = conn.readline().strip()
    while s:
        s = conn.readline().strip()

    # fake an HTTP response
    conn.write("HTTP/1.1 200 OK\r\n")
    conn.write("Content-Length: 7\r\n\r\n")
    conn.write("test ok")

    return apache.OK

def pipe_ext(req):

    # this is called by publisher

    return "pipe ext"


def Cookie_Cookie(req):

    from mod_python import Cookie

    cookies = Cookie.get_cookies(req)

    for k in cookies:
        Cookie.add_cookie(req, cookies[k])

    req.write("test ok")

    return apache.OK

def Cookie_MarshalCookie(req):

    from mod_python import Cookie

    cookies = Cookie.get_cookies(req, Cookie.MarshalCookie,
                                secret="secret")

    for k in cookies:
        Cookie.add_cookie(req, cookies[k])

    req.write("test ok")

    return apache.OK


def global_lock(req):

    import _apache

    _apache._global_lock(req.server, 1)
    time.sleep(1)
    _apache._global_unlock(req.server, 1)

    req.write("test ok")

    return apache.OK

def Session_Session(req):

    from mod_python import Session, Cookie
    s = Session.Session(req)
    if s.is_new():
        s.save()

    cookies = Cookie.get_cookies(req)
    if Session.COOKIE_NAME in cookies and s.is_new():
        req.write(str(cookies[Session.COOKIE_NAME]))
    else:
        req.write("test ok")

    return apache.OK

def files_directive(req):

    req.write(str(req.hlist.directory))
    return apache.OK

none_handler = None

def server_return_1(req):
    raise apache.SERVER_RETURN(apache.OK)

def server_return_2(req):
    req.write("test ok")
    return apache.OK

def phase_status_1(req):
    apache.log_error("phase_status_1")
    req.phases = [1]
    return apache.DECLINED

def phase_status_2(req):
    apache.log_error("phase_status_2")
    req.phases.append(2)
    req.user = "bogus"
    req.ap_auth_type = "bogus"
    return apache.OK

def phase_status_3(req):
    apache.log_error("phase_status_3")
    req.phases.append(3)
    return apache.OK

def phase_status_4(req):
    apache.log_error("phase_status_4")
    #req.phases.append(4)
    return apache.OK

def phase_status_5(req):
    apache.log_error("phase_status_5")
    req.phases.append(5)
    return apache.DECLINED

def phase_status_6(req):
    apache.log_error("phase_status_6")
    req.phases.append(6)
    return apache.OK

def phase_status_7(req):
    apache.log_error("phase_status_7")
    req.phases.append(7)
    return apache.OK

def phase_status_8(req):
    apache.log_error("phase_status_8")
    apache.log_error("phases = %s" % req.phases)
    if req.phases != [1, 2, 5, 6, 7]:
        req.write("test failed")
    else:
        req.write("test ok")
    return apache.OK

def phase_status_cleanup(req):
    apache.log_error("phase_status_cleanup_log_entry")
    return apache.OK

def test_sys_argv(req):
    import sys
    req.write(repr(sys.argv))
    return apache.OK

def PythonOption_items(req):
    options = list(req.get_options().items())

    # The tests may using PythonOption mod_python.* in the test configuration
    # We need to remove those particular options so they don't interfer
    # with this test result.
    options = [ o for o in options if not o[0].startswith('mod_python') ]

    options.sort()
    req.write(str(options))
    return apache.OK

def interpreter(req):
    req.write(req.interpreter)
    return apache.DONE

def index(req):
    return "test ok, interpreter=%s" % req.interpreter

def test_publisher(req):
    return "test ok, interpreter=%s" % req.interpreter

def test_publisher_auth_nested(req):
    def __auth__(req, user, password):
        test_globals = test_publisher
        req.notes["auth_called"] = "1"
        return user == "spam" and password == "eggs"
    def __access__(req, user):
        req.notes["access_called"] = "1"
        return 1
    assert(int(req.notes.get("auth_called",0)))
    assert(int(req.notes.get("access_called",0)))
    return "test ok, interpreter=%s" % req.interpreter

class _test_publisher_auth_method_nested:
    def method(self, req):
        def __auth__(req, user, password):
            test_globals = test_publisher
            req.notes["auth_called"] = "1"
            return user == "spam" and password == "eggs"
        def __access__(req, user):
            req.notes["access_called"] = "1"
            return 1
        assert(int(req.notes.get("auth_called",0)))
        assert(int(req.notes.get("access_called",0)))
        return "test ok, interpreter=%s" % req.interpreter

test_publisher_auth_method_nested = _test_publisher_auth_method_nested()

class OldStyleClassTest:
    def __init__(self):
        pass
    def __call__(self, req):
        return "test callable old-style instance ok"
    def traverse(self, req):
        return "test traversable old-style instance ok"
old_instance = OldStyleClassTest()

test_dict = {1:1, 2:2, 3:3}
test_dict_keys = test_dict.keys

def test_dict_iteration(req):
    return test_dict_keys()

def test_generator(req):
    c = 0
    while c < 10:
        yield c
        c += 1

def server_side_include(req):
    req.ssi_globals = { "data": "test" }
    return apache.OK

class InstanceTest(object):
    def __call__(self, req):
        return "test callable instance ok"
    def traverse(self, req):
        return "test traversable instance ok"
instance = InstanceTest()

# Hierarchy traversal tests
class Mapping(object):
    def __init__(self,name):
        self.name = name

    def __call__(self,req):
        return "Called %s"%self.name
hierarchy_root = Mapping("root");
hierarchy_root.page1 = Mapping("page1")
hierarchy_root.page1.subpage1 = Mapping("subpage1")
hierarchy_root.page2 = Mapping("page2")

class Mapping2:
    pass
hierarchy_root_2 = Mapping2()
hierarchy_root_2.__call__ = index
hierarchy_root_2.page1 = index
hierarchy_root_2.page2 = index

def _test_table():

    log = apache.log_error

    log("    starting _test_table")
    d = apache.table()
    if list(d.keys()) != []: raise TestFailed('{}.keys()')
    if ('a' in d) != 0: raise TestFailed("'a' in {}")
    if ('a' not in d) != 1: raise TestFailed("'a' not in {}")
    if len(d) != 0: raise TestFailed('len({})')
    d = {'a': 1, 'b': 2}
    if len(d) != 2: raise TestFailed('len(dict)')
    k = list(d.keys())
    k.sort()
    if k != ['a', 'b']: raise TestFailed('dict keys()')
    if 'a' in d and 'b' in d and 'c' not in d: pass
    else: raise TestFailed('dict keys()')
    if 'a' in d and 'b' in d and 'c' not in d: pass
    else: raise TestFailed('dict keys() # in/not in version')
    if d['a'] != 1 or d['b'] != 2: raise TestFailed('dict item')
    d['c'] = 3
    d['a'] = 4
    if d['c'] != 3 or d['a'] != 4: raise TestFailed('dict item assignment')
    del d['b']
    if d != {'a': 4, 'c': 3}: raise TestFailed('dict item deletion')

    # dict.clear()
    log("    table.clear()")
    d = apache.table()
    d['1'] = '1'
    d['2'] = '2'
    d['3'] = '3'
    d.clear()
    if d != apache.table(): raise TestFailed('dict clear')

    # dict.update()
    log("    table.update()")
    d.update({'1':'100'})
    d.update({'2':'20'})
    d.update({'1':'1', '2':'2', '3':'3'})
    if d != apache.table({'1':'1', '2':'2', '3':'3'}): raise TestFailed('dict update')
    d.clear()
    try: d.update(None)
    except AttributeError: pass
    else: raise TestFailed('dict.update(None), AttributeError expected')
    class SimpleUserDict:
        def __init__(self):
            self.d = {1:1, 2:2, 3:3}
        def keys(self):
            return list(self.d.keys())
        def __getitem__(self, i):
            return self.d[i]
    d.update(SimpleUserDict())
    if d != apache.table({1:1, 2:2, 3:3}): raise TestFailed('dict.update(instance)')
    d.clear()
    class FailingUserDict:
        def keys(self):
            raise ValueError
    try: d.update(FailingUserDict())
    except ValueError: pass
    else: raise TestFailed('dict.keys() expected ValueError')
    class FailingUserDict:
        def keys(self):
            class BogonIter:
                def __iter__(self):
                    raise ValueError
            return BogonIter()
    try: d.update(FailingUserDict())
    except ValueError: pass
    else: raise TestFailed('iter(dict.keys()) expected ValueError')
    class FailingUserDict:
        def keys(self):
            class BogonIter:
                def __init__(self):
                    self.i = 1
                def __iter__(self):
                    return self
                def __next__(self):
                    if self.i:
                        self.i = 0
                        return 'a'
                    raise ValueError
                def next(self):
                    return self.__next__()
            return BogonIter()
        def __getitem__(self, key):
            return key
    try: d.update(FailingUserDict())
    except ValueError: pass
    else: raise TestFailed('iter(dict.keys()).next() expected ValueError')
    class FailingUserDict:
        def keys(self):
            class BogonIter:
                def __init__(self):
                    self.i = ord('a')
                def __iter__(self):
                    return self
                def __next__(self):
                    if self.i <= ord('z'):
                        rtn = chr(self.i)
                        self.i += 1
                        return rtn
                    raise StopIteration
                def next(self):
                    return self.__next__()
            return BogonIter()
        def __getitem__(self, key):
            raise ValueError
    try: d.update(FailingUserDict())
    except ValueError: pass
    else: raise TestFailed('dict.update(), __getitem__ expected ValueError')
    # dict.copy()
    log("    table.copy()")
    d = {1:1, 2:2, 3:3}
    if d.copy() != {1:1, 2:2, 3:3}: raise TestFailed('dict copy')
    if apache.table().copy() != apache.table(): raise TestFailed('empty dict copy')
    # dict.get()
    log("    table.get()")
    d = apache.table()
    if d.get('c') is not None: raise TestFailed('missing {} get, no 2nd arg')
    if d.get('c', '3') != '3': raise TestFailed('missing {} get, w/ 2nd arg')
    d = apache.table({'a' : '1', 'b' : '2'})
    if d.get('c') is not None: raise TestFailed('missing dict get, no 2nd arg')
    if d.get('c', '3') != '3': raise TestFailed('missing dict get, w/ 2nd arg')
    if d.get('a') != '1': raise TestFailed('present dict get, no 2nd arg')
    if d.get('a', '3') != '1': raise TestFailed('present dict get, w/ 2nd arg')
    # dict.setdefault()
    log("    table.setdefault()")
    d = apache.table()
    d.setdefault('key0')
    if d.setdefault('key0') is not "":
        raise TestFailed('missing {} setdefault, no 2nd arg')
    if d.setdefault('key0') is not "":
        raise TestFailed('present {} setdefault, no 2nd arg')
    # dict.popitem()
    log("    table.popitem()")
    for copymode in -1, +1:
        # -1: b has same structure as a
        # +1: b is a.copy()
        for log2size in range(12):
            size = 2**log2size
            a = apache.table()
            b = apache.table()
            for i in range(size):
                a[repr(i)] = str(i)
                if copymode < 0:
                    b[repr(i)] = str(i)
            if copymode > 0:
                b = a.copy()
            for i in range(size):
                ka, va = ta = a.popitem()
                if va != ka: raise TestFailed("a.popitem: %s" % str(ta))
                kb, vb = tb = b.popitem()
                if vb != kb: raise TestFailed("b.popitem: %s" % str(tb))
                if copymode < 0 and ta != tb:
                    raise TestFailed("a.popitem != b.popitem: %s, %s" % (
                        str(ta), str(tb)))
            if a: raise TestFailed('a not empty after popitems: %s' % str(a))
            if b: raise TestFailed('b not empty after popitems: %s' % str(b))

    # iteration (just make sure we can iterate without a segfault)
    d = apache.table({'a' : '1', 'b' : '2', 'c' : '3'})
    log("    for k in table")
    for k in d:
        pass

    log("    _test_table test finished")

def okay(req):
    req.write("test ok")
    return apache.OK

def memory(req):

    # NB: This only works on Linux.

    ## warm up
    for x in range(10000):
        req.write("test ok")

    ## check memory usage before (the unit is pages, usually 4k)
    before = list(map(int, open("/proc/self/statm").read().split()))

    for x in range(100000):
        req.write("test ok")
        req.flush()

    ## check memory usage after
    after = list(map(int, open("/proc/self/statm").read().split()))

    req.write("|%s|%s" % (before[0], after[0]))

    return apache.OK