tests/graph_test.py

   1 """
   2 Test some of the basics underpinning the graph system.
   3 """
   4
   5 import os, unittest
   6
   7 from testlib import testutil, PygrTestProgram, SkipTest
   8 from pygr import mapping, graphquery, sqlgraph
   9
  10 class Query_Test(unittest.TestCase):
  11     "Pygr Query tests"
  12
  13     def dqcmp(self, datagraph, querygraph, result):
  14         try:
  15             g = self.datagraph
  16         except AttributeError:
  17             pass
  18         else:
  19             g.update(datagraph)
  20             datagraph = g
  21
  22         l = [ d.copy() for d in graphquery.GraphQuery(datagraph, querygraph) ]
  23         assert len(l) == len(result), 'length mismatch'
  24         l.sort()
  25         result.sort()
  26         for i in range(len(l)):
  27             assert l[i] == result[i], 'incorrect result'
  28
  29     def test_basicquery_test(self):
  30         "Basic query"
  31         datagraph = {0: {1: None, 2: None, 3: None},
  32                      1: {2: None}, 3: {4: None, 5: None},
  33                      4: {6: None}, 5: {6: None}, 2: {}, 6: {}}
  34         querygraph = {0: {1: None, 2: None, 3: None},
  35                       3:{4: None},1:{},2:{},4:{}}
  36         result = [{0: 0, 1: 1, 2: 2, 3: 3, 4: 4},
  37                   {0: 0, 1: 1, 2: 2, 3: 3, 4: 5},
  38                   {0: 0, 1: 2, 2: 1, 3: 3, 4: 4},
  39                   {0: 0, 1: 2, 2: 1, 3: 3, 4: 5}]
  40
  41         self.dqcmp(datagraph, querygraph, result)
  42
  43     def test_cyclicquery(self):
  44         "Cyclic QG against cyclic DG @CTB comment?"
  45         datagraph = { 1 :{2:None}, 2:{3:None}, 3:{4:None}, 4:{5:None},
  46                       5:{2:None}}
  47         querygraph = {0:{1:None}, 1:{2:None}, 2:{4:None}, 3:{1:None},
  48                       4:{3:None}}
  49         result = [ {0: 1, 1: 2, 2: 3, 3: 5, 4: 4} ]
  50         self.dqcmp(datagraph, querygraph, result)
  51
  52     def test_cyclicacyclicquery(self):
  53         "Cyclic QG against acyclic DG"
  54         datagraph = {0:{1:None}, 1:{3:None}, 5:{3:None}, 4:{5:None},
  55                      2:{4:None,1:None}, 3:{}}
  56         querygraph = {0:{1:None}, 1:{3:None}, 3:{5:None}, 5:{4:None},
  57                       4:{2:None}, 2:{1:None}}
  58         result = []
  59         self.dqcmp(datagraph,querygraph,result)
  60
  61     def test_symmetricquery_test(self):
  62         "Symmetrical QG against symmetrical DG"
  63         datagraph = {1:{2:None},2:{3:None,4:None},5:{2:None},3:{},4:{}}
  64         querygraph = {0:{1:None},1:{2:None},2:{}}
  65         result = [{0: 1, 1: 2, 2: 3}, {0: 1, 1: 2, 2: 4},
  66                   {0: 5, 1: 2, 2: 3}, {0: 5, 1: 2, 2: 4}]
  67         self.dqcmp(datagraph,querygraph,result)
  68
  69     def test_filteredquery(self):
  70         "Test a filter against a query"
  71         datagraph = {0: {1: None, 2: None, 3: None}, 1: {2: None, 3: None},
  72                      3: {4: None}}
  73         querygraph = {0:{1:{'filter':lambda toNode,**kw:toNode == 3}},1:{}}
  74         result = [{0: 0, 1: 3},{0: 1, 1: 3}]
  75         self.dqcmp(datagraph,querygraph,result)
  76
  77     def test_headlessquery(self):
  78         "Test a query with no head nodes"
  79         datagraph = {0:{1:None},1:{2:None},2:{3:None},3:{4:None},4:{1:None}}
  80         querygraph = {0:{1:None},1:{2:None},2:{3:None},3:{0:None}}
  81         result = [{0: 1, 1: 2, 2: 3, 3: 4},
  82                   {0: 2, 1: 3, 2: 4, 3: 1},
  83                   {0: 3, 1: 4, 2: 1, 3: 2},
  84                   {0: 4, 1: 1, 2: 2, 3: 3}]
  85         self.dqcmp(datagraph,querygraph,result)
  86
  87 class Mapping_Test(Query_Test):
  88     "Tests mappings"
  89
  90     def setUp(self):
  91         self.datagraph = mapping.dictGraph()
  92
  93     def test_graphdict(self):
  94         "Graph dictionary"
  95         datagraph = self.datagraph
  96         datagraph += 1
  97         datagraph[1] += 2
  98         results = {1: {2: None}, 2: {}}
  99         assert datagraph == results, 'incorrect result'
 100
 101     def test_nodedel(self):
 102         "Node deletion"
 103         datagraph = self.datagraph
 104         datagraph += 1
 105         datagraph += 2
 106         datagraph[2] += 3
 107         datagraph -= 1
 108         results = {2: {3: None}, 3: {}}
 109         assert datagraph == results, 'incorrect result'
 110
 111     def test_delraise(self):
 112         "Delete raise"
 113         datagraph = self.datagraph
 114         datagraph += 1
 115         datagraph += 2
 116         datagraph[2] += 3
 117         try:
 118             for i in range(0,2):
 119                 datagraph -= 3
 120             raise ValueError('failed to catch bad node deletion attempt')
 121         except KeyError:
 122             pass # THIS IS THE CORRECT RESULT
 123
 124     def test_setitemraise(self):
 125         "Setitemraise"
 126         datagraph = self.datagraph
 127         datagraph += 1
 128         try:
 129             datagraph[1] = 2
 130             raise KeyError('failed to catch bad setitem attempt')
 131         except ValueError:
 132             pass # THIS IS THE CORRECT RESULT
 133
 134     def test_graphedges(self):
 135         "Graphedges"
 136         datagraph = self.datagraph
 137         graphvals = {1:{2:None},2:{3:None,4:None},5:{2:None},3:{},4:{}}
 138         edge_list = [[1, 2,None], [2, 3,None], [2, 4,None], [5, 2,None]]
 139         for i in graphvals:
 140             datagraph += i
 141             for n in graphvals[i].keys():
 142                 datagraph[i] += n
 143         edge_results = []
 144         for e in datagraph.edges():
 145             edge_results.append(e)
 146         edge_results.sort()
 147         edge_results = [list(t) for t in edge_results]
 148         edge_list.sort()
 149         #print 'edge_results:',edge_results
 150         assert edge_results == edge_list, 'incorrect result'
 151
 152 class Graph_Test(Mapping_Test):
 153     "Run same tests on mapping.Graph class"
 154
 155     def setUp(self):
 156         self.datagraph = mapping.Graph()
 157
 158 class Graph_DB_Test(unittest.TestCase):
 159     "test mapping.Graph with sourceDB, targetDB but no edgeDB"
 160
 161     def setUp(self):
 162         class Node(object):
 163             def __init__(self, id):
 164                 self.id = id
 165         self.nodes = {1:Node(1), 2:Node(2)}
 166         self.datagraph = mapping.Graph(sourceDB=self.nodes,
 167                                        targetDB=self.nodes)
 168     def test_no_edge_db(self):
 169         'test behavior with no edgeDB'
 170         self.datagraph += self.nodes[1] # add node
 171         self.datagraph[self.nodes[1]][self.nodes[2]] = 3 # add edge
 172
 173         assert self.datagraph[self.nodes[1]][self.nodes[2]] == 3
 174
 175
 176 class GraphShelve_Test(Mapping_Test):
 177     "Run same tests on mapping.Graph class"
 178
 179     def setUp(self):
 180
 181         tmp = testutil.TempDir('graphshelve-test')
 182         filename = tmp.subfile() # needs a random name each time
 183         self.datagraph = mapping.Graph(filename=filename, intKeys=True)
 184
 185     def tearDown(self):
 186         self.datagraph.close()
 187
 188
 189 class SQLGraph_Test(Mapping_Test):
 190     "Runs the same tests on mapping.SQLGraph class"
 191     dbname = 'test.dumbo_foo_test'
 192
 193     def setUp(self):
 194         if not testutil.mysql_enabled():
 195             raise SkipTest, "no MySQL"
 196
 197         createOpts = dict(source_id='int', target_id='int', edge_id='int')
 198         self.datagraph = sqlgraph.SQLGraph(self.dbname, dropIfExists=True,
 199                                            createTable=createOpts)
 200
 201     def tearDown(self):
 202         self.datagraph.cursor.execute('drop table if exists %s' % self.dbname)
 203
 204 class SQLiteGraph_Test(testutil.SQLite_Mixin, Mapping_Test):
 205     'run same tests on mapping.SQLGraph class using sqlite'
 206     def sqlite_load(self):
 207         createOpts = dict(source_id='int', target_id='int', edge_id='int')
 208         self.datagraph = sqlgraph.SQLGraph('testgraph',
 209                                            serverInfo=self.serverInfo,
 210                                            dropIfExists=True,
 211                                            createTable=createOpts)
 212
 213 # test currently unused, requires access to leelab data
 214 ## from pygr import worldbase
 215 ## class Splicegraph_Test(unittest.TestCase):
 216
 217 ##     def setUp(self):
 218 ##         self.sg = worldbase.Bio.Annotation.ASAP2.Isoform.HUMAN.\
 219 ##                   hg17.splicegraph()
 220
 221 ##     def exonskip_megatest(self):
 222 ##         'perform exon skip query'
 223 ##         query = {0:{1:None,2:None},1:{2:None},2:{}}
 224 ##         gq = graphquery.GraphQuery(self.sg, query)
 225 ##         l = list(gq)
 226 ##         assert len(l) == 11546, 'test exact size of exonskip set'
 227
 228 if __name__ == '__main__':
 229     PygrTestProgram(verbosity=2)