Merge lp:~mfisch/ubuntu-accomplishments-daemon/ubuntu-accomplishments-daemon-unittest-accomplish into lp:ubuntu-accomplishments-daemon

[Matt Fischer]

More tests. For api.py, I cleaned up more whitespace. I also moved the trophy creation code to a separate function as it seemed to be an independent piece of code.

=== modified file 'accomplishments/daemon/api.py'
--- accomplishments/daemon/api.py	2012-07-17 17:00:06 +0000
+++ accomplishments/daemon/api.py	2012-07-17 17:07:34 +0000
@@ -259,7 +259,8 @@
         
         log.msg("--- Starting Running Scripts - %d items on the queue ---" % (queuesize))
         timestart = time.time()
-        self.parent.service.scriptrunner_start()
+        if not self.parent.test_mode:
+            self.parent.service.scriptrunner_start()
 
         while queuesize > 0:
             accomID = self.parent.scripts_queue.popleft()
@@ -309,7 +310,8 @@
 
         log.msg(
             "--- Emptied the scripts queue in %.2f seconds---" % timefinal)
-        self.parent.service.scriptrunner_finish()
+        if not self.parent.test_mode:
+            self.parent.service.scriptrunner_finish()
 
         self.scripts_state = NOT_RUNNING
 
@@ -1246,7 +1248,7 @@
             {'needs-signing': 'true', 'date-accomplished': '1990-04-12 02:22', 'needs-information': 'launchpad-email', 'version': '0.2', '__name__': 'trophy', 'launchpad-email': '[e-mail]', 'id': 'ubuntu-community/registered-on-launchpad'}
         """
         if not self.get_acc_is_completed(accomID):
-            return
+            return None
         else:
             cfg = ConfigParser.RawConfigParser()
             cfg.read(self.get_trophy_path(accomID))
@@ -1411,38 +1413,54 @@
     def get_published_status(self):
         """Detect if we are currently publishing online or not. Returns
         True if we are or False if we are not. """
-        
+
         trophydir = self.get_config_value("config", "trophypath")
         print trophydir
         if os.path.exists(os.path.join(trophydir, "WEBVIEW")):
             return True
         else:
             return False
-    
+
     def accomplish(self,accomID):
         log.msg("Accomplishing: %s" % accomID)
         if not self.get_acc_exists(accomID):
             log.msg("There is no such accomplishment.")
             return False #failure
-            
+
         # Check if is hasn't been already completed
         if self.get_acc_is_completed(accomID):
             log.msg("Not accomplishing " + accomID + ", it has already been completed.")
             return True #success
-            
+
         # Check if this accomplishment is unlocked
         if not self.get_acc_is_unlocked(accomID):
             log.msg("This accomplishment cannot be completed; it's locked.")
             return False
-            
+
         coll = self._coll_from_accomID(accomID)
         accdata = self.get_acc_data(accomID)
-        
+
         # Prepare extra-info
         needsinformation = self.get_acc_needs_info(accomID)
         for i in needsinformation:
             accdata[i] = self.get_extra_information(coll,i)[0][i]
-            
+
+        # Create .trophy file
+        self._create_trophy_file(accdata, accomID)
+
+        if not self.get_acc_needs_signing(accomID):
+            # The accomplishment does not need signing!
+            if not self.test_mode:
+                self.service.trophy_received(accomID)
+            self._display_accomplished_bubble(accomID)
+            self._display_unlocked_bubble(accomID)
+            # Mark as completed and get list of new opportunities
+            just_unlocked = self._mark_as_completed(accomID)
+            self.run_scripts(just_unlocked)
+
+        return True
+
+    def _create_trophy_file(self, accdata, accomID):
         # Create .trophy file
         cp = ConfigParser.RawConfigParser()
         cp.add_section("trophy")
@@ -1465,17 +1483,6 @@
         fp = open(trophypath, "w")
         cp.write(fp)
         fp.close()
-        
-        if not self.get_acc_needs_signing(accomID):
-            # The accomplishment does not need signing!
-            self.service.trophy_received(accomID)
-            self._display_accomplished_bubble(accomID)
-            self._display_unlocked_bubble(accomID)
-            # Mark as completed and get list of new opportunities
-            just_unlocked = self._mark_as_completed(accomID)
-            self.run_scripts(just_unlocked)
-            
-        return True
 
     def set_daemon_session_start(self,value):
         log.msg(value)
@@ -1545,10 +1552,10 @@
                         return False
         else:
             return False
-    
+
     def _coll_from_accomID(self,accomID):
         return accomID.split("/")[0]
-    
+
     def _display_accomplished_bubble(self,accomID):
         if self.show_notifications == True and pynotify and (
             pynotify.is_initted() or pynotify.init("icon-summary-body")):
@@ -1558,7 +1565,7 @@
                 self.get_acc_icon_path(accomID) )
             n.set_hint_string('append', 'allowed')
             n.show()
-    
+
     def _display_unlocked_bubble(self,accomID):
         unlocked = len(self.list_depending_on(accomID))
         if unlocked is not 0:

=== modified file 'accomplishments/daemon/tests/tests.py'
--- accomplishments/daemon/tests/tests.py	2012-07-16 18:34:08 +0000
+++ accomplishments/daemon/tests/tests.py	2012-07-17 17:07:34 +0000
@@ -7,19 +7,13 @@
 import shutil
 import subprocess
 import ConfigParser
+import datetime
 
 sys.path.insert(0, os.path.join(os.path.split(__file__)[0], "../../.."))
 from accomplishments.daemon import app, api
 
 # future tests:
 # create extra information files - marked for removal in the code
-# get_acc_date_completed - needs accomplish() to work to be useful
-# get trophy data
-# list trophies
-# list opportunities
-# list depending on
-# list unlocked
-# list unlocked not completed
 # run scripts/runscript
 # get published status
 # invalidate extra information
@@ -140,8 +134,131 @@
         del os.environ['ACCOMPLISHMENTS_ROOT_DIR']
         shutil.rmtree(self.td)
 
-    def test_handle_duplicate_accomplishments(self):
-        return
+    def test_get_acc_date_completed(self):
+        self.util_remove_all_accomps(self.accomp_dir)
+        self.util_copy_accomp(self.accomp_dir, "first")
+        self.util_copy_accomp(self.accomp_dir, "second")
+        self.util_copy_accomp(self.accomp_dir, "third")
+        self.util_copy_extrainfo(self.extrainfo_dir, "info")
+        self.util_copy_extrainfo(self.extrainfo_dir, "info2")
+        a = api.Accomplishments(None, None, True)
+        a.write_extra_information_file("info", "whatever")
+        a.write_extra_information_file("info2", "whatever2")
+
+        self.assertTrue(a.accomplish("%s/first" % self.ACCOMP_SET))
+        self.assertTrue(a.accomplish("%s/second" % self.ACCOMP_SET))
+        self.assertTrue(a.accomplish("%s/third" % self.ACCOMP_SET))
+
+        trophies = a.list_trophies()
+        # since "second" requires signing, it shouldn't be listed
+        self.assertEqual(len(trophies), 2)
+
+        d1 = a.get_acc_date_completed("%s/first" % self.ACCOMP_SET)
+        self.assertTrue(isinstance(d1, basestring))
+        dt1 = datetime.datetime.strptime(d1, "%Y-%m-%d %H:%M")
+        self.assertTrue(dt1 is not None)
+
+        d3 = a.get_acc_date_completed("%s/third" % self.ACCOMP_SET)
+        self.assertTrue(isinstance(d3, basestring))
+        dt3 = datetime.datetime.strptime(d3, "%Y-%m-%d %H:%M")
+        self.assertTrue(dt3 is not None)
+
+    # this tests:
+    # accomplish()
+    # list_opportunities
+    # list_trophies
+    # list_unlocked
+    # list_unlocked_not_completed
+    # get_trophy_data
+    def test_accomplish(self):
+        self.util_remove_all_accomps(self.accomp_dir)
+        self.util_copy_accomp(self.accomp_dir, "first")
+        self.util_copy_accomp(self.accomp_dir, "second")
+        self.util_copy_accomp(self.accomp_dir, "third")
+        a = api.Accomplishments(None, None, True)
+
+        # before accomplishing
+        opps = a.list_opportunitues()
+        self.assertEqual(len(opps), 3)
+        for accom in opps:
+            self.assertTrue(accom in ["%s/first" % self.ACCOMP_SET,
+                "%s/second" % self.ACCOMP_SET, "%s/third" % self.ACCOMP_SET])
+
+        unlocked = a.list_unlocked()
+        self.assertEqual(len(unlocked), 2)
+        for accom in unlocked:
+            self.assertTrue(accom in ["%s/first" % self.ACCOMP_SET,
+                "%s/third" % self.ACCOMP_SET])
+
+        unlocked_nc = a.list_unlocked_not_completed()
+        self.assertEqual(len(unlocked_nc), 2)
+        for accom in unlocked_nc:
+            self.assertTrue(accom in ["%s/first" % self.ACCOMP_SET,
+                "%s/third" % self.ACCOMP_SET])
+
+        trophies = a.list_trophies()
+        self.assertEqual(len(trophies), 0)
+
+        self.assertTrue(a.get_trophy_data("%s/first" % self.ACCOMP_SET) is None)
+        self.assertTrue(a.get_trophy_data("%s/second" % self.ACCOMP_SET)
+            is None)
+        self.assertTrue(a.get_trophy_data("%s/third" % self.ACCOMP_SET) is None)
+
+        # now let's accomplish something, it should fail without extra info
+        self.assertRaises(KeyError, a.accomplish, "%s/first" % self.ACCOMP_SET)
+
+        # this time it will work
+        self.util_copy_extrainfo(self.extrainfo_dir, "info")
+        self.util_copy_extrainfo(self.extrainfo_dir, "info2")
+        a.reload_accom_database()
+        self.assertTrue(a.accomplish("%s/first" % self.ACCOMP_SET))
+
+        opps = a.list_opportunitues()
+        self.assertEqual(len(opps), 2)
+        for accom in opps:
+            self.assertTrue(accom in ["%s/second" % self.ACCOMP_SET,
+                "%s/third" % self.ACCOMP_SET])
+
+        unlocked = a.list_unlocked()
+        self.assertEqual(len(unlocked), 3)
+        for accom in unlocked:
+            self.assertTrue(accom in ["%s/first" % self.ACCOMP_SET,
+                "%s/second" % self.ACCOMP_SET,
+                "%s/third" % self.ACCOMP_SET])
+
+        unlocked_nc = a.list_unlocked_not_completed()
+        self.assertEqual(len(unlocked_nc), 2)
+        for accom in unlocked_nc:
+            self.assertTrue(accom in ["%s/second" % self.ACCOMP_SET,
+                "%s/third" % self.ACCOMP_SET])
+
+        trophies = a.list_trophies()
+        self.assertEqual(len(trophies), 1)
+        for accom in trophies:
+            self.assertTrue(accom in ["%s/first" % self.ACCOMP_SET])
+
+        td = a.get_trophy_data("%s/first" % self.ACCOMP_SET)
+        self.assertTrue(isinstance(td, dict))
+
+        self.assertTrue(td['date-accomplished'] is not None)
+        self.assertTrue(td['version'] is not None)
+        self.assertTrue(td['__name__'] == "trophy")
+        self.assertTrue(td['id'] == "%s/first" % self.ACCOMP_SET)
+        self.assertTrue(td['needs-information'] is not None)
+
+    def test_list_depending_on(self):
+        self.util_remove_all_accomps(self.accomp_dir)
+        self.util_copy_accomp(self.accomp_dir, "first")
+        self.util_copy_accomp(self.accomp_dir, "second")
+        self.util_copy_accomp(self.accomp_dir, "third")
+        a = api.Accomplishments(None, None, True)
+
+        self.assertEquals(len(a.list_depending_on("%s/first" %
+            self.ACCOMP_SET)), 1)
+        self.assertEquals(len(a.list_depending_on("%s/second" %
+            self.ACCOMP_SET)), 0)
+        self.assertEquals(len(a.list_depending_on("%s/third" %
+            self.ACCOMP_SET)), 0)
 
     # tests all the get_acc_* functions, except for:
     # get_acc_icon
@@ -358,15 +475,6 @@
 
         self.util_remove_all_accomps(self.accomp_dir)
 
-    @unittest.skip("need the daemon running")
-    def test_accomplish(self):
-        a = api.Accomplishments(None, None, True)
-        self.assertEqual(len(a.list_trophies()), 0)
-        a.accomplish("%s/first" % self.ACCOMP_SET)
-        trophies = a.list_trophies()
-        self.assertEqual(len(trophies), 1)
-        self.assertEqual(trophies[0]["title"], "My First Accomplishment")
-
     def test_missing_about_file(self):
         os.remove(os.path.join(self.accomp_root, "ABOUT"))
         self.assertRaises(LookupError, api.Accomplishments, None, None, True)