<commit_before># Copyright 2019 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.

import collections
import difflib
import json
import multiprocessing
import os
import re
import sys
import tempfile
import traceback

import coverage

from gevent import subprocess
import gevent

from google.protobuf import json_format as jsonpb

from recipe_engine import __path__ as RECIPE_ENGINE_PATH

# pylint: disable=import-error
import PB
from PB.recipe_engine.internal.test.runner import Description, Outcome

from .... import config_types
from .... import types

from ...simple_cfg import RECIPES_CFG_LOCATION_REL
from ...test import magic_check_fn
from ...test.execute_test_case import execute_test_case

from .expectation_conversion import transform_exepctations
from .pipe import write_message, read_message


def _merge_presentation_updates(steps_ran, presentation_steps):
  """Merges the steps ran (from the SimulationStepRunner) with the steps
  presented (from the SimulationAnnotatorStreamEngine).

  Args:

    * steps_ran (Dict[str, dict]) - Mapping of step name to its run details as
      an expectation dict (e.g. 'cmd', 'env', etc.)
    * presentation_steps (OrderedDict[str, dict]) - Mapping of step name (in the
      order that they were presented) to a dict containing the collected
      annotations for that step.

  Returns OrderedDict[str, expectation: dict]. This will have the order of steps
  in the order that they were presented.
  """
  ret = collections.OrderedDict()
  for step_name, step_presented in presentation_steps.iteritems():
    # root annotations
    if step_name is None:
      continue
    ret[step_name] = steps_ran.get(step_name, {
      'name': step_name,
      # TODO(iannucci): Drop 'cmd' field for presentation-only steps.
      'cmd': [],
      'cost': None,
    })
    debug_logs = step_presented.get('logs', {}).get('$debug', None)
    # wowo hacks!
    # We only want to see $debug if it's got a crash in it.
    if debug_logs and 'Unhandled exception:' not in debug_logs.splitlines():
      step_presented['logs'].pop('$debug')

    ret[step_name].update(step_presented)

  return ret


def _check_bad_test(test_results, test_data, steps_ran, presentation_steps):
  """Check to see if the user-provided test was malformed in some way.

  Currently this only identifies issues around unconsumed or misplaced
  step_data.

  Args:

    * test_results (Outcome.Results) - The object to update in the event the
      test was bad.
    * test_data (recipe_engine.recipe_test_api.TestData) - The user-provided
      test data object, after running the test. We're checking to see that it's
      empty now.
    * steps_ran (List[str]) - The list of step names which the
      SimulationStepRunner saw. This will only include step names run via
      `api.step()`, and notably omits 'presentation only' steps such as parent
      nest steps or steps emitted by the engine for UI purposes (e.g. crash
      reports).
    * presentation_steps (List[str]) - The list of step names which the
      SimulationAnnotatorStreamEngine saw. This is the full list of steps which
      would occur on the build UI.
  """
  for step in test_data.step_data:
    # This is an unconsumed step name.

    if step in presentation_steps:
      # If the step is unconsumed but present in presentation_steps it means
      # that the step was really a presentation-only step (like a parent nesting
      # step) and not eligble for test data.
      test_results.bad_test.extend([
        'Mock data provided for presentation only step %r.' % step,
        'Presentation-only steps (like parent nesting steps) have no',
        'subprocess associated with them and so cannot have mock data.',
        'Please change your test so that it provides mock data for one of',
        'the real steps.',
      ])

    else:
      test_results.bad_test.append(
          'Mock data provided for non-existent step %r.' % step)

  if test_results.bad_test:
    test_results.bad_test.extend([
        '',
        'For reference, this test ran the following steps:',
    ])
    test_results.bad_test.extend('  ' + repr(s) for s in steps_ran)


def _check_exception(test_results, expected_exception, uncaught_exception_info):
  """Check to see if the test run failed with an exception from RunSteps.

  Args:

    * test_results (Outcome.Results) - The Outcome object to update in
      the event the test exception expectation was bad.
    * expected_exception (str|None) - The name of the exception that the test
      case expected.
    * uncaught_exception_info (Tuple[type, Exception, traceback]|None) - The
      exception info for any uncaught exception triggered by user recipe code.

  Returns CrashFailure|None.
  """
  # Check to see if the user expected the recipe to crash in this test case or
  # not.
  if uncaught_exception_info:
    exc_type, exc, tback = uncaught_exception_info
    exc_name = exc_type.__name__
  else:
    exc_name = exc = None
  if expected_exception:
    if not exc:
      test_results.crash_mismatch.append(
        'Expected exception mismatch in RunSteps. The test expected %r but '
        'the exception line was %r.' % (expected_exception, exc_name)
      )

    elif exc_name != expected_exception:<commit_msg>Report run steps in the order they were run.<commit_after># Copyright 2019 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.

import collections
import difflib
import json
import multiprocessing
import os
import re
import sys
import tempfile
import traceback

import coverage

from gevent import subprocess
import gevent

from google.protobuf import json_format as jsonpb

from recipe_engine import __path__ as RECIPE_ENGINE_PATH

# pylint: disable=import-error
import PB
from PB.recipe_engine.internal.test.runner import Description, Outcome

from .... import config_types
from .... import types

from ...simple_cfg import RECIPES_CFG_LOCATION_REL
from ...test import magic_check_fn
from ...test.execute_test_case import execute_test_case

from .expectation_conversion import transform_exepctations
from .pipe import write_message, read_message


def _merge_presentation_updates(steps_ran, presentation_steps):
  """Merges the steps ran (from the SimulationStepRunner) with the steps
  presented (from the SimulationAnnotatorStreamEngine).

  Args:

    * steps_ran (Dict[str, dict]) - Mapping of step name to its run details as
      an expectation dict (e.g. 'cmd', 'env', etc.)
    * presentation_steps (OrderedDict[str, dict]) - Mapping of step name (in the
      order that they were presented) to a dict containing the collected
      annotations for that step.

  Returns OrderedDict[str, expectation: dict]. This will have the order of steps
  in the order that they were presented.
  """
  ret = collections.OrderedDict()
  for step_name, step_presented in presentation_steps.iteritems():
    # root annotations
    if step_name is None:
      continue
    ret[step_name] = steps_ran.get(step_name, {
      'name': step_name,
      # TODO(iannucci): Drop 'cmd' field for presentation-only steps.
      'cmd': [],
      'cost': None,
    })
    debug_logs = step_presented.get('logs', {}).get('$debug', None)
    # wowo hacks!
    # We only want to see $debug if it's got a crash in it.
    if debug_logs and 'Unhandled exception:' not in debug_logs.splitlines():
      step_presented['logs'].pop('$debug')

    ret[step_name].update(step_presented)

  return ret


def _check_bad_test(test_results, test_data, steps_ran, presentation_steps):
  """Check to see if the user-provided test was malformed in some way.

  Currently this only identifies issues around unconsumed or misplaced
  step_data.

  Args:

    * test_results (Outcome.Results) - The object to update in the event the
      test was bad.
    * test_data (recipe_engine.recipe_test_api.TestData) - The user-provided
      test data object, after running the test. We're checking to see that it's
      empty now.
    * steps_ran (List[str]) - The list of step names which the
      SimulationStepRunner saw. This will only include step names run via
      `api.step()`, and notably omits 'presentation only' steps such as parent
      nest steps or steps emitted by the engine for UI purposes (e.g. crash
      reports).
    * presentation_steps (List[str]) - The list of step names which the
      SimulationAnnotatorStreamEngine saw. This is the full list of steps which
      would occur on the build UI.
  """
  for step in test_data.step_data:
    # This is an unconsumed step name.

    if step in presentation_steps:
      # If the step is unconsumed but present in presentation_steps it means
      # that the step was really a presentation-only step (like a parent nesting
      # step) and not eligble for test data.
      test_results.bad_test.extend([
        'Mock data provided for presentation only step %r.' % step,
        'Presentation-only steps (like parent nesting steps) have no',
        'subprocess associated with them and so cannot have mock data.',
        'Please change your test so that it provides mock data for one of',
        'the real steps.',
      ])

    else:
      test_results.bad_test.append(
          'Mock data provided for non-existent step %r.' % step)

  if test_results.bad_test:
    test_results.bad_test.extend([
        '',
        'For reference, this test ran the following steps:',
    ])
    test_results.bad_test.extend(
        '  ' + repr(s) for s in presentation_steps if s in steps_ran)


def _check_exception(test_results, expected_exception, uncaught_exception_info):
  """Check to see if the test run failed with an exception from RunSteps.

  Args:

    * test_results (Outcome.Results) - The Outcome object to update in
      the event the test exception expectation was bad.
    * expected_exception (str|None) - The name of the exception that the test
      case expected.
    * uncaught_exception_info (Tuple[type, Exception, traceback]|None) - The
      exception info for any uncaught exception triggered by user recipe code.

  Returns CrashFailure|None.
  """
  # Check to see if the user expected the recipe to crash in this test case or
  # not.
  if uncaught_exception_info:
    exc_type, exc, tback = uncaught_exception_info
    exc_name = exc_type.__name__
  else:
    exc_name = exc = None
  if expected_exception:
    if not exc:
      test_results.crash_mismatch.append(
        'Expected exception mismatch in RunSteps. The test expected %r but '
        'the exception line was %r.' % (expected_exception, exc_name)
      )

    elif exc_name != expected_exception:
{'fraction_non_alphanumeric': 0.07180580005559159, 'fraction_numerical': 0.0011118317427962568, 'mean_word_length': 3.389589263928426, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '406050', 'n_tokens_mistral': 3116, 'n_tokens_neox': 3015, 'n_words': 1221}