blob: ac5a00b044f70c00a9e94d0d68293fb094ead6e4 [file] [log] [blame]
# Copyright 2016 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Tests for the ast2sort module."""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
import unittest
from mrproto import tracker_pb2
from search import ast2sort
from search import query2ast
BUILTIN_ISSUE_FIELDS = query2ast.BUILTIN_ISSUE_FIELDS
ANY_FIELD = query2ast.BUILTIN_ISSUE_FIELDS['any_field']
class AST2SortTest(unittest.TestCase):
def setUp(self):
self.harmonized_labels = [
(101, 0, 'Hot'), (102, 1, 'Cold'), (103, None, 'Odd')]
self.harmonized_statuses = [
(201, 0, 'New'), (202, 1, 'Assigned'), (203, None, 'OnHold')]
self.harmonized_fields = []
self.fmt = lambda string, **kwords: string
def testBuildSortClauses_EmptySortDirectives(self):
left_joins, order_by = ast2sort.BuildSortClauses(
[], self.harmonized_labels, self.harmonized_statuses,
self.harmonized_fields)
self.assertEqual([], left_joins)
self.assertEqual([], order_by)
def testBuildSortClauses_Normal(self):
left_joins, order_by = ast2sort.BuildSortClauses(
['stars', 'status', 'pri', 'reporter', 'id'], self.harmonized_labels,
self.harmonized_statuses, self.harmonized_fields)
expected_left_joins = [
('User AS Sort3 ON Issue.reporter_id = Sort3.user_id', [])]
expected_order_by = [
('Issue.star_count ASC', []),
('FIELD(IF(ISNULL(Issue.status_id), Issue.derived_status_id, '
'Issue.status_id), %s,%s) DESC', [201, 202]),
('FIELD(IF(ISNULL(Issue.status_id), Issue.derived_status_id, '
'Issue.status_id), %s) DESC', [203]),
('ISNULL(Sort3.email) ASC', []),
('Sort3.email ASC', []),
('Issue.local_id ASC', [])]
self.assertEqual(expected_left_joins, left_joins)
self.assertEqual(expected_order_by, order_by)
def testProcessProjectSD(self):
left_joins, order_by = ast2sort._ProcessProjectSD(self.fmt)
self.assertEqual([], left_joins)
self.assertEqual(
[('Issue.project_id {sort_dir}', [])],
order_by)
def testProcessReporterSD(self):
left_joins, order_by = ast2sort._ProcessReporterSD(self.fmt)
self.assertEqual(
[('User AS {alias} ON Issue.reporter_id = {alias}.user_id', [])],
left_joins)
self.assertEqual(
[('ISNULL({alias}.email) {sort_dir}', []),
('{alias}.email {sort_dir}', [])],
order_by)
def testProcessOwnerSD(self):
left_joins, order_by = ast2sort._ProcessOwnerSD(self.fmt)
self.assertEqual(
[('User AS {alias}_exp ON Issue.owner_id = {alias}_exp.user_id', []),
('User AS {alias}_der ON '
'Issue.derived_owner_id = {alias}_der.user_id', [])],
left_joins)
self.assertEqual(
[('(ISNULL({alias}_exp.email) AND ISNULL({alias}_der.email)) '
'{sort_dir}', []),
('CONCAT({alias}_exp.email, {alias}_der.email) {sort_dir}', [])],
order_by)
def testProcessCcSD(self):
left_joins, order_by = ast2sort._ProcessCcSD(self.fmt)
self.assertEqual(
[('Issue2Cc AS {alias} ON Issue.id = {alias}.issue_id '
'LEFT JOIN User AS {alias}_user '
'ON {alias}.cc_id = {alias}_user.user_id', [])],
left_joins)
self.assertEqual(
[('ISNULL({alias}_user.email) {sort_dir}', []),
('{alias}_user.email {sort_dir}', [])],
order_by)
def testProcessComponentSD(self):
left_joins, order_by = ast2sort._ProcessComponentSD(self.fmt)
self.assertEqual(
[('Issue2Component AS {alias} ON Issue.id = {alias}.issue_id '
'LEFT JOIN ComponentDef AS {alias}_component '
'ON {alias}.component_id = {alias}_component.id', [])],
left_joins)
self.assertEqual(
[('ISNULL({alias}_component.path) {sort_dir}', []),
('{alias}_component.path {sort_dir}', [])],
order_by)
def testProcessSummarySD(self):
left_joins, order_by = ast2sort._ProcessSummarySD(self.fmt)
self.assertEqual(
[('IssueSummary AS {alias} ON Issue.id = {alias}.issue_id', [])],
left_joins)
self.assertEqual(
[('{alias}.summary {sort_dir}', [])],
order_by)
def testProcessStatusSD(self):
pass # TODO(jrobbins): fill in this test case
def testProcessBlockedSD(self):
left_joins, order_by = ast2sort._ProcessBlockedSD(self.fmt)
self.assertEqual(
[('IssueRelation AS {alias} ON Issue.id = {alias}.issue_id '
'AND {alias}.kind = %s', ['blockedon'])],
left_joins)
self.assertEqual(
[('ISNULL({alias}.dst_issue_id) {sort_dir}', [])],
order_by)
def testProcessBlockedOnSD(self):
left_joins, order_by = ast2sort._ProcessBlockedOnSD(self.fmt)
self.assertEqual(
[('IssueRelation AS {alias} ON Issue.id = {alias}.issue_id '
'AND {alias}.kind = %s', ['blockedon'])],
left_joins)
self.assertEqual(
[('ISNULL({alias}.dst_issue_id) {sort_dir}', []),
('{alias}.dst_issue_id {sort_dir}', [])],
order_by)
def testProcessBlockingSD(self):
left_joins, order_by = ast2sort._ProcessBlockingSD(self.fmt)
self.assertEqual(
[('IssueRelation AS {alias} ON Issue.id = {alias}.dst_issue_id '
'AND {alias}.kind = %s', ['blockedon'])],
left_joins)
self.assertEqual(
[('ISNULL({alias}.issue_id) {sort_dir}', []),
('{alias}.issue_id {sort_dir}', [])],
order_by)
def testProcessMergedIntoSD(self):
left_joins, order_by = ast2sort._ProcessMergedIntoSD(self.fmt)
self.assertEqual(
[('IssueRelation AS {alias} ON Issue.id = {alias}.issue_id '
'AND {alias}.kind = %s', ['mergedinto'])],
left_joins)
self.assertEqual(
[('ISNULL({alias}.dst_issue_id) {sort_dir}', []),
('{alias}.dst_issue_id {sort_dir}', [])],
order_by)
def testProcessCustomAndLabelSD(self):
pass # TODO(jrobbins): fill in this test case
def testProcessCustomAndLabelSD_PhaseField(self):
harmonized_labels = []
bear_fd = tracker_pb2.FieldDef(
field_id=1, field_name='DropBear', project_id=789,
field_type=tracker_pb2.FieldTypes.INT_TYPE)
bear2_fd = tracker_pb2.FieldDef(
field_id=2, field_name='DropBear', project_id=788,
field_type=tracker_pb2.FieldTypes.STR_TYPE)
koala_fd = tracker_pb2.FieldDef(
field_id=3, field_name='koala', project_id=789,
field_type=tracker_pb2.FieldTypes.INT_TYPE)
bear_app_fd = tracker_pb2.FieldDef(
field_id=4, field_name='dropbear', project_id=789,
field_type=tracker_pb2.FieldTypes.APPROVAL_TYPE)
harmonized_fields = [bear_fd, bear2_fd, koala_fd, bear_app_fd]
phase_name = 'stable'
alias = 'Sort0'
sort_dir = 'DESC'
sd = 'stable.dropbear'
left_joins, order_by = ast2sort._ProcessCustomAndLabelSD(
sd, harmonized_labels, harmonized_fields, alias, sort_dir,
self.fmt)
expected_joins = []
expected_order = []
int_left_joins, int_order_by = ast2sort._CustomFieldSortClauses(
[bear_fd, bear2_fd], tracker_pb2.FieldTypes.INT_TYPE, 'int_value',
alias, sort_dir, phase_name=phase_name)
str_left_joins, str_order_by = ast2sort._CustomFieldSortClauses(
[bear_fd, bear2_fd], tracker_pb2.FieldTypes.STR_TYPE, 'str_value',
alias, sort_dir, phase_name=phase_name)
user_left_joins, user_order_by = ast2sort._CustomFieldSortClauses(
[bear_fd, bear2_fd], tracker_pb2.FieldTypes.USER_TYPE, 'user_id',
alias, sort_dir, phase_name=phase_name)
label_left_joinss, label_order_by = ast2sort._LabelSortClauses(
sd, harmonized_labels, self.fmt)
expected_joins.extend(
int_left_joins + str_left_joins + user_left_joins + label_left_joinss)
expected_order.extend(
int_order_by + str_order_by + user_order_by + label_order_by)
self.assertEqual(left_joins, expected_joins)
self.assertEqual(order_by, expected_order)
def testApprovalFieldSortClauses_Status(self):
approval_fd_list = [
tracker_pb2.FieldDef(field_id=2, project_id=789,
field_type=tracker_pb2.FieldTypes.APPROVAL_TYPE),
tracker_pb2.FieldDef(field_id=4, project_id=788,
field_type=tracker_pb2.FieldTypes.APPROVAL_TYPE)
]
left_joins, order_by = ast2sort._ApprovalFieldSortClauses(
approval_fd_list, '-status', self.fmt)
self.assertEqual(
[('{tbl_name} AS {alias}_approval '
'ON Issue.id = {alias}_approval.issue_id '
'AND {alias}_approval.approval_id IN ({approval_ids_ph})', [2, 4])],
left_joins)
self.assertEqual(
[('FIELD({alias}_approval.status, {approval_status_ph}) {rev_sort_dir}',
ast2sort.APPROVAL_STATUS_SORT_ORDER)],
order_by)
def testApprovalFieldSortClauses_Approver(self):
approval_fd_list = [
tracker_pb2.FieldDef(field_id=2, project_id=789,
field_type=tracker_pb2.FieldTypes.APPROVAL_TYPE),
tracker_pb2.FieldDef(field_id=4, project_id=788,
field_type=tracker_pb2.FieldTypes.APPROVAL_TYPE)
]
left_joins, order_by = ast2sort._ApprovalFieldSortClauses(
approval_fd_list, '-approver', self.fmt)
self.assertEqual(
[('{tbl_name} AS {alias}_approval '
'ON Issue.id = {alias}_approval.issue_id '
'AND {alias}_approval.approval_id IN ({approval_ids_ph})', [2, 4]),
('User AS {alias}_approval_user '
'ON {alias}_approval.approver_id = {alias}_approval_user.user_id',
[])],
left_joins)
self.assertEqual(
[('ISNULL({alias}_approval_user.email) {sort_dir}', []),
('{alias}_approval_user.email {sort_dir}', [])],
order_by)
def testLabelSortClauses_NoSuchLabels(self):
sd = 'somethingelse'
harmonized_labels = [
(101, 0, 'Type-Defect'),
(102, 1, 'Type-Enhancement'),
(103, 2, 'Type-Task'),
(104, 0, 'Priority-High'),
(199, None, 'Type-Laundry'),
]
left_joins, order_by = ast2sort._LabelSortClauses(
sd, harmonized_labels, self.fmt)
self.assertEqual([], left_joins)
self.assertEqual([], order_by)
def testLabelSortClauses_Normal(self):
sd = 'type'
harmonized_labels = [
(101, 0, 'Type-Defect'),
(102, 1, 'Type-Enhancement'),
(103, 2, 'Type-Task'),
(104, 0, 'Priority-High'),
(199, None, 'Type-Laundry'),
]
left_joins, order_by = ast2sort._LabelSortClauses(
sd, harmonized_labels, self.fmt)
self.assertEqual(1, len(left_joins))
self.assertEqual(
('Issue2Label AS {alias} ON Issue.id = {alias}.issue_id AND '
'{alias}.label_id IN ({all_label_ph})',
[101, 102, 103, 199]),
left_joins[0])
self.assertEqual(2, len(order_by))
self.assertEqual(
('FIELD({alias}.label_id, {wk_label_ph}) {rev_sort_dir}',
[101, 102, 103]),
order_by[0])
self.assertEqual(
('FIELD({alias}.label_id, {odd_label_ph}) {rev_sort_dir}',
[199]),
order_by[1])
def testCustomFieldSortClauses_Normal(self):
fd_list = [
tracker_pb2.FieldDef(field_id=1, project_id=789,
field_type=tracker_pb2.FieldTypes.INT_TYPE),
tracker_pb2.FieldDef(field_id=2, project_id=788,
field_type=tracker_pb2.FieldTypes.STR_TYPE),
]
left_joins, order_by = ast2sort._CustomFieldSortClauses(
fd_list, tracker_pb2.FieldTypes.INT_TYPE, 'int_value', 'Sort0', 'DESC')
self.assertEqual(
left_joins, [
('Issue2FieldValue AS Sort0_int_value '
'ON Issue.id = Sort0_int_value.issue_id '
'AND Sort0_int_value.field_id IN (%s)', [1]),
])
self.assertEqual(
order_by, [
('ISNULL(Sort0_int_value.int_value) DESC', []),
('Sort0_int_value.int_value DESC', []),
])
def testCustomFieldSortClauses_PhaseUser(self):
fd_list = [
tracker_pb2.FieldDef(field_id=1, project_id=789,
field_type=tracker_pb2.FieldTypes.INT_TYPE),
tracker_pb2.FieldDef(field_id=2, project_id=788,
field_type=tracker_pb2.FieldTypes.STR_TYPE),
tracker_pb2.FieldDef(field_id=3, project_id=788,
field_type=tracker_pb2.FieldTypes.USER_TYPE),
]
left_joins, order_by = ast2sort._CustomFieldSortClauses(
fd_list, tracker_pb2.FieldTypes.USER_TYPE, 'user_id', 'Sort0', 'DESC',
phase_name='Stable')
self.assertEqual(
left_joins, [
('Issue2FieldValue AS Sort0_user_id '
'ON Issue.id = Sort0_user_id.issue_id '
'AND Sort0_user_id.field_id IN (%s)', [3]),
('IssuePhaseDef AS Sort0_user_id_phase '
'ON Sort0_user_id.phase_id = Sort0_user_id_phase.id '
'AND LOWER(Sort0_user_id_phase.name) = LOWER(%s)', ['Stable']),
('User AS Sort0_user_id_user '
'ON Sort0_user_id.user_id = Sort0_user_id_user.user_id', []),
])
self.assertEqual(
order_by, [
('ISNULL(Sort0_user_id_user.email) DESC', []),
('Sort0_user_id_user.email DESC', []),
])
def testOneSortDirective_NativeSortable(self):
left_joins, order_by = ast2sort._OneSortDirective(
1, 'opened', self.harmonized_labels, self.harmonized_statuses,
self.harmonized_fields)
self.assertEqual([], left_joins)
self.assertEqual([('Issue.opened ASC', [])], order_by)
left_joins, order_by = ast2sort._OneSortDirective(
1, 'stars', self.harmonized_labels, self.harmonized_statuses,
self.harmonized_fields)
self.assertEqual([], left_joins)
self.assertEqual([('Issue.star_count ASC', [])], order_by)
left_joins, order_by = ast2sort._OneSortDirective(
1, '-stars', self.harmonized_labels, self.harmonized_statuses,
self.harmonized_fields)
self.assertEqual([], left_joins)
self.assertEqual([('Issue.star_count DESC', [])], order_by)
left_joins, order_by = ast2sort._OneSortDirective(
1, 'componentmodified', self.harmonized_labels,
self.harmonized_statuses, self.harmonized_fields)
self.assertEqual([], left_joins)
self.assertEqual([('Issue.component_modified ASC', [])], order_by)