blob: efef48dd02a97637f25200c0bb2d2d878c78109d [file] [log] [blame]
# Copyright 2016 The Chromium Authors. All rights reserved.
# Use of this source code is govered by a BSD-style
# license that can be found in the LICENSE file or at
# https://developers.google.com/open-source/licenses/bsd
"""Tests for the ast2select module."""
import unittest
from proto import ast_pb2
from proto import tracker_pb2
from search import ast2select
from search import query2ast
from tracker import tracker_bizobj
BUILTIN_ISSUE_FIELDS = query2ast.BUILTIN_ISSUE_FIELDS
ANY_FIELD = query2ast.BUILTIN_ISSUE_FIELDS['any_field']
class AST2SelectTest(unittest.TestCase):
def setUp(self):
self.config = tracker_bizobj.MakeDefaultProjectIssueConfig(789)
def testBuildSQLQuery_EmptyAST(self):
ast = ast_pb2.QueryAST(conjunctions=[ast_pb2.Conjunction()]) # No conds
left_joins, where = ast2select.BuildSQLQuery(ast)
self.assertEqual([], left_joins)
self.assertEqual([], where)
def testBuildSQLQuery_Normal(self):
owner_field = BUILTIN_ISSUE_FIELDS['owner']
reporter_id_field = BUILTIN_ISSUE_FIELDS['reporter_id']
conds = [
ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [owner_field], ['example.com'], []),
ast_pb2.MakeCond(
ast_pb2.QueryOp.EQ, [reporter_id_field], [], [111L])]
ast = ast_pb2.QueryAST(conjunctions=[ast_pb2.Conjunction(conds=conds)])
left_joins, where = ast2select.BuildSQLQuery(ast)
self.assertEqual(
[('User AS Cond0 ON (Issue.owner_id = Cond0.user_id '
'OR Issue.derived_owner_id = Cond0.user_id)', [])],
left_joins)
self.assertEqual(
[('LOWER(Cond0.email) LIKE %s', ['%example.com%']),
('Issue.reporter_id = %s', [111L])],
where)
def testBlockingIDCond_SingleValue(self):
fd = BUILTIN_ISSUE_FIELDS['blocking_id']
txt_cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['1'], [])
num_cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [1L])
for cond, expected in ((txt_cond, '1'), (num_cond, 1L)):
left_joins, where = ast2select._ProcessBlockingIDCond(
cond, 'Cond1', 'Issue1')
self.assertEqual(
[('IssueRelation AS Cond1 ON Issue.id = Cond1.dst_issue_id AND '
'Cond1.kind = %s',
['blockedon'])],
left_joins)
self.assertEqual(
[('Cond1.issue_id = %s', [expected])],
where)
def testBlockingIDCond_MultiValue(self):
fd = BUILTIN_ISSUE_FIELDS['blocking_id']
txt_cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['1', '2', '3'], [])
num_cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [1L, 2L, 3L])
for cond, expected in ((txt_cond, ['1', '2', '3']),
(num_cond, [1L, 2L, 3L])):
left_joins, where = ast2select._ProcessBlockingIDCond(
cond, 'Cond1', 'Issue1')
self.assertEqual(
[('IssueRelation AS Cond1 ON Issue.id = Cond1.dst_issue_id AND '
'Cond1.kind = %s',
['blockedon'])],
left_joins)
self.assertEqual(
[('Cond1.issue_id IN (%s,%s,%s)', expected)],
where)
def testBlockedOnIDCond_SingleValue(self):
fd = BUILTIN_ISSUE_FIELDS['blockedon_id']
txt_cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['1'], [])
num_cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [1L])
for cond, expected in ((txt_cond, '1'), (num_cond, 1L)):
left_joins, where = ast2select._ProcessBlockedOnIDCond(
cond, 'Cond1', 'Issue1')
self.assertEqual(
[('IssueRelation AS Cond1 ON Issue.id = Cond1.issue_id AND '
'Cond1.kind = %s',
['blockedon'])],
left_joins)
self.assertEqual(
[('Cond1.dst_issue_id = %s', [expected])],
where)
def testBlockedIDCond_MultiValue(self):
fd = BUILTIN_ISSUE_FIELDS['blockedon_id']
txt_cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['1', '2', '3'], [])
num_cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [1L, 2L, 3L])
for cond, expected in ((txt_cond, ['1', '2', '3']),
(num_cond, [1L, 2L, 3L])):
left_joins, where = ast2select._ProcessBlockedOnIDCond(
cond, 'Cond1', 'Issue1')
self.assertEqual(
[('IssueRelation AS Cond1 ON Issue.id = Cond1.issue_id AND '
'Cond1.kind = %s',
['blockedon'])],
left_joins)
self.assertEqual(
[('Cond1.dst_issue_id IN (%s,%s,%s)', expected)],
where)
def testHasBlockedCond(self):
for op, expected in ((ast_pb2.QueryOp.IS_DEFINED, 'IS NOT NULL'),
(ast_pb2.QueryOp.IS_NOT_DEFINED, 'IS NULL')):
fd = BUILTIN_ISSUE_FIELDS['blockedon_id']
cond = ast_pb2.MakeCond(op, [fd], [], [])
left_joins, where = ast2select._ProcessBlockedOnIDCond(
cond, 'Cond1', None)
self.assertEqual(
[('IssueRelation AS Cond1 ON Issue.id = Cond1.issue_id AND '
'Cond1.kind = %s', ['blockedon'])],
left_joins)
self.assertEqual([('Cond1.issue_id %s' % expected, [])], where)
def testHasBlockingCond(self):
for op, expected in ((ast_pb2.QueryOp.IS_DEFINED, 'IS NOT NULL'),
(ast_pb2.QueryOp.IS_NOT_DEFINED, 'IS NULL')):
fd = BUILTIN_ISSUE_FIELDS['blocking_id']
cond = ast_pb2.MakeCond(op, [fd], [], [])
left_joins, where = ast2select._ProcessBlockingIDCond(cond, 'Cond1', None)
self.assertEqual(
[('IssueRelation AS Cond1 ON Issue.id = Cond1.dst_issue_id AND '
'Cond1.kind = %s', ['blockedon'])],
left_joins)
self.assertEqual([('Cond1.dst_issue_id %s' % expected, [])], where)
def testProcessOwnerCond(self):
fd = BUILTIN_ISSUE_FIELDS['owner']
cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['example.com'], [])
left_joins, where = ast2select._ProcessOwnerCond(cond, 'Cond1', 'User1')
self.assertEqual(
[('User AS Cond1 ON (Issue.owner_id = Cond1.user_id '
'OR Issue.derived_owner_id = Cond1.user_id)', [])],
left_joins)
self.assertEqual(
[('LOWER(Cond1.email) LIKE %s', ['%example.com%'])],
where)
def testProcessOwnerIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['owner_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [111L])
left_joins, where = ast2select._ProcessOwnerIDCond(cond, 'Cond1', 'User1')
self.assertEqual([], left_joins)
self.assertEqual(
[('(Issue.owner_id = %s OR Issue.derived_owner_id = %s)',
[111L, 111L])],
where)
def testProcessReporterCond(self):
fd = BUILTIN_ISSUE_FIELDS['reporter']
cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['example.com'], [])
left_joins, where = ast2select._ProcessReporterCond(cond, 'Cond1', 'User1')
self.assertEqual(
[('User AS Cond1 ON Issue.reporter_id = Cond1.user_id', [])],
left_joins)
self.assertEqual(
[('LOWER(Cond1.email) LIKE %s', ['%example.com%'])],
where)
def testProcessReporterIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['reporter_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [111L])
left_joins, where = ast2select._ProcessReporterIDCond(
cond, 'Cond1', 'User1')
self.assertEqual([], left_joins)
self.assertEqual(
[('Issue.reporter_id = %s', [111L])],
where)
def testProcessCcCond(self):
fd = BUILTIN_ISSUE_FIELDS['cc']
cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['example.com'], [])
left_joins, where = ast2select._ProcessCcCond(cond, 'Cond1', 'User1')
self.assertEqual(
[('(Issue2Cc AS Cond1 JOIN User AS User1 '
'ON Cond1.cc_id = User1.user_id AND LOWER(User1.email) LIKE %s) '
'ON Issue.id = Cond1.issue_id AND Issue.shard = Cond1.issue_shard',
['%example.com%'])],
left_joins)
self.assertEqual(
[('User1.email IS NOT NULL', [])],
where)
def testProcessCcIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['cc_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [111L])
left_joins, where = ast2select._ProcessCcIDCond(cond, 'Cond1', 'User1')
self.assertEqual(
[('Issue2Cc AS Cond1 ON Issue.id = Cond1.issue_id AND '
'Issue.shard = Cond1.issue_shard AND '
'Cond1.cc_id = %s',
[111L])],
left_joins)
self.assertEqual(
[('Cond1.cc_id IS NOT NULL', [])],
where)
def testProcessStarredByCond(self):
fd = BUILTIN_ISSUE_FIELDS['starredby']
cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['example.com'], [])
left_joins, where = ast2select._ProcessStarredByCond(
cond, 'Cond1', 'User1')
self.assertEqual(
[('(IssueStar AS Cond1 JOIN User AS User1 '
'ON Cond1.user_id = User1.user_id AND LOWER(User1.email) LIKE %s) '
'ON Issue.id = Cond1.issue_id', ['%example.com%'])],
left_joins)
self.assertEqual(
[('User1.email IS NOT NULL', [])],
where)
def testProcessStarredByIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['starredby_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [111L])
left_joins, where = ast2select._ProcessStarredByIDCond(
cond, 'Cond1', 'User1')
self.assertEqual(
[('IssueStar AS Cond1 ON Issue.id = Cond1.issue_id '
'AND Cond1.user_id = %s', [111L])],
left_joins)
self.assertEqual(
[('Cond1.user_id IS NOT NULL', [])],
where)
def testProcessCommentByCond(self):
fd = BUILTIN_ISSUE_FIELDS['commentby']
cond = ast_pb2.MakeCond(
ast_pb2.QueryOp.TEXT_HAS, [fd], ['example.com'], [])
left_joins, where = ast2select._ProcessCommentByCond(
cond, 'Cond1', 'User1')
self.assertEqual(
[('(Comment AS Cond1 JOIN User AS User1 '
'ON Cond1.commenter_id = User1.user_id '
'AND LOWER(User1.email) LIKE %s) '
'ON Issue.id = Cond1.issue_id', ['%example.com%'])],
left_joins)
self.assertEqual(
[('User1.email IS NOT NULL', [])],
where)
def testProcessCommentByIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['commentby_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [111L])
left_joins, where = ast2select._ProcessCommentByIDCond(
cond, 'Cond1', 'User1')
self.assertEqual(
[('Comment AS Cond1 ON Issue.id = Cond1.issue_id', [])],
left_joins)
self.assertEqual(
[('Cond1.commenter_id = %s', [111L])],
where)
def testProcessStatusIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['status_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [2])
left_joins, where = ast2select._ProcessStatusIDCond(cond, 'Cond1', 'User1')
self.assertEqual([], left_joins)
self.assertEqual(
[('(Issue.status_id = %s OR Issue.derived_status_id = %s)', [2, 2])],
where)
def testProcessLabelIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['label_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [1])
left_joins, where = ast2select._ProcessLabelIDCond(cond, 'Cond1', 'User1')
self.assertEqual(
[('Issue2Label AS Cond1 ON Issue.id = Cond1.issue_id AND '
'Issue.shard = Cond1.issue_shard AND '
'Cond1.label_id = %s', [1])],
left_joins)
self.assertEqual(
[('Cond1.label_id IS NOT NULL', [])],
where)
def testProcessComponentIDCond(self):
fd = BUILTIN_ISSUE_FIELDS['component_id']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.EQ, [fd], [], [101])
left_joins, where = ast2select._ProcessComponentIDCond(
cond, 'Cond1', 'User1')
self.assertEqual(
[('Issue2Component AS Cond1 ON Issue.id = Cond1.issue_id AND '
'Issue.shard = Cond1.issue_shard AND '
'Cond1.component_id = %s', [101])],
left_joins)
self.assertEqual(
[('Cond1.component_id IS NOT NULL', [])],
where)
def testProcessCustomFieldCond(self):
pass # TODO(jrobbins): fill in this test case.
def testProcessAttachmentCond_HasAttachment(self):
fd = BUILTIN_ISSUE_FIELDS['attachment']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.IS_DEFINED, [fd], [], [])
left_joins, where = ast2select._ProcessAttachmentCond(
cond, 'Cond1', 'User1')
self.assertEqual([], left_joins)
self.assertEqual(
[('(Issue.attachment_count IS NOT NULL AND '
'Issue.attachment_count != %s)',
[0])],
where)
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.IS_NOT_DEFINED, [fd], [], [])
left_joins, where = ast2select._ProcessAttachmentCond(
cond, 'Cond1', 'User1')
self.assertEqual([], left_joins)
self.assertEqual(
[('(Issue.attachment_count IS NULL OR '
'Issue.attachment_count = %s)',
[0])],
where)
def testProcessAttachmentCond_TextHas(self):
fd = BUILTIN_ISSUE_FIELDS['attachment']
cond = ast_pb2.MakeCond(ast_pb2.QueryOp.TEXT_HAS, [fd], ['jpg'], [])
left_joins, where = ast2select._ProcessAttachmentCond(
cond, 'Cond1', 'User1')
self.assertEqual(
[('Attachment AS Cond1 ON Issue.id = Cond1.issue_id AND '
'Cond1.deleted = %s',
[False])],
left_joins)
self.assertEqual(
[('Cond1.filename LIKE %s', ['%jpg%'])],
where)
def testCompare_IntTypes(self):
val_type = tracker_pb2.FieldTypes.INT_TYPE
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.IS_DEFINED, val_type, 'col', [1, 2])
self.assertEqual('(Alias.col IS NOT NULL AND Alias.col != %s)', cond_str)
self.assertEqual([0], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.EQ, val_type, 'col', [1])
self.assertEqual('Alias.col = %s', cond_str)
self.assertEqual([1], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.EQ, val_type, 'col', [1, 2])
self.assertEqual('Alias.col IN (%s,%s)', cond_str)
self.assertEqual([1, 2], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.NE, val_type, 'col', [])
self.assertEqual('TRUE', cond_str)
self.assertEqual([], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.NE, val_type, 'col', [1])
self.assertEqual('(Alias.col IS NULL OR Alias.col != %s)', cond_str)
self.assertEqual([1], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.NE, val_type, 'col', [1, 2])
self.assertEqual('(Alias.col IS NULL OR Alias.col NOT IN (%s,%s))',
cond_str)
self.assertEqual([1, 2], cond_args)
def testCompare_STRTypes(self):
val_type = tracker_pb2.FieldTypes.STR_TYPE
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.IS_DEFINED, val_type, 'col', ['a', 'b'])
self.assertEqual('(Alias.col IS NOT NULL AND Alias.col != %s)', cond_str)
self.assertEqual([''], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.EQ, val_type, 'col', ['a'])
self.assertEqual('Alias.col = %s', cond_str)
self.assertEqual(['a'], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.EQ, val_type, 'col', ['a', 'b'])
self.assertEqual('Alias.col IN (%s,%s)', cond_str)
self.assertEqual(['a', 'b'], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.NE, val_type, 'col', [])
self.assertEqual('TRUE', cond_str)
self.assertEqual([], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.NE, val_type, 'col', ['a'])
self.assertEqual('(Alias.col IS NULL OR Alias.col != %s)', cond_str)
self.assertEqual(['a'], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.NE, val_type, 'col', ['a', 'b'])
self.assertEqual('(Alias.col IS NULL OR Alias.col NOT IN (%s,%s))',
cond_str)
self.assertEqual(['a', 'b'], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.TEXT_HAS, val_type, 'col', ['a'])
self.assertEqual('Alias.col LIKE %s', cond_str)
self.assertEqual(['%a%'], cond_args)
cond_str, cond_args = ast2select._Compare(
'Alias', ast_pb2.QueryOp.NOT_TEXT_HAS, val_type, 'col', ['a'])
self.assertEqual('(Alias.col IS NULL OR Alias.col NOT LIKE %s)', cond_str)
self.assertEqual(['%a%'], cond_args)
def testCompareAlreadyJoined(self):
cond_str, cond_args = ast2select._CompareAlreadyJoined(
'Alias', ast_pb2.QueryOp.EQ, 'col')
self.assertEqual('Alias.col IS NOT NULL', cond_str)
self.assertEqual([], cond_args)
cond_str, cond_args = ast2select._CompareAlreadyJoined(
'Alias', ast_pb2.QueryOp.NE, 'col')
self.assertEqual('Alias.col IS NULL', cond_str)
self.assertEqual([], cond_args)
if __name__ == '__main__':
unittest.main()