More works and changes.

TODO

@@ -1 +1,3 @@
-Filtering records.
+search domains by Filtering.
+export to file (?import)
+

clouddns/connection.py

@@ -118,7 +118,8 @@ def retry_request():
             return self.connection.getresponse()
 
         try:
-            if 'PYTHON_CLOUDDNS_DEBUG' in os.environ:
+            if 'PYTHON_CLOUDDNS_DEBUG' in os.environ and \
+                    os.environ['PYTHON_CLOUDDNS_DEBUG'].strip():
                 import sys
                 sys.stderr.write("URL: %s\n" % (path))
                 sys.stderr.write("ARGS: %s\n" % (str(data)))
@@ -149,7 +150,7 @@ def get_domain(self, domain_id):
         pass
 
     # Take a reponse parse it if there is asyncResponse and wait for
-    # it
+    # it (should offer not too wait for it)
     def wait_for_async_request(self, response):
         if (response.status < 200) or (response.status > 299):
             response.read()
@@ -175,8 +176,6 @@ def wait_for_async_request(self, response):
             else:
                 return output
 
-    #TODO: We make it syncronous here, we should offer async as well
-    #      unlike API does (choice).
     def create_domain(self, name, ttl, emailAddress):
         if not ttl >= 300:
             raise Exception("Ttl is a minimun of 300 seconds")
@@ -190,7 +189,7 @@ def create_domain(self, name, ttl, emailAddress):
 
         if 'domains' in output:
             domain = output["domains"]["domain"]
-            return Domain(**domain[0])
+            return Domain(connection=self, **domain[0])
         else:
             raise Exception("This should not happen")
 

clouddns/domain.py

@@ -56,7 +56,7 @@ def get_record(self, record_id):
         records = self.list_records_info()
         for rec in  records:
             if rec['id'] == record_id:
-                return Record(self, record=rec)
+                return Record(self, **rec)
 
         #TODO: Exceptions
         raise Exception("Not found")
@@ -85,11 +85,45 @@ def __getitem__(self, key):
     def __str__(self):
         return self.name
 
-    def create_record(self, record_name):
+    def update(self, name=None,
+               ttl=None,
+               emailAddress=None):
+        xml = '<domain xmlns="http://docs.rackspacecloud.com/dns/api/v1.0" '
+
+        if name:
+            xml += ' name="%s"' % (name)
+            self.name = name
+        if ttl:
+            xml += ' ttl="%s"' % (ttl)
+            self.ttl = ttl
+        if emailAddress:
+            xml += ' emailAddress="%s"' % (emailAddress)
+            self.emailAddress = emailAddress
+        xml += ' />'
+
+        response = self.conn.make_request('PUT', ["domains", self.id],
+                                          data=xml)
+        output = self.conn.wait_for_async_request(response)
+        return output
+
+    def _record(self, name, data, type):
         pass
 
-    def delete_record(self, record_name):
-        pass
+    def create_record(self, name, data, type):
+        xml="""<records xmlns="http://docs.rackspacecloud.com/dns/api/v1.0">
+<record type="%(type)s" data="%(data)s" name="%(name)s"/>
+</records>
+        """ % locals()
+        response = self.conn.make_request('POST', ['domains', self.id, 'records'], data=xml)
+        output = self.conn.wait_for_async_request(response)
+        if 'records' in output:
+            record = output["records"]["record"]
+            return Record(domain=self, **record[0])
+        else:
+            raise Exception("This should not happen")
+
+    def delete_record(self, record_id):
+        response = self.conn.make_request('DELETE', ['domains', self.id, 'records', record_id])
 
 
 class DomainResults(object):

clouddns/record.py

@@ -3,21 +3,46 @@
 
 
 class Record(object):
-    def __init__(self, domain, record=None, name=None):
+    def __init__(self, domain,
+                 data=None,
+                 ttl=1800,
+                 name=None,
+                 type=None,
+                 id=None):
         self.domain = domain
+        self.data = data
+        self.name = name
+        self.id = id
+        self.ttl = ttl
+        self.type = type
 
-        if record:
-            self.data = record['data']
-            self.name = record['name']
-            self.id = record['id']
-            self.type = record['type']
-            self.ttl = record['ttl']
-        else:
-            self.data = None
-            self.name = None
-            self.id = None
-            self.ttl = 1800
-            self.type = None
+    def update(self, data=None,
+               name=None,
+               ttl=None,
+               type=None,
+               ):
+        def build_it(k, v, d):
+            if v:
+                return ' %s="%s"' % (k, v)
+            else:
+                return ' %s="%s"' % (k, d)
+        xml = '<record '
+        xml += 'id="%s"' % (self.id)
+        xml += build_it('name', name, self.name)
+        xml += build_it('ttl', ttl, self.ttl)
+        xml += build_it('data', data, self.data)
+        #xml += build_it('type', data, self.type)
+        xml += ' />'
+        print xml
+        response = self.domain.conn.make_request('PUT', ["domains",
+                                                  self.domain.id,
+                                                  "records", self.id, ""],
+                                          data=xml)
+        output = self.domain.conn.wait_for_async_request(response)
+        return output
+
+    def __str__(self):
+        return self.name
 
 
 class RecordResults(object):
@@ -37,7 +62,7 @@ def __init__(self, domain, records=None):
         self.domain = domain
 
     def __getitem__(self, key):
-        return Record(self.domain, record=self._records[key])
+        return Record(self.domain, **(self._records[key]))
 
     def __getslice__(self, i, j):
         return [Record(self.domain, record_id=k) \

tests/authentication_test.py

@@ -0,0 +1,42 @@
+import unittest
+from clouddns.authentication import BaseAuthentication as Auth
+from misc import printdoc
+
+
+class AuthenticationTest(unittest.TestCase):
+    """
+    Freerange Authentication class tests.
+    """
+
+    def test_get_uri(self):
+        """
+        Validate authentication uri construction.
+        """
+        self.assert_(self.auth.uri == "v1.0", \
+               "authentication URL was not properly constructed")
+
+    @printdoc
+    def test_authenticate(self):
+        """
+        Sanity check authentication method stub (lame).
+        """
+        self.assert_(self.auth.authenticate() == (None, None, None), \
+               "authenticate() did not return a two-tuple")
+
+    @printdoc
+    def test_headers(self):
+        """
+        Ensure headers are being set.
+        """
+        self.assert_(self.auth.headers['x-auth-user'] == 'jsmith', \
+               "storage user header not properly assigned")
+        self.assert_(self.auth.headers['x-auth-key'] == 'xxxxxxxx', \
+               "storage password header not properly assigned")
+
+    def setUp(self):
+        self.auth = Auth('jsmith', 'xxxxxxxx')
+
+    def tearDown(self):
+        del self.auth
+
+# vim:set ai ts=4 tw=0 sw=4 expandtab:

tests/t.py

@@ -0,0 +1,69 @@
+# -*- encoding: utf-8 -*-
+__author__ = "Chmouel Boudjnah <[email protected]>"
+import clouddns
+import os
+
+US_RCLOUD_USER = os.environ.get("US_RCLOUD_USER")
+US_RCLOUD_KEY = os.environ.get("US_RCLOUD_KEY")
+CNX = None
+
+def auth():
+    global CNX
+    if not CNX:
+        CNX = clouddns.connection.Connection(US_RCLOUD_USER, US_RCLOUD_KEY)
+    return CNX
+
+
+def test_create_delete(cnx):
+    # CREATE
+    domain_created = cnx.create_domain(name="chmoutest21.com",
+                                       ttl=300,
+                                       emailAddress="[email protected]")
+    # Delete
+    cnx.delete_domain(domain_created.id)
+
+
+def test():
+    cnx = auth()
+    #test_create_delete(cnx)
+
+    # Domain list
+    all_domains = cnx.get_domains()
+    # __getitem__
+    domain = all_domains[0]
+    # __getslice__
+    domain = all_domains[0:1][0]
+    # __contains__
+    assert(str(domain) in all_domains)
+    # __len__
+    len(all_domains)
+
+    for x in all_domains:
+        if str(x).startswith("chmoutest"):
+            print "Cleaning: %s" % (x.name)
+            cnx.delete_domain(x.id)
+
+    # Create Domain
+    domain_created = cnx.create_domain(name="chmoutesting.com",
+                                       ttl=300,
+                                       emailAddress="[email protected]")
+
+    domain = domain_created
+
+    ttl = 500
+    # Update Domain
+    domain.update(ttl=ttl)
+
+    assert(domain.ttl == ttl)
+
+    # Get All records
+    records = domain.get_records()
+    record = records[0]
+
+    # Get Record by ID
+    record_id = record.id
+    record = domain.get_record(record_id)
+
+    from IPython.Shell import IPShellEmbed;IPShellEmbed()()
+
+test()