Add foreign key support for sqlite

remoteappmanager/cli/remoteappdb/__main__.py

@@ -170,12 +170,8 @@ def remove(ctx, user):
             orm_user = session.query(orm.User).filter(
                 orm.User.name == user).one()
 
-            # Unfortunately sqlite does not support cascading, so we need to
-            # perform the cleanup of the accounting manually.
-            session.query(orm.Accounting).filter(
-                orm.Accounting.user == orm_user).delete()
-
             session.delete(orm_user)
+
     except sqlalchemy.orm.exc.NoResultFound:
         print_error("Could not find user {}".format(user))
 
@@ -281,8 +277,6 @@ def remove(ctx, image):
             app = session.query(orm.Application).filter(
                 orm.Application.image == image).one()
 
-            session.query(orm.Accounting).filter(
-                orm.Accounting.application_id == app.id).delete()
             session.delete(app)
     except sqlalchemy.orm.exc.NoResultFound:
         print_error("Could not find application for image {}".format(image))

remoteappmanager/db/orm.py

@@ -1,8 +1,10 @@
 import contextlib
+import sqlite3
 
 from sqlalchemy import (
     Column, Integer, Boolean, Unicode, ForeignKey, create_engine, Enum,
-    literal)
+    literal, event)
+from sqlalchemy.engine import Engine
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker, relationship
@@ -72,16 +74,17 @@ class Accounting(Base):
     __tablename__ = "accounting"
 
     user_id = Column(Integer,
-                     ForeignKey("user.id"),
+                     ForeignKey("user.id", ondelete="CASCADE"),
                      primary_key=True)
 
     application_id = Column(Integer,
-                            ForeignKey("application.id"),
+                            ForeignKey("application.id", ondelete="CASCADE"),
                             primary_key=True)
 
-    application_policy_id = Column(Integer,
-                                   ForeignKey("application_policy.id"),
-                                   primary_key=True)
+    application_policy_id = Column(
+        Integer,
+        ForeignKey("application_policy.id", ondelete="CASCADE"),
+        primary_key=True)
 
     user = relationship("User")
 
@@ -90,6 +93,14 @@ class Accounting(Base):
     application_policy = relationship("ApplicationPolicy")
 
 
+@event.listens_for(Engine, "connect")
+def set_sqlite_pragma(dbapi_connection, connection_record):
+    if type(dbapi_connection) is sqlite3.Connection:
+        cursor = dbapi_connection.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+
 class Database(LoggingMixin):
     def __init__(self, url, **kwargs):
         """Initialises a database connection to a given database url.
@@ -107,6 +118,7 @@ def __init__(self, url, **kwargs):
 
         self.log.info("Creating session to db: {}".format(self.url))
         self.engine = create_engine(self.url, **kwargs)
+
         try:
             self.session_class = sessionmaker(bind=self.engine)
         except OperationalError:

tests/cli/test_remoteappdb.py

@@ -132,3 +132,57 @@ def test_app_revoke(self):
         self.assertEqual(len(out.split('\n')), 2)
         self.assertIn("frobniz", out)
         self.assertIn("froble", out)
+
+    def test_delete_user_cascade(self):
+        """ Test if deleting user cascade to deleting accounting rows
+        """
+        # Given user is created with two accountings (application, policy)
+        self._remoteappdb("app create myapp --no-verify")
+        out = self._remoteappdb("user create user")
+        self.assertEqual(out, "1\n")
+
+        self._remoteappdb("app grant myapp user")
+        self._remoteappdb("app grant myapp user --allow-view")
+
+        out = self._remoteappdb("user list --show-apps --no-decoration")
+        self.assertEqual(len(out.split('\n')), 3)
+
+        # When the user is deleted, the accounting rows should be deleted
+        # So when you add the user back later, those accountings should
+        # not exist (This test relies on the fact that there is only one
+        # user and so has the same id as before)
+        self._remoteappdb("user remove user")
+        out = self._remoteappdb("user create user")
+        self.assertEqual(out, "1\n")
+
+        out = self._remoteappdb("user list --show-apps --no-decoration")
+        self.assertEqual(len(out.split('\n')), 2)
+        self.assertNotIn("myapp", out)
+
+    def test_delete_application_cascade(self):
+        """ Test if deleting application cascade to deleting accounting rows
+        """
+        # Given user is created with two accountings (application, policy)
+        out = self._remoteappdb("app create myapp --no-verify")
+        self.assertEqual(out, "1\n")
+        out = self._remoteappdb("user create user")
+        self.assertEqual(out, "1\n")
+
+        self._remoteappdb("app grant myapp user")
+        self._remoteappdb("app grant myapp user --allow-view")
+
+        out = self._remoteappdb("user list --show-apps --no-decoration")
+        self.assertEqual(len(out.split('\n')), 3)
+
+        # When the application is deleted, the associated accounting rows
+        # should be deleted.  So when you add the application back later,
+        # those accountings should not exist
+        # (This test relies on the fact that there is only one app and so
+        # the application row has the same id as before)
+        self._remoteappdb("app remove myapp")
+        out = self._remoteappdb("app create myapp --no-verify")
+        self.assertEqual(out, "1\n")
+
+        out = self._remoteappdb("user list --show-apps --no-decoration")
+        self.assertEqual(len(out.split('\n')), 2)
+        self.assertNotIn("myapp", out)