test: fix tests to use in-memory sqlite database.

Add a proxy for activity hooks to initialize the database.

pomito/hooks/activity.py

@@ -1,41 +1,41 @@
 # -*- coding: utf-8 -*-
-# Pomito - Pomodoro timer on steroids
-# Maintains a log of all user activities
+# Pomito - Pomodoro timer on steroids.
+"""Activity module maintains a log of all user activities."""
 
 
 from datetime import datetime
 
 from pomito.hooks import Hook
 
-from peewee import Model, DateTimeField, CharField, TextField
+from peewee import Model, Proxy, DateTimeField, CharField, TextField
+
+database_proxy = Proxy()
 
 
 class ActivityHook(Hook):
-    """Activity hook listens to session and interruption events. Keeps a record
-    of pomodoro statistics.
+    """Activity hook listens to session and interruption events.
+
+    Keeps a record of pomodoro statistics.
     """
-    activity_db = None
 
     def __init__(self, service):
         self._service = service
-        return
 
     def initialize(self):
-        ActivityHook.activity_db = self._service.get_db()
+        database_proxy.initialize(self._service.get_db())
         ActivityModel.create_table(True)
 
         self._service.signal_session_stopped\
             .connect(self.handle_session_stopped)
         self._service.signal_break_stopped.connect(self.handle_break_stopped)
         self._service.signal_interruption_stopped\
             .connect(self.handle_interruption_stopped)
-        return
 
     def log(self, category, data):
-        activity = ActivityModel(timestamp=datetime.now(), category=category,
+        activity = ActivityModel(timestamp=datetime.now(),
+                                 category=category,
                                  data=data)
         activity.save()
-        return
 
     def close(self):
         self._service.signal_session_stopped\
@@ -44,7 +44,6 @@ def close(self):
             .disconnect(self.handle_break_stopped)
         self._service.signal_interruption_stopped\
             .disconnect(self.handle_interruption_stopped)
-        return
 
     ####
     # Pomodoro service signal handlers
@@ -75,4 +74,4 @@ class ActivityModel(Model):
     data = TextField()
 
     class Meta:
-        database = ActivityHook.activity_db
+        database = database_proxy

pomito/main.py

@@ -157,7 +157,7 @@ def initialize(self):
         if not os.path.exists(database_path) and not self._database:
             self._database = SqliteDatabase(None)
             self._database.init(database_path)
-            self._database.connect()
+        self._database.connect()
 
         # Initialize the plugins
         self.ui_plugin.initialize()
@@ -189,6 +189,8 @@ def exit(self):
             hook.close()
         #self._validate_state()
         #self._parser.write()
+        if self._database is not None:
+            self._database.close()
         return
 
     def get_db(self):

pomito/test.py

@@ -11,6 +11,7 @@
 from pomito.plugins.ui import UIPlugin
 from pomito.plugins.task import TaskPlugin
 
+
 class FakeTimer(object):
     """A synchronous fake timer."""
 
@@ -71,10 +72,11 @@ def initialize(self):
     def get_tasks(self):
         return self.task_list
 
+
 class PomitoTestFactory(object):
     """Creates fake pomodoro framework instances for testing."""
     config_file = None
-    database = SqliteDatabase(':memory')
+    database = SqliteDatabase(':memory:')
     message_dispatcher = FakeMessageDispatcher()
 
     def create_fake_service(self):

tests/hooks/test_activity.py

@@ -2,16 +2,15 @@
 # Tests for the Activity hook
 
 import unittest
-from unittest.mock import Mock, patch
-
-from peewee import SqliteDatabase
+from peewee import Using
 
 from pomito.hooks.activity import ActivityHook, ActivityModel
 from pomito.test import PomitoTestFactory
 
 from pyfakefs import fake_filesystem
 import sure
 
+
 class ActivityHookTests(unittest.TestCase):
     def setUp(self):
         test_factory = PomitoTestFactory()
@@ -22,23 +21,19 @@ def setUp(self):
         test_factory.create_patch(self, 'os.path', os_module.path)
         test_factory.create_patch(self, 'os.makedirs', os_module.makedirs)
 
-        ActivityModel.drop_table(True)
-
         self.pomodoro_service = test_factory.create_fake_service()
         self.activityhook = ActivityHook(self.pomodoro_service)
         self.activityhook.initialize()
+        self.database = self.pomodoro_service.get_db()
 
     def tearDown(self):
         ActivityModel.drop_table(True)
         self.activityhook.close()
 
     def test_initialize_sets_activity_db(self):
-        activityhook = ActivityHook(self.pomodoro_service)
+        count = ActivityModel.select().count()
 
-        activityhook.initialize()
-
-        ActivityHook.activity_db.shouldnt.be.equal(None)
-        activityhook.close()
+        count.should.be.equal(0)
 
     def test_initialize_creates_signal_handlers(self):
         activityhook = ActivityHook(self.pomodoro_service)
@@ -72,40 +67,36 @@ def test_log_handles_session_stop_event(self):
         self.pomodoro_service.start_session(test_task)
         self.pomodoro_service.stop_session()
 
-        ActivityModel.select().where(ActivityModel.category ==\
-                'session').count().should.be.equal(1)
+        activities = ActivityModel.get(ActivityModel.category == 'session')
+        activities.should.not_be.none
 
     def test_log_handles_break_stop_event(self):
         self.pomodoro_service.start_break()
         self.pomodoro_service.stop_break()
 
         self._dump_activity_model()
-        ActivityModel.select().where(ActivityModel.category ==\
-                'break').count().should.be.equal(1)
+        activities = ActivityModel.get(ActivityModel.category == 'break')
+        activities.should.not_be.none
 
     def test_log_handles_interruption_stop_event(self):
         self.pomodoro_service.start_interruption(None, False, False)
         self.pomodoro_service.stop_interruption()
 
-        ActivityModel.select().where(ActivityModel.category ==\
-                'interruption').count().should.be.equal(1)
-        pass
+        act = ActivityModel.get(ActivityModel.category == 'interruption')
+        act.shouldnt.be.none
 
     def _create_task(self, uid, description):
         from pomito.task import Task
-        return Task(uid=uid, description=description, estimate=0, actual=0,
-                tags=None)
-
-    def _get_latest_entry(self):
-        pass
+        return Task(uid=uid, description=description,
+                    estimate=0, actual=0,
+                    tags=None)
 
     def _dump_activity_model(self):
         for activity in ActivityModel.select():
             print("{0};{1};{2}".format(activity.timestamp, activity.category,
-                activity.data))
+                                       activity.data))
 
     def _count_receivers(self, signal):
         from blinker.base import ANY
 
         return sum(1 for r in signal.receivers_for(ANY))
-