Merge pull request #6136 from zeha/apitests

[thirdparty/pdns.git] / regression-tests.api / runtests.py

#!/usr/bin/env python
#
# Shell-script style.

import os
import requests
import shutil
import subprocess
import sys
import tempfile
import time

SQLITE_DB = 'pdns.sqlite3'
WEBPORT = 5556
DNSPORT = 5300
APIKEY = '1234567890abcdefghijklmnopq-key'
PDNSUTIL_CMD = ["../pdns/pdnsutil", "--config-dir=."]

NAMED_CONF_TPL = """
# Generated by runtests.py
options { directory "../regression-tests/zones/"; };
zone "example.com" { type master; file "example.com"; };
zone "powerdnssec.org" { type master; file "powerdnssec.org"; };
zone "cryptokeys.org" { type master; file "cryptokeys.org"; };
"""

AUTH_CONF_TPL = """
# Generated by runtests.py
launch=gsqlite3,bind
gsqlite3-dnssec=on
gsqlite3-database="""+SQLITE_DB+"""
module-dir=../regression-tests/modules
bind-config=bindbackend.conf
"""

BINDBACKEND_CONF_TPL = """
# Generated by runtests.py
"""

ACL_LIST_TPL = """
# Generated by runtests.py
# local host
127.0.0.1
::1
"""

REC_EXAMPLE_COM_CONF_TPL = """
# Generated by runtests.py
auth-zones+=example.com=../regression-tests/zones/example.com
"""

REC_CONF_TPL = """
# Generated by runtests.py
auth-zones=
forward-zones=
forward-zones-recurse=
api-config-dir=%(conf_dir)s
include-dir=%(conf_dir)s
"""


def ensure_empty_dir(name):
    if os.path.exists(name):
        shutil.rmtree(name)
    os.mkdir(name)


def format_call_args(cmd):
    return "$ '%s'" % ("' '".join(cmd))


def run_check_call(cmd, *args, **kwargs):
    print format_call_args(cmd)
    subprocess.check_call(cmd, *args, **kwargs)


wait = ('--wait' in sys.argv)
if wait:
    sys.argv.remove('--wait')

tests = [opt for opt in sys.argv if opt.startswith('--tests=')]
if tests:
    for opt in tests:
        sys.argv.remove(opt)
tests = [opt.split('=', 1)[1] for opt in tests]

daemon = (len(sys.argv) == 2) and sys.argv[1] or None
if daemon not in ('authoritative', 'recursor'):
    print "Usage: ./runtests (authoritative|recursor)"
    sys.exit(2)

daemon = sys.argv[1]

pdns_server = os.environ.get("PDNSSERVER", "../pdns/pdns_server")
pdns_recursor = os.environ.get("PDNSRECURSOR", "../pdns/recursordist/pdns_recursor")
common_args = [
    "--daemon=no", "--socket-dir=.", "--config-dir=.",
    "--local-address=127.0.0.1", "--local-port="+str(DNSPORT),
    "--webserver=yes", "--webserver-port="+str(WEBPORT), "--webserver-address=127.0.0.1", "--webserver-password=something",
    "--api-key="+APIKEY
]

if daemon == 'authoritative':

    # Prepare sqlite DB with some zones.
    run_check_call(["rm", "-f", SQLITE_DB])
    run_check_call(["make", "-C", "../pdns", "zone2sql"])

    with open('../modules/gsqlite3backend/schema.sqlite3.sql', 'r') as schema_file:
        run_check_call(["sqlite3", SQLITE_DB], stdin=schema_file)

    with open('named.conf', 'w') as named_conf:
        named_conf.write(NAMED_CONF_TPL)
    with tempfile.TemporaryFile() as tf:
        p = subprocess.Popen(["../pdns/zone2sql", "--transactions", "--gsqlite", "--named-conf=named.conf"], stdout=tf)
        p.communicate()
        if p.returncode != 0:
            raise Exception("zone2sql failed")
        tf.seek(0, os.SEEK_SET)  # rewind
        run_check_call(["sqlite3", SQLITE_DB], stdin=tf)

    with open('bindbackend.conf', 'w') as bindbackend_conf:
        bindbackend_conf.write(BINDBACKEND_CONF_TPL)

    with open('pdns.conf', 'w') as pdns_conf:
        pdns_conf.write(AUTH_CONF_TPL)

    run_check_call(PDNSUTIL_CMD + ["secure-zone", "powerdnssec.org"])
    servercmd = [pdns_server] + common_args + ["--local-ipv6=", "--no-shuffle", "--dnsupdate=yes", "--cache-ttl=0", "--api=yes"]

else:
    conf_dir = 'rec-conf.d'
    ensure_empty_dir(conf_dir)
    with open('acl.list', 'w') as acl_list:
        acl_list.write(ACL_LIST_TPL)
    with open('recursor.conf', 'w') as recursor_conf:
        recursor_conf.write(REC_CONF_TPL % locals())
    with open(conf_dir+'/example.com..conf', 'w') as conf_file:
        conf_file.write(REC_EXAMPLE_COM_CONF_TPL)

    servercmd = [pdns_recursor] + common_args + ["--allow-from-file=acl.list"]


# Now run pdns and the tests.
print "Launching server..."
print format_call_args(servercmd)
serverproc = subprocess.Popen(servercmd, close_fds=True)

print "Waiting for webserver port to become available..."
available = False
for try_number in range(0, 10):
    try:
        res = requests.get('http://127.0.0.1:%s/' % WEBPORT)
        available = True
        break
    except:
        time.sleep(0.5)

if not available:
    print "Webserver port not reachable after 10 tries, giving up."
    serverproc.terminate()
    serverproc.wait()
    sys.exit(2)

print "Running tests..."
returncode = 0
test_env = {}
test_env.update(os.environ)
test_env.update({
    'WEBPORT': str(WEBPORT),
    'APIKEY': APIKEY,
    'DAEMON': daemon,
    'SQLITE_DB': SQLITE_DB,
    'PDNSUTIL_CMD': ' '.join(PDNSUTIL_CMD),
})

try:
    print ""
    run_check_call(["nosetests", "--with-xunit", "-v"] + tests, env=test_env)
except subprocess.CalledProcessError as ex:
    returncode = ex.returncode
finally:
    if wait:
        print "Waiting as requested, press ENTER to stop."
        raw_input()
    serverproc.terminate()
    serverproc.wait()

sys.exit(returncode)