[Findit] Reorganizing findit_for_*.py

appengine/findit/crash/test/findit_test.py

+# 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.
+
+import copy
+import logging
+
+from google.appengine.api import app_identity
+
+from crash import crash_pipeline
+from crash import findit
+from crash import findit_for_chromecrash
+from crash.stacktrace import Stacktrace
+from crash.crash_report import CrashReport
+from crash.findit_for_chromecrash import FinditForChromeCrash
+from crash.findit_for_chromecrash import FinditForFracas
+from crash.test.crash_testcase import CrashTestCase
+from crash.type_enums import CrashClient
+from model.crash.fracas_crash_analysis import FracasCrashAnalysis
+
+class UnsupportedClient(findit.Findit):
+  @property
+  def client_id(self):
+    return self._client_id
+
+  def __init__(self, client_id=None):
+    super(UnsupportedClient, self).__init__()
+    if client_id is None:
+      client_id = 'unsupported_client'
+    self._client_id = client_id
+
+
+# TODO(wrengr): Our use of this highly suggests that |crash_identifiers|
+# don't actually need to be passed around everywhere. The only parts that
+# matter are the |channel| and the |process_type|, which can be passed
+# separately.
+def CrashIdentifiers(report):
+  return {
+      'chrome_version': report.crashed_version,
+      'signature': report.signature,
+      'channel': 'canary',
+      'platform': report.platform,
+      'process_type': 'browser'
+    }
+
+DUMMY_REPORT1 = CrashReport(
+    crashed_version = '1',
+    signature = 'signature',
+    platform = 'win',
+    stacktrace = Stacktrace(),
+    regression_range = None)
+
+class FindTest(CrashTestCase):
+
+  def testCheckPolicyUnsupportedClient(self):
+    crash_identifiers = {'signature': 'sig'}
+    report = DUMMY_REPORT1._replace(
+        platform = 'canary',
+        signature = crash_identifiers['signature'])
+    channel = 'canary'
+    new_report = UnsupportedClient().CheckPolicy(crash_identifiers, report, channel)
+    self.assertIsNone(new_report)
+
+  def testCheckPolicyUnsupportedPlatform(self):
+    report = DUMMY_REPORT1._replace(
+        platform = 'unsupported_platform')
+    crash_identifiers = CrashIdentifiers(report)
+    channel = crash_identifiers['channel']
+
+    new_report = FinditForFracas().CheckPolicy(crash_identifiers, report, channel)
+    self.assertIsNone(new_report)
+
+  def testCheckPolicyBlacklistedSignature(self):
+    report = DUMMY_REPORT1._replace(
+        signature = 'Blacklist marker signature')
+    crash_identifiers = CrashIdentifiers(report)
+    channel = crash_identifiers['channel']
+
+    new_report = FinditForFracas().CheckPolicy(crash_identifiers, report, channel)
+    self.assertIsNone(new_report)
+
+  def testCheckPolicyPlatformRename(self):
+    report = DUMMY_REPORT1._replace(
+        platform = 'linux')
+    crash_identifiers = CrashIdentifiers(report)
+    channel = crash_identifiers['channel']
+
+    new_report = FinditForFracas().CheckPolicy(crash_identifiers, report, channel)
+    self.assertIsNotNone(new_report)
+    self.assertEqual(new_report.platform, 'unix')
+
+  def testGetAnalysisForClient(self):
+    crash_identifiers = {'signature': 'sig'}
+    analysis = FracasCrashAnalysis.Create(crash_identifiers)
+    analysis.put()
+
+    self.assertEqual(FinditForFracas().GetAnalysis(crash_identifiers), analysis)
+
+  def testGetAnalysisForUnsuportedClient(self):
+    crash_identifiers = {'signature': 'sig'}
+    analysis = FracasCrashAnalysis.Create(crash_identifiers)
+    analysis.put()
+
+    self.assertIsNone(UnsupportedClient('Unsupported_client').GetAnalysis(crash_identifiers), analysis)
+
+# TODO(wrengr): figure out why this doesn't work anymore.
+#  def testResetAnalysisForFracas(self):
+#    report = DUMMY_REPORT1._replace(
+#        platform = 'linux')
+#    crash_identifiers = CrashIdentifiers(report)
+#    channel = crash_identifiers['channel']
+#
+#    analysis = FracasCrashAnalysis.Create(crash_identifiers)
+#
+#    # TODO(wrengr): the old version also reset |channel|, but
+#    # I think that's a no-op here. Did the old version also set
+#    # |historical_metadata|?
+#    FinditForFracas().ResetAnalysis(analysis, report)
+#    analysis.channel = channel
+#
+#    logging.info(str(report))
+#    logging.info(str(crash_identifiers))
+#    analysis = FracasCrashAnalysis.Get(crash_identifiers)
+#    self.assertIsNotNone(analysis)
+#    self.assertEqual(analysis.crashed_version, chrome_version)
+#    self.assertEqual(analysis.signature, signature)
+#    self.assertEqual(analysis.platform, platform)
+#    self.assertEqual(analysis.stack_trace, stack_trace) # TODO(wrengr): this is going to fail because of type mismatch between str and Stacktrace
+#    self.assertEqual(analysis.channel, channel)
+
+  def testCreateAnalysisForClient(self):
+    crash_identifiers = {'signature': 'sig'}
+    self.assertIsNotNone(FinditForFracas().CreateAnalysis(crash_identifiers))
+
+  def testCreateAnalysisForUnsupportedClientId(self):
+    crash_identifiers = {'signature': 'sig'}
+    self.assertIsNone(UnsupportedClient('unsupported_id').CreateAnalysis(crash_identifiers))
+
+  # TODO(wrengr): move this to crash_pipeline_test.py
+  def testGetPublishResultFromAnalysisFoundTrue(self):
+    mock_host = 'https://host.com'
+    self.mock(app_identity, 'get_default_version_hostname', lambda: mock_host)
+
+    analysis_result = {
+        'found': True,
+        'suspected_cls': [
+            {'confidence': 0.21434,
+             'reason': ['reason1', 'reason2'],
+             'other': 'data'}
+        ],
+        'other_data': 'data',
+    }
+
+    processed_analysis_result = copy.deepcopy(analysis_result)
+    processed_analysis_result['feedback_url'] = (
+        mock_host + '/crash/fracas-result-feedback?'
+        'key=agx0ZXN0YmVkLXRlc3RyQQsSE0ZyYWNhc0NyYXNoQW5hbHlzaXMiKDMzNTY5MDU3'
+        'M2ZlYTFlZGZhMjViOTVjZmI4OGZhODFlNDk0YTEyODkM')
+
+    for cl in processed_analysis_result['suspected_cls']:
+      cl['confidence'] = round(cl['confidence'], 2)
+      cl.pop('reason', None)
+
+    crash_identifiers = {'signature': 'sig'}
+    expected_messages_data = {
+        'crash_identifiers': crash_identifiers,
+        'client_id': CrashClient.FRACAS,
+        'result': processed_analysis_result,
+    }
+
+    analysis = FracasCrashAnalysis.Create(crash_identifiers)
+    analysis.client_id = CrashClient.FRACAS
+    analysis.result = analysis_result
+
+    self.assertEqual(crash_pipeline.GetPublishResultFromAnalysis(
+        analysis, crash_identifiers), expected_messages_data)
+
+  # TODO(wrengr): move this to crash_pipeline_test.py
+  def testGetPublishResultFromAnalysisFoundFalse(self):
+    mock_host = 'https://host.com'
+    self.mock(app_identity, 'get_default_version_hostname', lambda: mock_host)
+
+    analysis_result = {
+        'found': False,
+    }
+
+    processed_analysis_result = copy.deepcopy(analysis_result)
+    processed_analysis_result['feedback_url'] = (
+        mock_host + '/crash/fracas-result-feedback?'
+        'key=agx0ZXN0YmVkLXRlc3RyQQsSE0ZyYWNhc0NyYXNoQW5hbHlzaXMiKDMzNTY5MDU3'
+        'M2ZlYTFlZGZhMjViOTVjZmI4OGZhODFlNDk0YTEyODkM')
+
+    crash_identifiers = {'signature': 'sig'}
+    expected_messages_data = {
+            'crash_identifiers': crash_identifiers,
+            'client_id': CrashClient.FRACAS,
+            'result': processed_analysis_result,
+    }
+
+    analysis = FracasCrashAnalysis.Create(crash_identifiers)
+    analysis.client_id = CrashClient.FRACAS
+    analysis.result = analysis_result
+
+    self.assertEqual(crash_pipeline.GetPublishResultFromAnalysis(
+        analysis, crash_identifiers), expected_messages_data)
+
+  def testFindCulprit(self):
+    expected_result = {'found': False}
+    expected_tags = {'found_suspects': False,
+                     'has_regression_range': False}
+
+    class _MockFinditForChromeCrash(FinditForChromeCrash):
+      def __init__(self, *_):
+        # TODO(wrengr): this is looping for some reason...
+        #super(_MockFinditForChromeCrash, self).__init__()
+        pass
+      def FindCulprit(self, *_):
+        # TODO(wrengr): this has to return a Culprit object
+        return expected_result, expected_tags
+    self.mock(findit_for_chromecrash, 'FinditForChromeCrash',
+              _MockFinditForChromeCrash)
+
+    analysis = FracasCrashAnalysis.Create({'signature': 'sig'})
+    analysis.client_id = CrashClient.FRACAS
+
+    result, tags = _MockFinditForChromeCrash().FindCulprit(analysis)
+    self.assertEqual(result, expected_result)
+    self.assertEqual(tags, expected_tags)
+