blob: 351ec62f8fbf8bd918cb734fe8b3ec0289a1850b [file] [log] [blame]
# Copyright 2016 The Chromium Authors. All rights reserved.
# Use of this source code is governed 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 spam service."""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
import mock
import unittest
try:
from mox3 import mox
except ImportError:
import mox
from google.appengine.ext import testbed
import settings
from framework import sql
from framework import framework_constants
from proto import user_pb2
from proto import tracker_pb2
from services import spam_svc
from testing import fake
from mock import Mock
def assert_unreached():
raise Exception('This code should not have been called.') # pragma: no cover
class SpamServiceTest(unittest.TestCase):
def setUp(self):
self.testbed = testbed.Testbed()
self.testbed.activate()
self.mox = mox.Mox()
self.mock_report_tbl = self.mox.CreateMock(sql.SQLTableManager)
self.mock_verdict_tbl = self.mox.CreateMock(sql.SQLTableManager)
self.mock_issue_tbl = self.mox.CreateMock(sql.SQLTableManager)
self.cnxn = self.mox.CreateMock(sql.MonorailConnection)
self.issue_service = fake.IssueService()
self.spam_service = spam_svc.SpamService()
self.spam_service.report_tbl = self.mock_report_tbl
self.spam_service.verdict_tbl = self.mock_verdict_tbl
self.spam_service.issue_tbl = self.mock_issue_tbl
self.spam_service.report_tbl.Delete = Mock()
self.spam_service.verdict_tbl.Delete = Mock()
def tearDown(self):
self.testbed.deactivate()
self.mox.UnsetStubs()
self.mox.ResetAll()
def testLookupIssuesFlaggers(self):
self.mock_report_tbl.Select(
self.cnxn, cols=['issue_id', 'user_id', 'comment_id'],
issue_id=[234, 567, 890]).AndReturn([
[234, 111, None],
[234, 222, 1],
[567, 333, None]])
self.mox.ReplayAll()
reporters = (
self.spam_service.LookupIssuesFlaggers(self.cnxn, [234, 567, 890]))
self.mox.VerifyAll()
self.assertEqual({
234: ([111], {1: [222]}),
567: ([333], {}),
}, reporters)
def testLookupIssueFlaggers(self):
self.mock_report_tbl.Select(
self.cnxn, cols=['issue_id', 'user_id', 'comment_id'],
issue_id=[234]).AndReturn(
[[234, 111, None], [234, 222, 1]])
self.mox.ReplayAll()
issue_reporters, comment_reporters = (
self.spam_service.LookupIssueFlaggers(self.cnxn, 234))
self.mox.VerifyAll()
self.assertItemsEqual([111], issue_reporters)
self.assertEqual({1: [222]}, comment_reporters)
def testFlagIssues_overThresh(self):
issue = fake.MakeTestIssue(
project_id=789,
local_id=1,
reporter_id=111,
owner_id=456,
summary='sum',
status='Live',
issue_id=78901,
project_name='proj')
issue.assume_stale = False # We will store this issue.
self.mock_report_tbl.InsertRows(self.cnxn,
['issue_id', 'reported_user_id', 'user_id'],
[(78901, 111, 111)], ignore=True)
self.mock_report_tbl.Select(self.cnxn,
cols=['issue_id', 'COUNT(*)'], group_by=['issue_id'],
issue_id=[78901]).AndReturn([(78901, settings.spam_flag_thresh)])
self.mock_verdict_tbl.Select(
self.cnxn, cols=['issue_id', 'reason', 'MAX(created)'],
group_by=['issue_id'], issue_id=[78901], comment_id=None).AndReturn([])
self.mock_verdict_tbl.InsertRows(
self.cnxn, ['issue_id', 'is_spam', 'reason', 'project_id'],
[(78901, True, 'threshold', 789)], ignore=True)
self.mox.ReplayAll()
self.spam_service.FlagIssues(
self.cnxn, self.issue_service, [issue], 111, True)
self.mox.VerifyAll()
self.assertIn(issue, self.issue_service.updated_issues)
self.assertEqual(
1,
self.spam_service.issue_actions.get(
fields={
'type': 'flag',
'reporter_id': str(111),
'issue': 'proj:1'
}))
def testFlagIssues_underThresh(self):
issue = fake.MakeTestIssue(
project_id=789,
local_id=1,
reporter_id=111,
owner_id=456,
summary='sum',
status='Live',
issue_id=78901,
project_name='proj')
self.mock_report_tbl.InsertRows(self.cnxn,
['issue_id', 'reported_user_id', 'user_id'],
[(78901, 111, 111)], ignore=True)
self.mock_report_tbl.Select(self.cnxn,
cols=['issue_id', 'COUNT(*)'], group_by=['issue_id'],
issue_id=[78901]).AndReturn([(78901, settings.spam_flag_thresh - 1)])
self.mock_verdict_tbl.Select(
self.cnxn, cols=['issue_id', 'reason', 'MAX(created)'],
group_by=['issue_id'], issue_id=[78901], comment_id=None).AndReturn([])
self.mox.ReplayAll()
self.spam_service.FlagIssues(
self.cnxn, self.issue_service, [issue], 111, True)
self.mox.VerifyAll()
self.assertNotIn(issue, self.issue_service.updated_issues)
self.assertIsNone(
self.spam_service.issue_actions.get(
fields={
'type': 'flag',
'reporter_id': str(111),
'issue': 'proj:1'
}))
def testUnflagIssue_overThresh(self):
issue = fake.MakeTestIssue(
project_id=789, local_id=1, reporter_id=111, owner_id=456,
summary='sum', status='Live', issue_id=78901, is_spam=True)
self.mock_report_tbl.Delete(self.cnxn, issue_id=[issue.issue_id],
comment_id=None, user_id=111)
self.mock_report_tbl.Select(self.cnxn,
cols=['issue_id', 'COUNT(*)'], group_by=['issue_id'],
issue_id=[78901]).AndReturn([(78901, settings.spam_flag_thresh)])
self.mock_verdict_tbl.Select(
self.cnxn, cols=['issue_id', 'reason', 'MAX(created)'],
group_by=['issue_id'], issue_id=[78901], comment_id=None).AndReturn([])
self.mox.ReplayAll()
self.spam_service.FlagIssues(
self.cnxn, self.issue_service, [issue], 111, False)
self.mox.VerifyAll()
self.assertNotIn(issue, self.issue_service.updated_issues)
self.assertEqual(True, issue.is_spam)
def testUnflagIssue_underThresh(self):
"""A non-member un-flagging an issue as spam should not be able
to overturn the verdict to ham. This is different from previous
behavior. See https://crbug.com/monorail/2232 for details."""
issue = fake.MakeTestIssue(
project_id=789, local_id=1, reporter_id=111, owner_id=456,
summary='sum', status='Live', issue_id=78901, is_spam=True)
issue.assume_stale = False # We will store this issue.
self.mock_report_tbl.Delete(self.cnxn, issue_id=[issue.issue_id],
comment_id=None, user_id=111)
self.mock_report_tbl.Select(self.cnxn,
cols=['issue_id', 'COUNT(*)'], group_by=['issue_id'],
issue_id=[78901]).AndReturn([(78901, settings.spam_flag_thresh - 1)])
self.mock_verdict_tbl.Select(
self.cnxn, cols=['issue_id', 'reason', 'MAX(created)'],
group_by=['issue_id'], issue_id=[78901], comment_id=None).AndReturn([])
self.mox.ReplayAll()
self.spam_service.FlagIssues(
self.cnxn, self.issue_service, [issue], 111, False)
self.mox.VerifyAll()
self.assertNotIn(issue, self.issue_service.updated_issues)
self.assertEqual(True, issue.is_spam)
def testUnflagIssue_underThreshNoManualOverride(self):
issue = fake.MakeTestIssue(
project_id=789, local_id=1, reporter_id=111, owner_id=456,
summary='sum', status='Live', issue_id=78901, is_spam=True)
self.mock_report_tbl.Delete(self.cnxn, issue_id=[issue.issue_id],
comment_id=None, user_id=111)
self.mock_report_tbl.Select(self.cnxn,
cols=['issue_id', 'COUNT(*)'], group_by=['issue_id'],
issue_id=[78901]).AndReturn([(78901, settings.spam_flag_thresh - 1)])
self.mock_verdict_tbl.Select(
self.cnxn, cols=['issue_id', 'reason', 'MAX(created)'],
group_by=['issue_id'], comment_id=None,
issue_id=[78901]).AndReturn([(78901, 'manual', '')])
self.mox.ReplayAll()
self.spam_service.FlagIssues(
self.cnxn, self.issue_service, [issue], 111, False)
self.mox.VerifyAll()
self.assertNotIn(issue, self.issue_service.updated_issues)
self.assertEqual(True, issue.is_spam)
def testIsExempt_RegularUser(self):
author = user_pb2.MakeUser(111, email='test@example.com')
self.assertFalse(self.spam_service._IsExempt(author, False))
author = user_pb2.MakeUser(111, email='test@chromium.org.example.com')
self.assertFalse(self.spam_service._IsExempt(author, False))
def testIsExempt_ProjectMember(self):
author = user_pb2.MakeUser(111, email='test@example.com')
self.assertTrue(self.spam_service._IsExempt(author, True))
def testIsExempt_AllowlistedDomain(self):
author = user_pb2.MakeUser(111, email='test@google.com')
self.assertTrue(self.spam_service._IsExempt(author, False))
def testClassifyIssue_spam(self):
issue = fake.MakeTestIssue(
project_id=789, local_id=1, reporter_id=111, owner_id=456,
summary='sum', status='Live', issue_id=78901, is_spam=True)
self.spam_service._predict = lambda body: 1.0
# Prevent missing service inits to fail the test.
self.spam_service.ml_engine = True
comment_pb = tracker_pb2.IssueComment()
comment_pb.content = "this is spam"
reporter = user_pb2.MakeUser(111, email='test@test.com')
res = self.spam_service.ClassifyIssue(issue, comment_pb, reporter, False)
self.assertEqual(1.0, res['confidence_is_spam'])
reporter.email = 'test@chromium.org.spam.com'
res = self.spam_service.ClassifyIssue(issue, comment_pb, reporter, False)
self.assertEqual(1.0, res['confidence_is_spam'])
reporter.email = 'test.google.com@test.com'
res = self.spam_service.ClassifyIssue(issue, comment_pb, reporter, False)
self.assertEqual(1.0, res['confidence_is_spam'])
def testClassifyIssue_Allowlisted(self):
issue = fake.MakeTestIssue(
project_id=789, local_id=1, reporter_id=111, owner_id=456,
summary='sum', status='Live', issue_id=78901, is_spam=True)
self.spam_service._predict = assert_unreached
# Prevent missing service inits to fail the test.
self.spam_service.ml_engine = True
comment_pb = tracker_pb2.IssueComment()
comment_pb.content = "this is spam"
reporter = user_pb2.MakeUser(111, email='test@google.com')
res = self.spam_service.ClassifyIssue(issue, comment_pb, reporter, False)
self.assertEqual(0.0, res['confidence_is_spam'])
reporter.email = 'test@chromium.org'
res = self.spam_service.ClassifyIssue(issue, comment_pb, reporter, False)
self.assertEqual(0.0, res['confidence_is_spam'])
def testClassifyComment_spam(self):
self.spam_service._predict = lambda body: 1.0
# Prevent missing service inits to fail the test.
self.spam_service.ml_engine = True
commenter = user_pb2.MakeUser(111, email='test@test.com')
res = self.spam_service.ClassifyComment('this is spam', commenter, False)
self.assertEqual(1.0, res['confidence_is_spam'])
commenter.email = 'test@chromium.org.spam.com'
res = self.spam_service.ClassifyComment('this is spam', commenter, False)
self.assertEqual(1.0, res['confidence_is_spam'])
commenter.email = 'test.google.com@test.com'
res = self.spam_service.ClassifyComment('this is spam', commenter, False)
self.assertEqual(1.0, res['confidence_is_spam'])
def testClassifyComment_Allowlisted(self):
self.spam_service._predict = assert_unreached
# Prevent missing service inits to fail the test.
self.spam_service.ml_engine = True
commenter = user_pb2.MakeUser(111, email='test@google.com')
res = self.spam_service.ClassifyComment('this is spam', commenter, False)
self.assertEqual(0.0, res['confidence_is_spam'])
commenter.email = 'test@chromium.org'
res = self.spam_service.ClassifyComment('this is spam', commenter, False)
self.assertEqual(0.0, res['confidence_is_spam'])
def test_ham_classification(self):
actual = self.spam_service.ham_classification()
self.assertEqual(actual['confidence_is_spam'], 0.0)
self.assertEqual(actual['failed_open'], False)
def testExpungeUsersInSpam(self):
user_ids = [3, 4, 5]
self.spam_service.ExpungeUsersInSpam(self.cnxn, user_ids=user_ids)
self.spam_service.report_tbl.Delete.assert_has_calls(
[
mock.call(self.cnxn, reported_user_id=user_ids, commit=False),
mock.call(self.cnxn, user_id=user_ids, commit=False)
])
self.spam_service.verdict_tbl.Delete.assert_called_once_with(
self.cnxn, user_id=user_ids, commit=False)
def testLookupIssueVerdicts(self):
self.spam_service.verdict_tbl.Select = Mock(return_value=[
[5, 10], [4, 11], [6, 12],
])
actual = self.spam_service.LookupIssueVerdicts(self.cnxn, [4, 5, 6])
self.spam_service.verdict_tbl.Select.assert_called_once_with(
self.cnxn, cols=['issue_id', 'reason', 'MAX(created)'],
issue_id=[4, 5, 6], comment_id=None, group_by=['issue_id'])
self.assertEqual(actual, {
5: 10,
4: 11,
6: 12,
})