Merge lp:~thisfred/desktopcouch/fix-unique-id-test into lp:desktopcouch

[Eric Casteleijn]

This fixes get_my_host_unique_id to pass the context it gets to the methods it calls, so that tests talk to the correct database.

This fixes get_my_host_unique_id to pass the context it gets to the methods it calls, so that tests talk to the correct database.

[Manuel de la Peña (mandel)]

+1 I like those pylint error out too :)

[dobey (dobey)]

review: Approve

[Vincenzo Di Somma (vds)]

+1

review: Approve

[Manuel de la Peña (mandel)]

me stupid :P, I forgot to approve

review: Approve

=== modified file 'desktopcouch/pair/couchdb_pairing/couchdb_io.py'
--- desktopcouch/pair/couchdb_pairing/couchdb_io.py	2010-11-03 02:47:32 +0000
+++ desktopcouch/pair/couchdb_pairing/couchdb_io.py	2010-11-09 17:39:48 +0000
@@ -33,22 +33,29 @@
 PAIRED_SERVER_RECORD_TYPE = RECTYPE_BASE + "paired_server"
 MY_ID_RECORD_TYPE = RECTYPE_BASE + "server_identity"
 
-def obsfuscate(d):
-    def maybe_hide(k, v):
-        if hasattr(k, "endswith") and k.endswith("secret"):
-            return "".join(rep for rep, vi in zip(cycle('Hidden'), v))
+
+def obsfuscate(dictionary):
+    """Obfuscate a string by replacing all secret values."""
+    def maybe_hide(key, value):
+        """Replace with gibberish where necessary."""
+        if hasattr(key, "endswith") and key.endswith("secret"):
+            return "".join(rep for rep, vi in zip(cycle('Hidden'), value))
         else:
-            return v
-
-    if not hasattr(d, "iteritems"):
-        return d
-    return dict((k, maybe_hide(k, obsfuscate(v))) for k, v in d.iteritems())
+            return value
+
+    if not hasattr(dictionary, "iteritems"):
+        return dictionary
+    return dict(
+        (key, maybe_hide(key, obsfuscate(value))) for key, value
+        in dictionary.iteritems())
+
 
 def mkuri(hostname, port, has_ssl=False, path="", auth_pair=None):
     """Create a URI from parts."""
     protocol = "https" if has_ssl else "http"
     if auth_pair:
-        auth = (":".join(map(urllib.quote, auth_pair)) + "@")
+        auth = (":".join(
+            map(urllib.quote, auth_pair)) + "@")  # pylint: disable-msg=W0141
     else:
         auth = ""
     if (protocol, port) in (("http", 80), ("https", 443)):
@@ -57,13 +64,14 @@
         port = str(port)
         return "%s://%s%s:%s/%s" % (protocol, auth, hostname, port, path)
 
-def _get_db(name, create=True, uri=None,
-        ctx=local_files.DEFAULT_CONTEXT):
+
+def _get_db(name, create=True, uri=None, ctx=local_files.DEFAULT_CONTEXT):
     """Get (and create?) a database."""
     return server.CouchDatabase(name, create=create, uri=uri, ctx=ctx)
 
+
 def put_paired_host(oauth_data, uri=None,
-        ctx=local_files.DEFAULT_CONTEXT, **kwargs):
+                    ctx=local_files.DEFAULT_CONTEXT, **kwargs):
     """Create a new paired-host record.  OAuth information is required, and
     after the uri, keyword parameters are added to the record."""
     pairing_id = str(uuid.uuid4())
@@ -80,25 +88,28 @@
             "token_secret": str(oauth_data["token_secret"]),
         }
     data.update(kwargs)
-    d = _get_db("management", uri=uri, ctx=ctx)
-    r = Record(data)
-    record_id = d.put_record(r)
+    database = _get_db("management", uri=uri, ctx=ctx)
+    record = Record(data)
+    record_id = database.put_record(record)
     return record_id
 
+
 def put_static_paired_service(oauth_data, service_name, uri=None,
-        ctx=local_files.DEFAULT_CONTEXT):
+                              ctx=local_files.DEFAULT_CONTEXT):
     """Create a new service record."""
     return put_paired_host(oauth_data, uri=uri, service_name=service_name,
             pull_from_server=True, push_to_server=True, ctx=ctx)
 
+
 def put_dynamic_paired_host(hostname, remote_uuid, oauth_data, uri=None,
-        ctx=local_files.DEFAULT_CONTEXT):
+                            ctx=local_files.DEFAULT_CONTEXT):
     """Create a new dynamic-host record."""
     return put_paired_host(oauth_data, uri=uri, pairing_identifier=remote_uuid,
             push_to_server=True, server=hostname, ctx=ctx)
 
-def get_static_paired_hosts(uri=None,
-        ctx=local_files.DEFAULT_CONTEXT, port=None):
+
+def get_static_paired_hosts(uri=None,  # pylint: disable-msg=R0914
+                            ctx=local_files.DEFAULT_CONTEXT, port=None):
     """Retreive a list of static hosts' information in the form of
     (ID, service name, to_push, to_pull) ."""
     if not uri and port is not None:
@@ -121,6 +132,7 @@
     logging.debug("static pairings are %s", unique_hosts)
     return unique_hosts
 
+
 def get_database_names_replicatable(uri, oauth_tokens=None, service=False,
         ctx=local_files.DEFAULT_CONTEXT, user_id=None):
     """Find a list of local databases, minus dbs that we do not want to
@@ -143,7 +155,7 @@
                 "Got list of databases from %s/_all_dbs?user_id=%s:\n    %s",
                 couchdb_server, user_id, all_dbs)
 
-    except socket.error, e:
+    except socket.error:
         logging.error("Can't get list of databases from %s", couchdb_server)
         return set()
 
@@ -158,14 +170,17 @@
 
     return set([n for n in all_dbs - excluded if not n.startswith("_")])
 
+
 def get_my_host_unique_id(uri=None, create=True,
-        ctx=local_files.DEFAULT_CONTEXT):
+                          ctx=local_files.DEFAULT_CONTEXT):
     """Returns a list of ids we call ourselves.  We complain in the log if it's
     more than one, but it's really no error.  If there are zero (id est, we've
     never paired with anyone), then returns None."""
 
     db = _get_db("management", uri=uri, ctx=ctx)
-    ids = _get_management_data(MY_ID_RECORD_TYPE, "self_identity", uri=uri)
+    db.ensure_full_commit()
+    ids = _get_management_data(
+        MY_ID_RECORD_TYPE, "self_identity", uri=uri, ctx=ctx)
     ids = list(set(ids))  # uniqify
     if len(ids) > 1:
         logging.error("DANGER!  We have more than one record claiming to be "
@@ -178,16 +193,15 @@
     if not create:
         return None
     else:
-        data = { "self_identity": str(uuid.uuid4()),
-                "record_type": MY_ID_RECORD_TYPE,
-            }
+        data = {"self_identity": str(uuid.uuid4()),
+                "record_type": MY_ID_RECORD_TYPE}
         data["_id"] = data["self_identity"]
         db.put_record(Record(data))
         logging.debug("set new self-identity value: %r", data["self_identity"])
         return [data["self_identity"]]
 
-def get_all_known_pairings(uri=None,
-        ctx=local_files.DEFAULT_CONTEXT):
+
+def get_all_known_pairings(uri=None, ctx=local_files.DEFAULT_CONTEXT):
     """Info dicts about all pairings, even if marked "unpaired", keyed on
     hostid with another dict as the value."""
     d = {}
@@ -204,8 +218,10 @@
         d[hostid] = v
     return d
 
+
 def _get_management_data(record_type, key, uri=None,
-        ctx=local_files.DEFAULT_CONTEXT):
+                         ctx=local_files.DEFAULT_CONTEXT):
+    """Get the management data from couchdb."""
     db = _get_db("management", uri=uri, ctx=ctx)
     results = db.get_all_records(record_type=record_type)
     values = list()
@@ -226,19 +242,20 @@
     logging.debug("found %d %s records", len(values), key)
     return values
 
+
 def create_database(dst_host, dst_port, dst_name, use_ssl=False,
-        oauth_tokens=None):
+                    oauth_tokens=None):
     """Given parts, create a database."""
     dst_url = mkuri(dst_host, dst_port, use_ssl)
     return server.CouchDatabase(dst_name, dst_url, create=True,
             oauth_tokens=oauth_tokens)
 
-def replicate(source_database, target_database, target_host=None,
-        target_port=None, source_host=None, source_port=None,
-        source_ssl=False, target_ssl=False, source_oauth=None,
-        target_oauth=None, local_uri=None):
+
+def replicate(source_database, target_database,   # pylint: disable-msg=R0914
+              target_host=None, target_port=None, source_host=None,
+              source_port=None, source_ssl=False, target_ssl=False,
+              source_oauth=None, target_oauth=None, local_uri=None):
     """This replication is instant and blocking, and does not persist. """
-
     try:
         if target_host:
             # Target databases must exist before replicating to them.
@@ -251,9 +268,10 @@
         else:
             server.CouchDatabase(target_database, create=True, uri=local_uri)
         logging.debug("db exists, and we're ready to replicate")
-    except:
-        logging.exception("can't create/verify %r %s:%d  oauth=%s",
-                target_database, target_host, target_port, obsfuscate(target_oauth))
+    except:                             # pylint: disable-msg=W0702
+        logging.exception(
+            "can't create/verify %r %s:%d  oauth=%s", target_database,
+            target_host, target_port, obsfuscate(target_oauth))
     if source_host:
         source = mkuri(source_host, source_port, source_ssl, urllib.quote(
             source_database, safe=""))
@@ -285,20 +303,20 @@
                 "asking %r to replicate %s to %s", obsfuscate(local_uri),
                 obsfuscate(source), obsfuscate(target),)
 
-        ### All until python-couchdb gets a Server.replicate() function
+        # All until python-couchdb gets a Server.replicate() function
         local_server = server.OAuthCapableServer(local_uri)
         resp, data = local_server.resource.post(path='/_replicate',
                 content=record)
-
-        logging.debug("replicate result:  %r  %r", obsfuscate(resp), obsfuscate(data))
-        ###
-    except:
+        logging.debug(
+            "replicate result:  %r  %r", obsfuscate(resp), obsfuscate(data))
+    except:                             # pylint: disable-msg=W0702
         logging.exception("can't replicate %r  %r <== %r", source_database,
                 local_uri, obsfuscate(record))
 
+
 def get_pairings(uri=None, ctx=local_files.DEFAULT_CONTEXT):
     """Get a list of paired servers."""
-    db = _get_db("management", create=True, uri=None, ctx=ctx)
+    db = _get_db("management", create=True, uri=uri, ctx=ctx)
 
     design_doc = "paired_servers"
     if not db.view_exists("paired_servers", design_doc):
@@ -306,8 +324,10 @@
             if (doc.record_type == %r && ! doc.unpaired)  // unset or False
                 if (doc.application_annotations &&
                     doc.application_annotations["Ubuntu One"] &&
-                    doc.application_annotations["Ubuntu One"].private_application_annotations &&
-                    doc.application_annotations["Ubuntu One"].private_application_annotations.deleted) {
+                    doc.application_annotations[
+                    "Ubuntu One"].private_application_annotations &&
+                    doc.application_annotations[
+                    "Ubuntu One"].private_application_annotations.deleted) {
                         // don't emit deleted or unpaired items
                     } else {
                         if (doc.server) {
@@ -316,22 +336,25 @@
                             emit(doc.service_name, doc);
                         }
                     }
-                }"""  % (PAIRED_SERVER_RECORD_TYPE,)
+                }""" % (PAIRED_SERVER_RECORD_TYPE,)
         db.add_view("paired_servers", map_js, None, design_doc)
 
     return db.execute_view("paired_servers")
 
+
 def remove_pairing(record_id, is_reconciled, uri=None,
-        ctx=local_files.DEFAULT_CONTEXT):
+                   ctx=local_files.DEFAULT_CONTEXT):
     """Remove a pairing record (or mark it as dead so it can be cleaned up
     properly later)."""
-    db = _get_db("management", create=True, uri=None, ctx=ctx)
+    db = _get_db("management", create=True, uri=uri, ctx=ctx)
     if is_reconciled:
         db.delete_record(record_id)
     else:
-        db.update_fields(record_id, { "unpaired": True })
+        db.update_fields(record_id, {"unpaired": True})
+
 
 def expunge_pairing(host_id, uri=None):
+    """Remove pairing record for host_id."""
     try:
         d = get_all_known_pairings(uri)
         record_id = d[host_id]["record_id"]

=== modified file 'desktopcouch/pair/tests/test_couchdb_io.py'
--- desktopcouch/pair/tests/test_couchdb_io.py	2010-11-02 21:20:35 +0000
+++ desktopcouch/pair/tests/test_couchdb_io.py	2010-11-09 17:39:48 +0000
@@ -1,3 +1,4 @@
+"""Test Couchdb IO"""
 # Copyright 2009 Canonical Ltd.
 #
 # This file is part of desktopcouch.
@@ -31,7 +32,9 @@
 import socket
 URI = None  # use autodiscovery that desktopcouch.tests permits.
 
+
 class TestCouchdbIo(unittest.TestCase):
+    """Test case for Couchdb IO"""
 
     def setUp(self):
         """setup each test"""
@@ -64,22 +67,39 @@
 
     def tearDown(self):
         """tear down each test"""
-        del self.mgt_database._server['management']
-        del self.mgt_database._server['foo']
-
+        del self.mgt_database._server['management']  # pylint: disable-msg=W0212
+        del self.mgt_database._server['foo']         # pylint: disable-msg=W0212
 
     def test_obsfuscation(self):
-        t = {'url': 'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest', 'auth': {'oauth': {'consumer_secret': 'SeCrEtSe', 'token': '3XRjQrWX92TTTJFDTWJJ', 'consumer_key': 'ubuntuone', 'token_secret': 'jBmSeCrEtawkefwklefliwuregqwlkeh347wq87w4fiuq4fyu3q4fiqwu4fqwfiqufM6xjsPwSeCrEt4'}}}
+        """Test the obfuscation of sensitive data."""
+        t = {
+            'url':
+            'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest',
+            'auth': {
+                'oauth': {
+                    'consumer_secret': 'SeCrEtSe',
+                    'token': '3XRjQrWX92TTTJFDTWJJ',
+                    'consumer_key': 'ubuntuone',
+                    'token_secret': 'jBmSeCrEtawkefwklefliwuregqwlkeh347wq87w4f'
+                                    'iuq4fyu3q4fiqwu4fqwfiqufM6xjsPwSeCrEt4'}}}
         cleaned_t = couchdb_io.obsfuscate(t)
-        self.failIf("SeCrEt" in str(cleaned_t), {'url': 'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest', 'auth': {'oauth': {'consumer_secret': 'HiddenHidd', 'token': '3XRjQrWX92TTTJFDTWJJ', 'consumer_key': 'ubuntuone', 'token_secret': 'HiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHi'}}})
-
+        self.failIf(
+            "SeCrEt" in str(cleaned_t), {
+                'url':
+                'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest',
+                'auth': {'oauth': {
+                    'consumer_secret': 'HiddenHidd',
+                    'token': '3XRjQrWX92TTTJFDTWJJ',
+                    'consumer_key': 'ubuntuone',
+                    'token_secret': 'HiddenHiddenHiddenHiddenHiddenHiddenHidden'
+                                    'HiddenHiddenHiddenHiddenHiddenHiddenHi'}}})
         self.assertEqual(couchdb_io.obsfuscate(""), "")
         self.assertEqual(couchdb_io.obsfuscate({}), {})
-        self.assertEqual(couchdb_io.obsfuscate({1:{}}), {1:{}})
-        self.assertEqual(couchdb_io.obsfuscate({1:1}), {1:1})
-
-
-    def test_put_static_paired_service(self):
+        self.assertEqual(couchdb_io.obsfuscate({1: {}}), {1: {}})
+        self.assertEqual(couchdb_io.obsfuscate({1: 1}), {1: 1})
+
+    def test_put_static_paired_service(self):  # pylint: disable-msg=R0201
+        """Test putting a static paired service."""
         service_name = "dummyfortest"
         oauth_data = {
                 "consumer_key": str("abcdef"),
@@ -87,10 +107,14 @@
                 "token": str("opqrst"),
                 "token_secret": str("uvwxyz"),
             }
-        couchdb_io.put_static_paired_service(oauth_data, service_name, uri=URI, ctx=test_environment.test_context)
-        pairings = list(couchdb_io.get_pairings(ctx=test_environment.test_context))
+        couchdb_io.put_static_paired_service(
+            oauth_data, service_name, uri=URI,
+            ctx=test_environment.test_context)
+        couchdb_io.get_pairings(ctx=test_environment.test_context)
+        # Assert something?
 
     def test_put_dynamic_paired_host(self):
+        """Test putting a dynamically paired host."""
         hostname = "host%d" % (os.getpid(),)
         remote_uuid = str(uuid.uuid4())
         oauth_data = {
@@ -107,20 +131,24 @@
         couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
                 uri=URI, ctx=test_environment.test_context)
 
-        pairings = list(couchdb_io.get_pairings(ctx=test_environment.test_context))
-        self.assertEqual(4, len(pairings))  #  3, plus 1 from setUp()
+        pairings = list(couchdb_io.get_pairings(
+            ctx=test_environment.test_context))
+        # 3, plus 1 from setUp()
+        self.assertEqual(4, len(pairings))
         self.assertEqual(pairings[0].value["oauth"], oauth_data)
         self.assertEqual(pairings[0].value["server"], hostname)
         self.assertEqual(pairings[0].value["pairing_identifier"], remote_uuid)
 
         for i, row in enumerate(pairings):
-            couchdb_io.remove_pairing(row.id, i == 1, ctx=test_environment.test_context)
+            couchdb_io.remove_pairing(
+                row.id, i == 1, ctx=test_environment.test_context)
 
-        pairings = list(couchdb_io.get_pairings(ctx=test_environment.test_context))
+        pairings = list(
+            couchdb_io.get_pairings(ctx=test_environment.test_context))
         self.assertEqual(0, len(pairings))
 
-
     def test_get_database_names_replicatable_bad_server(self):
+        """Test get database names from the wrong url."""
         hostname = "test.desktopcouch.example.com"
         try:
             socket.gethostbyname(hostname)
@@ -136,39 +164,26 @@
             pass
 
     def test_get_database_names_replicatable(self):
-        names = couchdb_io.get_database_names_replicatable(uri=URI, ctx=test_environment.test_context)
+        """Test get the names of replicatable databases."""
+        names = couchdb_io.get_database_names_replicatable(
+            uri=URI, ctx=test_environment.test_context)
         self.assertFalse('management' in names)
         self.assertTrue('foo' in names)
         self.assertFalse('bar' in names, names)  # is excluded
 
     def test_get_my_host_unique_id(self):
-        got = couchdb_io.get_my_host_unique_id(uri=URI, ctx=test_environment.test_context)
-        again = couchdb_io.get_my_host_unique_id(uri=URI, ctx=test_environment.test_context)
+        """Test get the unique id of the host record."""
+        got = couchdb_io.get_my_host_unique_id(
+            uri=URI, ctx=test_environment.test_context)
+        again = couchdb_io.get_my_host_unique_id(
+            uri=URI, ctx=test_environment.test_context)
         self.assertEquals(len(got), 1)
         self.assertEquals(got, again)
 
     def test_mkuri(self):
+        """Test creating a URI."""
         uri = couchdb_io.mkuri(
             'fnord.org', 55241, has_ssl=True, path='a/b/c',
             auth_pair=('f o o', 'b=a=r'))
         self.assertEquals(
             'https://f%20o%20o:[e-mail]:55241/a/b/c', uri)
-
-    def Xtest_replication_good(self):
-        pass
-
-    def Xtest_replication_no_oauth_remote(self):
-        pass
-
-    def Xtest_replication_bad_oauth_remote(self):
-        pass
-
-    def Xtest_replication_no_oauth_local(self):
-        pass
-
-    def Xtest_replication_bad_oauth_local(self):
-        pass
-
-
-if __name__ == "__main__":
-    unittest.main()