python类unified_diff()的实例源码

def assert_equal_diff(self, received, expected):
        if received != expected:
            diff_string = '\n'.join(difflib.unified_diff(expected.split('\n'),
                                                         received.split('\n'),
                                                         'EXPECTED', 'RECEIVED',
                                                         lineterm=''))
            self.fail(diff_string)

def assert_startswith_diff(self, received, prefix):
        if not received.startswith(prefix):
            prefix_lines = prefix.split('\n')
            diff_string = '\n'.join(difflib.unified_diff(prefix_lines,
                                                         received.split('\n')[:len(prefix_lines)],
                                                         'EXPECTED', 'RECEIVED',
                                                         lineterm=''))
            self.fail(diff_string)

def get_diff(self, article_path, a, b):
        """Return a diff string between two revisions of a given
        article title.
        """
        revision_a = self.get_revision(article_path, a)
        revision_b = self.get_revision(article_path, b)

        unified_diff = '\n'.join(
            list(
                difflib.unified_diff(
                    revision_a['raw'].splitlines(),
                    revision_b['raw'].splitlines(),
                    fromfile='{}/{}'.format('a', article_path),
                    tofile='{}/{}'.format('b', article_path),
                    lineterm='',
                )
            )
        )

        diff_template = """diff --git a/{title} b/{title}
index {sha_a}..{sha_b} {file_mode}
{diff}
"""

        unified_diff = diff_template.format(
            title=article_path,
            diff=unified_diff,
            sha_a=a[0:7],
            sha_b=b[0:7],
            file_mode=oct(
                os.stat(self.full_article_path(article_path)).st_mode
            )[2:]
        )

        # Escape HTML and "non-breaking space"
        return self.escape_html(unified_diff)

def _diff_monitors(self, monitor1, monitor1_name, monitor2, monitor2_name):
        """
        Convenience method for performing a diff of two monitors, returning the diff as a string.
        """
        return "\n".join(difflib.unified_diff(
            json.dumps(monitor1, indent=4, sort_keys=True).splitlines(),
            json.dumps(monitor2, indent=4, sort_keys=True).splitlines(),
            fromfile=monitor1_name,
            tofile=monitor2_name
        ))

def diff(args, config, connection=None):
    verbose = args.verbose
    last_schema = get_last_schema()

    current_schema = get_current_schema()

    added, removed, changed = compare_schemas(last_schema, current_schema)

    if any((added, removed, changed)):
        if added:
            print('New items:')
            for item in added:
                print('  %s' % item)

        if removed:
            print('Removed items:')
            for item in removed:
                print('  %s' % item)

        if changed:
            print('Changed items:')
            for item in changed:
                print('  %s' % item)
                if verbose:
                    _diff = difflib.unified_diff(last_schema[item]['up'],
                                                 current_schema[item]['up'])
                    print('\n'.join(_diff))

    else:
        print('No changes')

def _show_diff(self, file_contents):
        for line in unified_diff(
            file_contents.splitlines(1),
            self.output.splitlines(1),
            fromfile=self.file_path + ':before',
            tofile=self.file_path + ':after',
            fromfiledate=str(datetime.fromtimestamp(os.path.getmtime(self.file_path))
                             if self.file_path else datetime.now()),
            tofiledate=str(datetime.now())
        ):
            sys.stdout.write(line)

def diff(self):
        texts = []
        for old_path, (new_path, old_l, new_l) in self.change_dct.items():
            if old_path:
                udiff = difflib.unified_diff(old_l, new_l)
            else:
                udiff = difflib.unified_diff(old_l, new_l, old_path, new_path)
            texts.append('\n'.join(udiff))
        return '\n'.join(texts)

def diff(self):
        texts = []
        for old_path, (new_path, old_l, new_l) in self.change_dct.items():
            if old_path:
                udiff = difflib.unified_diff(old_l, new_l)
            else:
                udiff = difflib.unified_diff(old_l, new_l, old_path, new_path)
            texts.append('\n'.join(udiff))
        return '\n'.join(texts)

def indent_files(files, diff=False, debug=False, level=0, inplace=False):
    output = []
    for f in files:
        dst = indent(f, debug=debug, level=level)
        output.append([f,dst])

    # First, copy to inplace
    if inplace:
        for src,dst in output:
            shutil.copyfile(dst,src)
        return True

    # now, compare
    failed = []
    for src,dst in output:
        if filecmp.cmp(src,dst) == 0:
            failed.append([src, dst])
    if len(failed) > 0:
        if not diff:
            print 'Found %u badly indented files:' % len(failed)
            for src,dst in failed:
                print '  ' + src
        else:
            for src,dst in failed:
                s = open(src, 'r').readlines()
                d = open(dst, 'r').readlines()
                for line in difflib.unified_diff(s, d, fromfile=src, tofile=dst):
                    sys.stdout.write(line)
        return False
    return True

def _assertExpectedWithOutput(self, name):
        filename = name + '.conf'
        expected_path = os.path.join(self.expected, filename)
        test_path = os.path.join(self.out_dir, filename)
        self.assertTrue(os.path.exists(expected_path))
        self.assertTrue(os.path.exists(test_path))

        with io.open(expected_path, encoding='utf8') as expected_file:
            with io.open(test_path, encoding='utf8') as test_file:
                expected_data = expected_file.readlines()
                test_data = test_file.readlines()
                diff = difflib.unified_diff(
                    expected_data, test_data, lineterm='')
                diff_data = ''.join(list(diff))
                self.assertTrue(diff_data == '', msg=diff_data)

def diffTwoFiles( self, title, old_lines, new_lines, header_left, header_right ):
        if self.app.prefs.view.isDiffUnified():
            all_lines = list( difflib.unified_diff( old_lines, new_lines ) )

            self.showDiffText( title, all_lines )

        elif self.app.prefs.view.isDiffSideBySide():
            window = wb_diff_side_by_side_view.DiffSideBySideView(
                        self.app, None,
                        title,
                        old_lines, header_left,
                        new_lines, header_right )
            window.show()

def _print_diff(self, diff, indent_level):
        if isinstance(diff, dict):
            try:
                diff = '\n'.join(difflib.unified_diff(diff['before'].splitlines(),
                                                      diff['after'].splitlines(),
                                                      fromfile=diff.get('before_header',
                                                                        'new_file'),
                                                      tofile=diff['after_header']))
            except AttributeError:
                diff = dict_diff(diff['before'], diff['after'])
        if diff:
            diff = colorize(diff, 'changed')
            print(self._indent_text(diff, indent_level+4))

def test_all_correct_contents(self):
    """Tests that the content of all non-static output files is expected."""
    for dirname, _, filenames in os.walk(TTS_REFERENCE_PATH):
      for filename in filenames:
        if filename == caterpillar.SW_SCRIPT_NAME:
          # Service worker is partly random, so test it elsewhere.
          continue

        reference_path = os.path.join(dirname, filename)
        relpath = os.path.relpath(reference_path, TTS_REFERENCE_PATH)

        if not (relpath.startswith(self.boilerplate_dir)
                or relpath.startswith(self.report_dir)):
          output_path = os.path.join(self.output_dir, relpath)
          with open(reference_path) as reference_file:
            with open(output_path) as output_file:
              output_data = output_file.read().decode('utf-8')
              reference_data = reference_file.read().decode('utf-8')
              self.assertEqual(output_data, reference_data,
                  'Difference found in file `{}`.\n{}'.format(relpath,
                      '\n'.join(difflib.unified_diff(
                          output_data.split('\n'),
                          reference_data.split('\n'),
                          fromfile=reference_path,
                          tofile=output_path,
                          n=0))))

def test_generated_service_worker(self):
    """Tests that the generated service worker is as expected."""
    output_sw_path = os.path.join(self.output_dir, caterpillar.SW_SCRIPT_NAME)
    reference_sw_path = os.path.join(
        TTS_REFERENCE_PATH, caterpillar.SW_SCRIPT_NAME)

    with open(output_sw_path) as output_file:
      with open(reference_sw_path) as reference_file:
        output_data = output_file.read().decode('utf-8')
        reference_data = reference_file.read().decode('utf-8')

        # The cache version is randomly generated in the output service worker,
        # so reset it to be 0, the same cache version as the reference service
        # worker.
        output_data = re.sub(
            r'CACHE_VERSION = \d+', 'CACHE_VERSION = 0', output_data)

        self.assertEqual(output_data, reference_data,
            'Difference found in file `{}`.\n{}'.format(
                caterpillar.SW_SCRIPT_NAME,
                '\n'.join(difflib.unified_diff(
                    output_data.split('\n'),
                    reference_data.split('\n'),
                    fromfile=reference_sw_path,
                    tofile=output_sw_path,
                    n=0))))

def _handle_existing_policy(name, new_rules, existing_rules):
  ret = { 'name': name }
  if new_rules == existing_rules:
    ret['result'] = True
    ret['changes'] = None
    ret['comment'] = 'Policy exists, and has the correct content'
    return ret

  change = ''.join(difflib.unified_diff(existing_rules.splitlines(True), new_rules.splitlines(True)))
  if __opts__['test']:
    ret['result'] = None
    ret['changes'] = { name: { 'change': change } }
    ret['comment'] = 'Policy would be changed'
    return ret

  payload = { 'rules': new_rules }

  url = "v1/sys/policy/{0}".format(name)
  response = __utils__['vault.make_request']('PUT', url, json=payload)
  if response.status_code != 204:
    return {
      'name': name,
      'changes': None,
      'result': False,
      'comment': 'Failed to change policy: {0}'.format(response.reason)
    }

  ret['result'] = True
  ret['changes'] = { name: { 'change': change } }
  ret['comment'] = 'Policy was updated'

  return ret

def clobber_if_necessary(new_landmines, src_dir):
  """Does the work of setting, planting, and triggering landmines."""
  out_dir = get_build_dir(src_dir)
  landmines_path = os.path.normpath(os.path.join(src_dir, '.landmines'))
  try:
    os.makedirs(out_dir)
  except OSError as e:
    if e.errno == errno.EEXIST:
      pass

  if os.path.exists(landmines_path):
    with open(landmines_path, 'r') as f:
      old_landmines = f.readlines()
    if old_landmines != new_landmines:
      old_date = time.ctime(os.stat(landmines_path).st_ctime)
      diff = difflib.unified_diff(old_landmines, new_landmines,
          fromfile='old_landmines', tofile='new_landmines',
          fromfiledate=old_date, tofiledate=time.ctime(), n=0)
      sys.stdout.write('Clobbering due to:\n')
      sys.stdout.writelines(diff)
      sys.stdout.flush()

      clobber.clobber(out_dir)

  # Save current set of landmines for next time.
  with open(landmines_path, 'w') as f:
    f.writelines(new_landmines)

def apply_deployment(project, template):
    body = {
        'name': template.name,
        'description': 'project: {}, name: {}'.format(project, template.name),
        'target': {
            'config': {
                'content': str(template)
            }
        }
    }
    result = None
    try:
        deployment = get_deployment(project, template.name)
        if deployment:
            logging.info('Deployment already exists. Getting changes for {}...'.format(template.name))
            body['fingerprint'] = deployment.get('fingerprint')
            changed = False
            existing_template = get_manifest(project, deployment)['config']['content']
            for diff in color_diff(difflib.unified_diff(existing_template.splitlines(),
                                                        str(template).splitlines(),
                                                        fromfile='Existing template', tofile='Proposed template')):
                changed = True
                print(diff)
            if changed and confirm_action():
                result = dm.deployments().update(project=project, deployment=template.name, body=body).execute()
            else:
                logging.info('No changes in the template.')
                sys.exit(0)
        else:
            logging.info('Generated template:\n{}\n'.format(template))
            logging.info('Launching a new deployment: {}'.format(template.name))
            if confirm_action():
                result = dm.deployments().insert(project=project, body=body).execute()
    except errors.HttpError as e:
        raise e
    if result:
        return wait_for_completion(project, result)

def unified_diff(gold_path, current_path):
    gold = json.dumps(json.load(open(gold_path, "r")), indent=4, sort_keys=True)
    current = json.dumps(json.load(open(current_path, "r")), indent=4, sort_keys=True)

    udiff = list(difflib.unified_diff(gold.split("\n"), current.split("\n"),
                                      fromfile=os.path.basename(gold_path),
                                      tofile=os.path.basename(current_path),
                                      lineterm=""))

    return udiff

#### Test functions

def fetch(self):
        urlp = urllib.parse.urlparse(self.url)
        lastupd = self.status.load()
        old_entries = lastupd.get('entries')
        fetch_time = int(time.time())
        with ftputil.FTPHost(urlp.hostname, urlp.username or 'anonymous', urlp.password) as host:
            try:
                st_mtime = host.lstat(urlp.path.rstrip('/')).st_mtime
            except ftputil.error.RootDirError:
                st_mtime = None
            if st_mtime and st_mtime == lastupd.get('mtime'):
                return
            else:
                lastupd['mtime'] = st_mtime
                entries = sorted(x for x in host.listdir(urlp.path) if
                                 (not self.regex or self.regex.search(x)))
                lastupd['entries'] = entries
        self.status = self.status.save(fetch_time, lastupd)
        if not old_entries or entries == old_entries:
            return
        else:
            diff = tuple(difflib.unified_diff(old_entries, entries, lineterm=''))
            title = 'FTP directory changed'
            for text in diff[2:]:
                if text[0] == '+':
                    title = text[1:]
                    break
            content = (markupsafe.Markup('<pre>%s</pre>') % '\n'.join(diff[2:]))
        yield Event(self.name, self.category,
                    lastupd.get('mtime') or fetch_time, title, content, self.url)

def check_path(self, path, files, zip_data):
        logging.info("path={0}".format(path))
        tally = defaultdict(int)
        for filename in files:
            if path is None or path == '':
                testfile_path = os.path.abspath(os.path.join(self.ref_dir, filename))
                testfile_key = filename
            else:
                testfile_path = os.path.abspath(os.path.join(self.ref_dir, path, filename))
                testfile_key = os.path.join(path, filename)

            # set up score value for matching output correctly
            score = self.default_score
            if path in self.path_score:
                score = self.path_score[path]

            logging.info("Checking {0}".format(testfile_key))
            if testfile_key in zip_data:
                with open(testfile_path, 'r') as ref:
                    ref_data = map(lambda x: x.strip(), ref.read().splitlines())
                    output_data = map(lambda x: x.strip(), zip_data[testfile_key].splitlines())
                    diff_lines = list(difflib.unified_diff(ref_data, output_data, "reference", "your-output", lineterm=''))
                    if len(diff_lines) > 0:
                        logging.info("Diff between reference and your output for {0}".format(testfile_key))
                        logging.info("{0}{1}".format(self.linesep, self.linesep.join(list(diff_lines))))
                    else:
                        tally['score'] += score
                        tally['num_correct'] += 1
                        logging.info("{0} Correct!".format(testfile_key))
                    tally['total'] += 1
        self.perf[path] = dict(tally)