tests/dummy.py

   1 import datasette_connectors as dc
   2
   3
   4 class DummyConnector(dc.Connector):
   5     connector_type = 'dummy'
   6
   7     def table_names(self):
   8         return ['table1', 'table2']
   9
  10     def hidden_table_names(self):
  11         return []
  12
  13     def detect_spatialite(self):
  14         return False
  15
  16     def view_names(self):
  17         return []
  18
  19     def table_count(self, table_name):
  20         return 2
  21
  22     def table_info(self, table_name):
  23         return [
  24             {
  25                 'idx': 0,
  26                 'name': 'c1',
  27                 'primary_key': False,
  28             },
  29             {
  30                 'idx': 0,
  31                 'name': 'c2',
  32                 'primary_key': False,
  33             },
  34             {
  35                 'idx': 0,
  36                 'name': 'c3',
  37                 'primary_key': False,
  38             },
  39         ]
  40
  41     def detect_fts(self, table_name):
  42         return False
  43
  44     def foreign_keys(self, table_name):
  45         return []
  46
  47     def table_exists(self, table_name):
  48         return table_name in ['table1', 'table2']
  49
  50     def table_definition(self, table_type, table_name):
  51         return 'CREATE TABLE ' + table_name + ' (c1, c2, c3)'
  52
  53     def indices_definition(self, table_name):
  54         return []
  55
  56     def execute(
  57         self,
  58         sql,
  59         params=None,
  60         truncate=False,
  61         custom_time_limit=None,
  62         page_size=None,
  63         log_sql_errors=True,
  64     ):
  65         results = []
  66         truncated = False
  67         description = ()
  68
  69         if sql == 'select c1 from table1':
  70             results = [
  71                 {'c1': 10},
  72                 {'c1': 20},
  73             ]
  74             description = (('c1',),)
  75         elif sql == 'select c1, c2, c3 from table2 limit 51':
  76             results = [
  77                 {'c1': 100, 'c2': 120, 'c3': 130},
  78                 {'c1': 200, 'c2': 220, 'c3': 230},
  79             ]
  80             description = (('c1',), ('c2',), ('c3',))
  81         elif sql == "select * from (select c1, c2, c3 from table2 ) limit 0":
  82             pass
  83         elif sql == "select c1, count(*) as n from ( select c1, c2, c3 from table2 ) where c1 is not null group by c1 limit 31":
  84             results = [
  85                 {'c1': 100, 'n': 1},
  86                 {'c1': 200, 'n': 1},
  87             ]
  88             description = (('c1',), ('n',))
  89         elif sql == "select c2, count(*) as n from ( select c1, c2, c3 from table2 ) where c2 is not null group by c2 limit 31":
  90             results = [
  91                 {'c2': 120, 'n': 1},
  92                 {'c2': 220, 'n': 1},
  93             ]
  94             description = (('c2',), ('n',))
  95         elif sql == "select c3, count(*) as n from ( select c1, c2, c3 from table2 ) where c3 is not null group by c3 limit 31":
  96             results = [
  97                 {'c3': 130, 'n': 1},
  98                 {'c3': 230, 'n': 1},
  99             ]
 100             description = (('c3',), ('n',))
 101         elif sql == 'select date(c1) from ( select c1, c2, c3 from table2 ) where c1 glob "????-??-*" limit 100;':
 102             pass
 103         elif sql == "select c1, c2, c3 from blah limit 51":
 104             raise dc.OperationalError("no such table: blah")
 105         else:
 106             raise Exception("Unexpected query:", sql)
 107
 108         return results, truncated, description