Merge lp:~brendan-donegan/ubuntu-autopilot-tests/contacts_dbus into lp:ubuntu-autopilot-tests/ubuntu-experience-tests

[Brendan Donegan]

Put ContactsDbusService in helpers.py, so that all UX tests can use it.

[lp:~brendan-donegan/ubuntu-autopilot-tests/contacts_dbus]

22. By Brendan Donegan on 2014-06-19

Fix call to get_vcard_uid

[Richard Huddie (rhuddie)]

review: Approve

[Dan Chapman  (dpniel)]

Looks good to me

review: Approve

=== modified file 'ubuntu_experience_tests/helpers.py'
--- ubuntu_experience_tests/helpers.py	2014-05-20 12:35:59 +0000
+++ ubuntu_experience_tests/helpers.py	2014-06-19 09:50:40 +0000
@@ -2,6 +2,7 @@
 #
 # Copyright 2014 Canonical Ltd.
 # Author: Omer Akram <[e-mail]>
+#         Brendan Donegan <[e-mail]>
 #
 # This file is part of ubuntu-experience-tests.
 #
@@ -19,6 +20,7 @@
 
 """Helper functions to make writing Inter-app integration tests easier."""
 
+import dbus
 import subprocess
 import time
 
@@ -26,6 +28,71 @@
 from ubuntuuitoolkit import emulators as toolkit_emulators
 from unity8.shell.emulators import UnityEmulatorBase
 
+DBUS_IFACE_ADD_BOOK = 'com.canonical.pim.AddressBook'
+DBUS_IFACE_ADD_BOOKVIEW = 'com.canonical.pim.AddressBookView'
+
+
+class ContacstDbusService(object):
+    def __init__(self):
+        self.bus = None
+        self.addr = None
+        self.addr_iface = None
+        self.bus = dbus.SessionBus()
+        self.addr = self.bus.get_object(
+            'com.canonical.pim', '/com/canonical/pim/AddressBook')
+        self.addr_iface = dbus.Interface(
+            self.addr, dbus_interface=DBUS_IFACE_ADD_BOOK)
+
+    def query_contacts(self, fields='', query='', sources=[]):
+        view_path = self.addr_iface.query(fields, query, [])
+        view = self.bus.get_object(
+            'com.canonical.pim', view_path)
+        view_iface = dbus.Interface(
+            view, dbus_interface=DBUS_IFACE_ADD_BOOKVIEW)
+        contacts = view_iface.contactsDetails([], 0, -1)
+        view.close()
+        return contacts
+
+    def update_contact(self, vcard):
+        return self.addr_iface.updateContacts([vcard])
+
+    def create_contact(self, vcard):
+        return self.addr_iface.createContact(vcard, "")
+
+    def delete_contact(self, ids):
+        return self.addr_iface.removeContacts(ids)
+
+    def get_vcard_uid(self, vcard):
+        """
+        Return the UID of the provided vcard
+        :param vcard: vcard formatted data
+        :return: uid value from vcard, else None
+        """
+        uid_pattern = re.compile(r'''
+        ^          # start of string
+        .*         # any characters
+        UID:(.*?)  # UID value, saved to group, non-greedy
+        \r\n       # new line
+        .*         # anything else
+        $          # end of string
+        ''', re.VERBOSE | re.IGNORECASE | re.DOTALL)
+        uid = None
+        match = uid_pattern.match(vcard)
+        if match:
+            uid = match.group(1)
+        return uid
+
+    def delete_all_contacts(self):
+        """
+        Delete all contacts using the dbus service
+        """
+        uids = []
+        contacts = self.query_contacts()
+        for contact in contacts:
+            uid = self.get_vcard_uid(contact)
+            if uid:
+                uids.append(uid)
+        self.delete_contact(uids)
 
 def get_proxy_object_by_process_name(proc_name):
     """Helper function to return proxy object of a process