db: Add a Map object

For creating a table that maps one value into another

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>

db/objects.py

@@ -76,3 +76,45 @@ class Table(GObject.GObject):
 
     @GObject.Signal(arg_types=(Row,))
     def row_deleted(self, row): pass
+
+
+class Map(GObject.GObject):
+    def __init__(self, name, lhs, rhs):
+        GObject.GObject.__init__(self)
+        self.map_name = name
+        self.map_lhs  = lhs
+        self.map_rhs  = rhs
+
+        self.do_create()
+
+    def do_create(self): raise NotImplementedError
+    def do_insert(self, lhs, rhs): raise NotImplementedError
+    def do_delete(self, lhs, rhs): raise NotImplementedError
+    def do_lookup_rhs(self, lhs): raise NotImplementedError
+    def do_lookup_lhs(self, rhs): raise NotImplementedError
+
+    def insert(self, lhs, rhs):
+        cursor = self.do_insert(lhs, rhs)
+        self.emit("row-inserted", lhs, rhs)
+
+    def delete(self, lhs, rhs):
+        self.emit("row-deleted", lhs, rhs)
+        self.do_delete(lhs, rhs)
+
+    def lookup_rhs(self, lhs):
+        cursor = self.do_lookup_rhs(lhs)
+        return [ self.map_rhs(row[0]) for row in cursor.fetchall() ]
+
+    def lookup_lhs(self, rhs):
+        cursor = self.do_lookup_lhs(rhs)
+        return [ self.map_lhs(row[0]) for row in cursor.fetchall() ]
+
+    def reset(self):
+        execute(f"DROP TABLE {self.map_name}")
+        self.do_create()
+
+    @GObject.Signal(arg_types=(Row,Row))
+    def row_inserted(self, lhs, rhs): pass
+
+    @GObject.Signal(arg_types=(Row,Row))
+    def row_deleted(self, lhs, rhs): pass

db/test_objects.py

@@ -6,6 +6,7 @@ from . import objects
 class FakeCursor:
     def __init__(self, lastrowid): self.lastrowid = lastrowid
     def fetchone(self): return (self.lastrowid,) if self.lastrowid else None
+    def fetchall(self): return [ (2,), (3,) ] if self.lastrowid else None
 
 
 class TestRow(unittest.TestCase):
@@ -58,6 +59,7 @@ class TestTag(unittest.TestCase):
         self.assertFalse(a > b)
         self.assertFalse(b < a)
 
+
 class FakeTable(objects.Table):
     def do_create(self): self.create_called = True
     def do_insert(self, insert, args): return FakeCursor(1)
@@ -110,3 +112,53 @@ class TestTable(unittest.TestCase):
         self.row_inserted = None
         self.assertEqual(table.find("find", "arguments"), objects.Row(1))
         self.assertEqual(self.row_inserted, objects.Row(1))
+
+
+class FakeMap(objects.Map):
+    def do_create(self): self.create_called = True
+    def do_insert(self, lhs, rhs): return FakeCursor(1)
+    def do_delete(self, lhs, rhs): return FakeCursor(1)
+    def do_lookup_rhs(self, lhs): return FakeCursor(1)
+    def do_lookup_lhs(self, lhs): return FakeCursor(1)
+
+class TestMap(unittest.TestCase):
+    def on_row_inserted(self, table, lhs, rhs):
+        self.row_inserted = (lhs, rhs)
+
+    def on_row_deleted(self, table, lhs, rhs):
+        self.row_deleted = (lhs, rhs)
+
+    def test_map_errors(self):
+        with self.assertRaises(NotImplementedError):
+            map = objects.Map("test_map", objects.Tag, objects.Row)
+        with self.assertRaises(NotImplementedError):
+            objects.Map.do_insert(None, True, False)
+        with self.assertRaises(NotImplementedError):
+            objects.Map.do_delete(None, True, False)
+        with self.assertRaises(NotImplementedError):
+            objects.Map.do_lookup_rhs(None, None)
+        with self.assertRaises(NotImplementedError):
+            objects.Map.do_lookup_lhs(None, None)
+
+    def test_map(self):
+        map = FakeMap("test_map", objects.Row, objects.Row)
+        map.connect("row-inserted", self.on_row_inserted)
+        map.connect("row-deleted", self.on_row_deleted)
+
+        self.assertIsInstance(map, GObject.GObject)
+        self.assertEqual(map.map_name, "test_map")
+        self.assertEqual(map.map_lhs, objects.Row)
+        self.assertEqual(map.map_rhs, objects.Row)
+        self.assertTrue(map.create_called)
+
+        map.insert(objects.Row(1), objects.Row(2))
+        self.assertEqual(self.row_inserted, (objects.Row(1), objects.Row(2)))
+
+        map.delete(objects.Row(1), objects.Row(2))
+        self.assertEqual(self.row_deleted, (objects.Row(1), objects.Row(2)))
+
+        rhs_list = map.lookup_rhs(objects.Row(1))
+        self.assertEqual(rhs_list, [ objects.Row(2), objects.Row(3) ])
+
+        lhs_list = map.lookup_lhs(objects.Row(2))
+        self.assertEqual(lhs_list, [ objects.Row(2), objects.Row(3) ])