Cloud DNS testing

clouddns-test.py

import httplib
import urlparse
import json
import os


def get_auth(username, key, auth_url):
    u = urlparse.urlparse(auth_url)
    cls = u.scheme == 'https' and httplib.HTTPSConnection or \
        httplib.HTTPConnection
    h = cls(u.netloc)
    headers = {"X-Auth-User": username,
               "X-Auth-Key": key,
               }
    h.request("GET", u.path, headers=headers)
    ret = {}
    for k, v in h.getresponse().getheaders():
        if k.startswith("x-"):
            ret[k.replace("x-", "")] = v
    return ret


def req(auth, rest, method="GET", body=None):
    surl = urlparse.urlparse(auth["server-management-url"])
    cls = surl.scheme == 'https' and httplib.HTTPSConnection or \
        httplib.HTTPConnection

    host = 'dns.api.rackspacecloud.com'
    if 'lon.' in surl.netloc:
        host = 'lon.' + host

    h = cls(host)
    headers = {"X-Auth-Token": auth['auth-token']}
    url = surl.path + rest
    h.request(method, url, headers=headers, body=body)
    resp = h.getresponse()
    return resp


def list_domains(auth):
    return json.loads(req(auth, "/domains").read())['domains']['domain']


def get_domain(auth, domain_id):
    ret = json.loads(req(auth, "/domains/%d" % \
                             (domain_id)).read())
    if not ret:
        return
    return ret['domains']['domain']


def delete_record(auth, domain_id, record_id):
    response = req(auth, "/domains/%d/records/%s" % \
                       (domain_id, record_id),
                   method="DELETE")
    if response.status // 100 != 2:
        raise Exception(response.read())

    return response.status


def add_record(auth, domain_id, dnstype, data, name):
    dico = {'name': name,
            'data': data,
            'type': dnstype,
            }
    # Json not working yet, using XML instead still need to figure out
    # why (and why can't you set the ttl directly')
    xmlblob = """<records xmlns="http://docs.rackspacecloud.com/dns/api/v1.0">
    <record type="%(type)s" data="%(data)s" name="%(name)s"/>
</records>
""" % dico
    response = req(auth, "/domains/%d/records/" % \
                       (domain_id), method="POST", body=xmlblob)
    if response.status // 100 != 2:
        raise Exception(response.read())
    return response.read()


def modify_record(auth, domain_id, record_id, data, name):
    dico = {'name': name,
            'data': data,
            'record_id': record_id,
            }

    xmlblob = """<records xmlns="http://docs.rackspacecloud.com/dns/api/v1.0">
    <record id="%(record_id)s" data="%(data)s" name="%(name)s" ttl="1800"/>
</records>
""" % dico

    response = req(auth, "/domains/%d/records/" % \
                       (domain_id), method="PUT", body=xmlblob)

    if response.status // 100 != 2:
        raise Exception(response.read())
    return response.read()


def test3():
    auth = get_auth(os.environ.get('US_RCLOUD_USER'),
                    os.environ.get('US_RCLOUD_KEY'),
                    os.environ.get('US_RCLOUD_AURL'))
    domain_id = 2785869
    record_id = "A-6936918"
    delete_record(auth, domain_id, record_id)


def test2():
    record_id = "A-6922227"
    domain_id = 2785869
    auth = get_auth(os.environ.get('US_RCLOUD_USER'),
                    os.environ.get('US_RCLOUD_KEY'),
                    os.environ.get('US_RCLOUD_AURL'))
    modify_record(auth, domain_id, record_id,
                  "192.168.1.200", "local3.rackspaceclouduk.com")


def test():
    raw_input("Authenticating: ")
    DOMAIN = "rackspaceclouduk.com"
    auth = get_auth(os.environ.get('US_RCLOUD_USER'),
                    os.environ.get('US_RCLOUD_KEY'),
                    os.environ.get('US_RCLOUD_AURL'))

    raw_input("Listing domains: ")
    all_domains = list_domains(auth)

    domain_id = None
    for row in all_domains:
        print row
        if row['name'] == DOMAIN:
            domain_id = row['id']

    if not domain_id:
        print "Cannot find your domain: %s" % (DOMAIN)
        sys.exit(1)

    raw_input("Getting details for domain %s: " % (DOMAIN))
    domain = get_domain(auth, domain_id)
    from pprint import pprint as p
    p(domain)

    a = raw_input("Adding record t.rackspaceclouduk.com to 127.0.0.1: ")
    if a != "n":
        x = add_record(auth, domain_id, "A", "127.0.0.1", \
                           't.rackspaceclouduk.com')
        print x

test1 = test

if __name__ == '__main__':
    import sys
    test2()
    sys.exit(0)
    # auth = get_auth(os.environ.get('US_RCLOUD_USER'),
    #                 os.environ.get('US_RCLOUD_KEY'),
    #                 os.environ.get('US_RCLOUD_AURL'))

    # domain_id = 2785869
    # record_id = 'A-6922373'

    # # Get All domains
    # domains = list_domains(auth)
    # from pprint import pprint as p
    # p(domains)
    # sys.exit(0)
    # # Get detail of the domain
    # domain = get_domain(auth, domain_id)

    # #Add records
    # x = add_record(auth, domain_id, "A", "127.0.0.1", \
    #                    'local2.rackspaceclouduk.com')