78 self.failIf(osp.isfile(expected_filepath)) |
78 self.failIf(osp.isfile(expected_filepath)) |
79 |
79 |
80 def test_bfss_sqlite_fspath(self): |
80 def test_bfss_sqlite_fspath(self): |
81 f1 = self.create_file() |
81 f1 = self.create_file() |
82 expected_filepath = osp.join(self.tempdir, '%s_data' % f1.eid) |
82 expected_filepath = osp.join(self.tempdir, '%s_data' % f1.eid) |
83 fspath = self.execute('Any fspath(F, "File", "data") WHERE F eid %(f)s', |
83 fspath = self.execute('Any fspath(D) WHERE F eid %(f)s, F data D', |
84 {'f': f1.eid})[0][0] |
84 {'f': f1.eid})[0][0] |
85 self.assertEquals(fspath.getvalue(), expected_filepath) |
85 self.assertEquals(fspath.getvalue(), expected_filepath) |
86 |
86 |
87 def test_bfss_fs_importing_doesnt_touch_path(self): |
87 def test_bfss_fs_importing_doesnt_touch_path(self): |
88 self.session.transaction_data['fs_importing'] = True |
88 self.session.transaction_data['fs_importing'] = True |
89 f1 = self.session.create_entity('File', data=Binary('/the/path'), |
89 f1 = self.session.create_entity('File', data=Binary('/the/path'), |
90 data_format=u'text/plain', data_name=u'foo') |
90 data_format=u'text/plain', data_name=u'foo') |
91 fspath = self.execute('Any fspath(F, "File", "data") WHERE F eid %(f)s', |
91 fspath = self.execute('Any fspath(D) WHERE F eid %(f)s, F data D', |
92 {'f': f1.eid})[0][0] |
92 {'f': f1.eid})[0][0] |
93 self.assertEquals(fspath.getvalue(), '/the/path') |
93 self.assertEquals(fspath.getvalue(), '/the/path') |
94 |
94 |
95 def test_source_storage_transparency(self): |
95 def test_source_storage_transparency(self): |
96 self.vreg._loadedmods[__name__] = {} |
96 self.vreg._loadedmods[__name__] = {} |
100 self.create_file() |
100 self.create_file() |
101 finally: |
101 finally: |
102 self.vreg.unregister(DummyBeforeHook) |
102 self.vreg.unregister(DummyBeforeHook) |
103 self.vreg.unregister(DummyAfterHook) |
103 self.vreg.unregister(DummyAfterHook) |
104 |
104 |
|
105 def test_source_mapped_attribute_error_cases(self): |
|
106 ex = self.assertRaises(QueryError, self.execute, |
|
107 'Any X WHERE X data ~= "hop", X is File') |
|
108 self.assertEquals(str(ex), 'can\'t use File.data (X data ILIKE "hop") in restriction') |
|
109 ex = self.assertRaises(QueryError, self.execute, |
|
110 'Any X, Y WHERE X data D, Y data D, ' |
|
111 'NOT X identity Y, X is File, Y is File') |
|
112 self.assertEquals(str(ex), "can't use D as a restriction variable") |
|
113 # query returning mix of mapped / regular attributes (only file.data |
|
114 # mapped, not image.data for instance) |
|
115 ex = self.assertRaises(QueryError, self.execute, |
|
116 'Any X WITH X BEING (' |
|
117 ' (Any NULL)' |
|
118 ' UNION ' |
|
119 ' (Any D WHERE X data D, X is File)' |
|
120 ')') |
|
121 self.assertEquals(str(ex), 'query fetch some source mapped attribute, some not') |
|
122 ex = self.assertRaises(QueryError, self.execute, |
|
123 '(Any D WHERE X data D, X is File)' |
|
124 ' UNION ' |
|
125 '(Any D WHERE X data D, X is Image)') |
|
126 self.assertEquals(str(ex), 'query fetch some source mapped attribute, some not') |
|
127 ex = self.assertRaises(QueryError, |
|
128 self.execute, 'Any D WHERE X data D') |
|
129 self.assertEquals(str(ex), 'query fetch some source mapped attribute, some not') |
|
130 |
|
131 def test_source_mapped_attribute_advanced(self): |
|
132 f1 = self.create_file() |
|
133 rset = self.execute('Any X,D WITH D,X BEING (' |
|
134 ' (Any D, X WHERE X eid %(x)s, X data D)' |
|
135 ' UNION ' |
|
136 ' (Any D, X WHERE X eid %(x)s, X data D)' |
|
137 ')', {'x': f1.eid}, 'x') |
|
138 self.assertEquals(len(rset), 2) |
|
139 self.assertEquals(rset[0][0], f1.eid) |
|
140 self.assertEquals(rset[1][0], f1.eid) |
|
141 self.assertEquals(rset[0][1].getvalue(), 'the-data') |
|
142 self.assertEquals(rset[1][1].getvalue(), 'the-data') |
|
143 rset = self.execute('Any X,LENGTH(D) WHERE X eid %(x)s, X data D', |
|
144 {'x': f1.eid}, 'x') |
|
145 self.assertEquals(len(rset), 1) |
|
146 self.assertEquals(rset[0][0], f1.eid) |
|
147 self.assertEquals(rset[0][1], len('the-data')) |
|
148 rset = self.execute('Any X,LENGTH(D) WITH D,X BEING (' |
|
149 ' (Any D, X WHERE X eid %(x)s, X data D)' |
|
150 ' UNION ' |
|
151 ' (Any D, X WHERE X eid %(x)s, X data D)' |
|
152 ')', {'x': f1.eid}, 'x') |
|
153 self.assertEquals(len(rset), 2) |
|
154 self.assertEquals(rset[0][0], f1.eid) |
|
155 self.assertEquals(rset[1][0], f1.eid) |
|
156 self.assertEquals(rset[0][1], len('the-data')) |
|
157 self.assertEquals(rset[1][1], len('the-data')) |
|
158 ex = self.assertRaises(QueryError, self.execute, |
|
159 'Any X,UPPER(D) WHERE X eid %(x)s, X data D', |
|
160 {'x': f1.eid}, 'x') |
|
161 self.assertEquals(str(ex), 'UPPER can not be called on mapped attribute') |
|
162 |
105 if __name__ == '__main__': |
163 if __name__ == '__main__': |
106 unittest_main() |
164 unittest_main() |