parameters: (self, stream)
return_statement: return (self.errors == 0)
docstring: parse unified diff
          return True on success
docstring_summary: parse unified diff
          return True on success
docstring_tokens: parse unified diff return True on success
function: def parse(self, stream):
      """ parse unified diff
          return True on success
      """
      lineends = dict(lf=0, crlf=0, cr=0)
      nexthunkno = 0    #: even if index starts with 0 user messages number hunks from 1
  
      p = None
      hunk = None
      # hunkactual variable is used to calculate hunk lines for comparison
      hunkactual = dict(linessrc=None, linestgt=None)
  
  
      class wrapumerate(enumerate):
        """Enumerate wrapper that uses boolean end of stream status instead of
        StopIteration exception, and properties to access line information.
        """
  
        def __init__(self, *args, **kwargs):
          # we don't call parent, it is magically created by __new__ method
  
          self._exhausted = False
          self._lineno = False     # after end of stream equal to the num of lines
          self._line = False       # will be reset to False after end of stream
  
        def next(self):
          """Try to read the next line and return True if it is available,
             False if end of stream is reached."""
          if self._exhausted:
            return False
  
          try:
            self._lineno, self._line = compat_next(super(wrapumerate, self))
          except StopIteration:
            self._exhausted = True
            self._line = False
            return False
          return True
  
        @property
        def is_empty(self):
          return self._exhausted
  
        @property
        def line(self):
          return self._line
  
        @property
        def lineno(self):
          return self._lineno
  
      # define states (possible file regions) that direct parse flow
      headscan  = True  # start with scanning header
      filenames = False # lines starting with --- and +++
  
      hunkhead = False  # @@ -R +R @@ sequence
      hunkbody = False  #
      hunkskip = False  # skipping invalid hunk mode
  
      hunkparsed = False # state after successfully parsed hunk
  
      # regexp to match start of hunk, used groups - 1,3,4,6
      re_hunk_start = re.compile(b"^@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@")
  
      self.errors = 0
      # temp buffers for header and filenames info
      header = []
      srcname = None
      tgtname = None
  
      # start of main cycle
      # each parsing block already has line available in fe.line
      fe = wrapumerate(stream)
      while fe.next():
  
        # -- deciders: these only switch state to decide who should process
        # --           line fetched at the start of this cycle
        if hunkparsed:
          hunkparsed = False
          if re_hunk_start.match(fe.line):
              hunkhead = True
          elif fe.line.startswith(b"--- "):
              filenames = True
          else:
              headscan = True
        # -- ------------------------------------
  
        # read out header
        if headscan:
          while not fe.is_empty and not fe.line.startswith(b"--- "):
              header.append(fe.line)
              fe.next()
          if fe.is_empty:
              if p is None:
                debug("no patch data found")  # error is shown later
                self.errors += 1
              else:
                info("%d unparsed bytes left at the end of stream" % len(b''.join(header)))
                self.warnings += 1
                # TODO check for \No new line at the end..
                # TODO test for unparsed bytes
                # otherwise error += 1
              # this is actually a loop exit
              continue
  
          headscan = False
          # switch to filenames state
          filenames = True
  
        line = fe.line
        lineno = fe.lineno
  
  
        # hunkskip and hunkbody code skipped until definition of hunkhead is parsed
        if hunkbody:
          # [x] treat empty lines inside hunks as containing single space
          #     (this happens when diff is saved by copy/pasting to editor
          #      that strips trailing whitespace)
          if line.strip(b"\r\n") == b"":
              debug("expanding empty line in a middle of hunk body")
              self.warnings += 1
              line = b' ' + line
  
          # process line first
          if re.match(b"^[- \\+\\\\]", line):
              # gather stats about line endings
              if line.endswith(b"\r\n"):
                p.hunkends["crlf"] += 1
              elif line.endswith(b"\n"):
                p.hunkends["lf"] += 1
              elif line.endswith(b"\r"):
                p.hunkends["cr"] += 1
  
              if line.startswith(b"-"):
                hunkactual["linessrc"] += 1
              elif line.startswith(b"+"):
                hunkactual["linestgt"] += 1
              elif not line.startswith(b"\\"):
                hunkactual["linessrc"] += 1
                hunkactual["linestgt"] += 1
              hunk.text.append(line)
              # todo: handle \ No newline cases
          else:
              warning("invalid hunk no.%d at %d for target file %s" % (nexthunkno, lineno+1, p.target))
              # add hunk status node
              hunk.invalid = True
              p.hunks.append(hunk)
              self.errors += 1
              # switch to hunkskip state
              hunkbody = False
              hunkskip = True
  
          # check exit conditions
          if hunkactual["linessrc"] > hunk.linessrc or hunkactual["linestgt"] > hunk.linestgt:
              warning("extra lines for hunk no.%d at %d for target %s" % (nexthunkno, lineno+1, p.target))
              # add hunk status node
              hunk.invalid = True
              p.hunks.append(hunk)
              self.errors += 1
              # switch to hunkskip state
              hunkbody = False
              hunkskip = True
          elif hunk.linessrc == hunkactual["linessrc"] and hunk.linestgt == hunkactual["linestgt"]:
              # hunk parsed successfully
              p.hunks.append(hunk)
              # switch to hunkparsed state
              hunkbody = False
              hunkparsed = True
  
              # detect mixed window/unix line ends
              ends = p.hunkends
              if ((ends["cr"]!=0) + (ends["crlf"]!=0) + (ends["lf"]!=0)) > 1:
                warning("inconsistent line ends in patch hunks for %s" % p.source)
                self.warnings += 1
              if debugmode:
                debuglines = dict(ends)
                debuglines.update(file=p.target, hunk=nexthunkno)
                debug("crlf: %(crlf)d  lf: %(lf)d  cr: %(cr)d\t - file: %(file)s hunk: %(hunk)d" % debuglines)
              # fetch next line
              continue
  
        if hunkskip:
          if re_hunk_start.match(line):
            # switch to hunkhead state
            hunkskip = False
            hunkhead = True
          elif line.startswith(b"--- "):
            # switch to filenames state
            hunkskip = False
            filenames = True
            if debugmode and len(self.items) > 0:
              debug("- %2d hunks for %s" % (len(p.hunks), p.source))
  
        if filenames:
          if line.startswith(b"--- "):
            if srcname != None:
              # XXX testcase
              warning("skipping false patch for %s" % srcname)
              srcname = None
              # XXX header += srcname
              # double source filename line is encountered
              # attempt to restart from this second line
  
              # Files dated at Unix epoch don't exist, e.g.:
              # '1970-01-01 01:00:00.000000000 +0100'
              # They include timezone offsets.
              # .. which can be parsed (if we remove the nanoseconds)
              # .. by strptime() with:
              # '%Y-%m-%d %H:%M:%S %z'
              # .. but unfortunately this relies on the OSes libc
              # strptime function and %z support is patchy, so we drop
              # everything from the . onwards and group the year and time
              # separately.
            re_filename_date_time = b"^--- ([^\t]+)(?:\s([0-9-]+)\s([0-9:]+)|.*)"
            match = re.match(re_filename_date_time, line)
            # todo: support spaces in filenames
            if match:
              srcname = match.group(1).strip()
              date = match.group(2)
              time = match.group(3)
              if (date == b'1970-01-01' or date == b'1969-12-31') and time.split(b':',1)[1] == b'00:00':
                srcname = b'/dev/null'
            else:
              warning("skipping invalid filename at line %d" % (lineno+1))
              self.errors += 1
              # XXX p.header += line
              # switch back to headscan state
              filenames = False
              headscan = True
          elif not line.startswith(b"+++ "):
            if srcname != None:
              warning("skipping invalid patch with no target for %s" % srcname)
              self.errors += 1
              srcname = None
              # XXX header += srcname
              # XXX header += line
            else:
              # this should be unreachable
              warning("skipping invalid target patch")
            filenames = False
            headscan = True
          else:
            if tgtname != None:
              # XXX seems to be a dead branch
              warning("skipping invalid patch - double target at line %d" % (lineno+1))
              self.errors += 1
              srcname = None
              tgtname = None
              # XXX header += srcname
              # XXX header += tgtname
              # XXX header += line
              # double target filename line is encountered
              # switch back to headscan state
              filenames = False
              headscan = True
            else:
              re_filename_date_time = b"^\+\+\+ ([^\t]+)(?:\s([0-9-]+)\s([0-9:]+)|.*)"
              match = re.match(re_filename_date_time, line)
              if not match:
                warning("skipping invalid patch - no target filename at line %d" % (lineno+1))
                self.errors += 1
                srcname = None
                # switch back to headscan state
                filenames = False
                headscan = True
              else:
                tgtname = match.group(1).strip()
                date = match.group(2)
                time = match.group(3)
                if (date == b'1970-01-01' or date == b'1969-12-31') and time.split(b':',1)[1] == b'00:00':
                    tgtname = b'/dev/null'
                if p: # for the first run p is None
                  self.items.append(p)
                p = Patch()
                p.source = srcname
                srcname = None
                p.target = tgtname
                tgtname = None
                p.header = header
                header = []
                # switch to hunkhead state
                filenames = False
                hunkhead = True
                nexthunkno = 0
                p.hunkends = lineends.copy()
                continue
  
        if hunkhead:
          match = re.match(b"^@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@(.*)", line)
          if not match:
            if not p.hunks:
              warning("skipping invalid patch with no hunks for file %s" % p.source)
              self.errors += 1
              # XXX review switch
              # switch to headscan state
              hunkhead = False
              headscan = True
              continue
            else:
              # TODO review condition case
              # switch to headscan state
              hunkhead = False
              headscan = True
          else:
            hunk = Hunk()
            hunk.startsrc = int(match.group(1))
            hunk.linessrc = 1
            if match.group(3): hunk.linessrc = int(match.group(3))
            hunk.starttgt = int(match.group(4))
            hunk.linestgt = 1
            if match.group(6): hunk.linestgt = int(match.group(6))
            hunk.invalid = False
            hunk.desc = match.group(7)[1:].rstrip()
            hunk.text = []
  
            hunkactual["linessrc"] = hunkactual["linestgt"] = 0
  
            # switch to hunkbody state
            hunkhead = False
            hunkbody = True
            nexthunkno += 1
            continue
  
      # /while fe.next()
  
      if p:
        self.items.append(p)
  
      if not hunkparsed:
        if hunkskip:
          warning("warning: finished with errors, some hunks may be invalid")
        elif headscan:
          if len(self.items) == 0:
            warning("error: no patch data found!")
            return False
          else: # extra data at the end of file
            pass
        else:
          warning("error: patch stream is incomplete!")
          self.errors += 1
          if len(self.items) == 0:
            return False
  
      if debugmode and len(self.items) > 0:
          debug("- %2d hunks for %s" % (len(p.hunks), p.source))
  
      # XXX fix total hunks calculation
      debug("total files: %d  total hunks: %d" % (len(self.items),
          sum(len(p.hunks) for p in self.items)))
  
      # ---- detect patch and patchset types ----
      for idx, p in enumerate(self.items):
        self.items[idx].type = self._detect_type(p)
  
      types = set([p.type for p in self.items])
      if len(types) > 1:
        self.type = MIXED
      else:
        self.type = types.pop()
      # --------
  
      self._normalize_filenames()
  
      return (self.errors == 0)
function_tokens: def parse ( self , stream ) : lineends = dict ( lf = 0 , crlf = 0 , cr = 0 ) nexthunkno = 0 #: even if index starts with 0 user messages number hunks from 1 p = None hunk = None # hunkactual variable is used to calculate hunk lines for comparison hunkactual = dict ( linessrc = None , linestgt = None ) class wrapumerate ( enumerate ) : """Enumerate wrapper that uses boolean end of stream status instead of
        StopIteration exception, and properties to access line information.
        """ def __init__ ( self , * args , * * kwargs ) : # we don't call parent, it is magically created by __new__ method self . _exhausted = False self . _lineno = False # after end of stream equal to the num of lines self . _line = False # will be reset to False after end of stream def next ( self ) : """Try to read the next line and return True if it is available,
             False if end of stream is reached.""" if self . _exhausted : return False try : self . _lineno , self . _line = compat_next ( super ( wrapumerate , self ) ) except StopIteration : self . _exhausted = True self . _line = False return False return True @ property def is_empty ( self ) : return self . _exhausted @ property def line ( self ) : return self . _line @ property def lineno ( self ) : return self . _lineno # define states (possible file regions) that direct parse flow headscan = True # start with scanning header filenames = False # lines starting with --- and +++ hunkhead = False # @@ -R +R @@ sequence hunkbody = False # hunkskip = False # skipping invalid hunk mode hunkparsed = False # state after successfully parsed hunk # regexp to match start of hunk, used groups - 1,3,4,6 re_hunk_start = re . compile ( b"^@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@" ) self . errors = 0 # temp buffers for header and filenames info header = [ ] srcname = None tgtname = None # start of main cycle # each parsing block already has line available in fe.line fe = wrapumerate ( stream ) while fe . next ( ) : # -- deciders: these only switch state to decide who should process # --           line fetched at the start of this cycle if hunkparsed : hunkparsed = False if re_hunk_start . match ( fe . line ) : hunkhead = True elif fe . line . startswith ( b"--- " ) : filenames = True else : headscan = True # -- ------------------------------------ # read out header if headscan : while not fe . is_empty and not fe . line . startswith ( b"--- " ) : header . append ( fe . line ) fe . next ( ) if fe . is_empty : if p is None : debug ( "no patch data found" ) # error is shown later self . errors += 1 else : info ( "%d unparsed bytes left at the end of stream" % len ( b'' . join ( header ) ) ) self . warnings += 1 # TODO check for \No new line at the end.. # TODO test for unparsed bytes # otherwise error += 1 # this is actually a loop exit continue headscan = False # switch to filenames state filenames = True line = fe . line lineno = fe . lineno # hunkskip and hunkbody code skipped until definition of hunkhead is parsed if hunkbody : # [x] treat empty lines inside hunks as containing single space #     (this happens when diff is saved by copy/pasting to editor #      that strips trailing whitespace) if line . strip ( b"\r\n" ) == b"" : debug ( "expanding empty line in a middle of hunk body" ) self . warnings += 1 line = b' ' + line # process line first if re . match ( b"^[- \\+\\\\]" , line ) : # gather stats about line endings if line . endswith ( b"\r\n" ) : p . hunkends [ "crlf" ] += 1 elif line . endswith ( b"\n" ) : p . hunkends [ "lf" ] += 1 elif line . endswith ( b"\r" ) : p . hunkends [ "cr" ] += 1 if line . startswith ( b"-" ) : hunkactual [ "linessrc" ] += 1 elif line . startswith ( b"+" ) : hunkactual [ "linestgt" ] += 1 elif not line . startswith ( b"\\" ) : hunkactual [ "linessrc" ] += 1 hunkactual [ "linestgt" ] += 1 hunk . text . append ( line ) # todo: handle \ No newline cases else : warning ( "invalid hunk no.%d at %d for target file %s" % ( nexthunkno , lineno + 1 , p . target ) ) # add hunk status node hunk . invalid = True p . hunks . append ( hunk ) self . errors += 1 # switch to hunkskip state hunkbody = False hunkskip = True # check exit conditions if hunkactual [ "linessrc" ] > hunk . linessrc or hunkactual [ "linestgt" ] > hunk . linestgt : warning ( "extra lines for hunk no.%d at %d for target %s" % ( nexthunkno , lineno + 1 , p . target ) ) # add hunk status node hunk . invalid = True p . hunks . append ( hunk ) self . errors += 1 # switch to hunkskip state hunkbody = False hunkskip = True elif hunk . linessrc == hunkactual [ "linessrc" ] and hunk . linestgt == hunkactual [ "linestgt" ] : # hunk parsed successfully p . hunks . append ( hunk ) # switch to hunkparsed state hunkbody = False hunkparsed = True # detect mixed window/unix line ends ends = p . hunkends if ( ( ends [ "cr" ] != 0 ) + ( ends [ "crlf" ] != 0 ) + ( ends [ "lf" ] != 0 ) ) > 1 : warning ( "inconsistent line ends in patch hunks for %s" % p . source ) self . warnings += 1 if debugmode : debuglines = dict ( ends ) debuglines . update ( file = p . target , hunk = nexthunkno ) debug ( "crlf: %(crlf)d  lf: %(lf)d  cr: %(cr)d\t - file: %(file)s hunk: %(hunk)d" % debuglines ) # fetch next line continue if hunkskip : if re_hunk_start . match ( line ) : # switch to hunkhead state hunkskip = False hunkhead = True elif line . startswith ( b"--- " ) : # switch to filenames state hunkskip = False filenames = True if debugmode and len ( self . items ) > 0 : debug ( "- %2d hunks for %s" % ( len ( p . hunks ) , p . source ) ) if filenames : if line . startswith ( b"--- " ) : if srcname != None : # XXX testcase warning ( "skipping false patch for %s" % srcname ) srcname = None # XXX header += srcname # double source filename line is encountered # attempt to restart from this second line # Files dated at Unix epoch don't exist, e.g.: # '1970-01-01 01:00:00.000000000 +0100' # They include timezone offsets. # .. which can be parsed (if we remove the nanoseconds) # .. by strptime() with: # '%Y-%m-%d %H:%M:%S %z' # .. but unfortunately this relies on the OSes libc # strptime function and %z support is patchy, so we drop # everything from the . onwards and group the year and time # separately. re_filename_date_time = b"^--- ([^\t]+)(?:\s([0-9-]+)\s([0-9:]+)|.*)" match = re . match ( re_filename_date_time , line ) # todo: support spaces in filenames if match : srcname = match . group ( 1 ) . strip ( ) date = match . group ( 2 ) time = match . group ( 3 ) if ( date == b'1970-01-01' or date == b'1969-12-31' ) and time . split ( b':' , 1 ) [ 1 ] == b'00:00' : srcname = b'/dev/null' else : warning ( "skipping invalid filename at line %d" % ( lineno + 1 ) ) self . errors += 1 # XXX p.header += line # switch back to headscan state filenames = False headscan = True elif not line . startswith ( b"+++ " ) : if srcname != None : warning ( "skipping invalid patch with no target for %s" % srcname ) self . errors += 1 srcname = None # XXX header += srcname # XXX header += line else : # this should be unreachable warning ( "skipping invalid target patch" ) filenames = False headscan = True else : if tgtname != None : # XXX seems to be a dead branch warning ( "skipping invalid patch - double target at line %d" % ( lineno + 1 ) ) self . errors += 1 srcname = None tgtname = None # XXX header += srcname # XXX header += tgtname # XXX header += line # double target filename line is encountered # switch back to headscan state filenames = False headscan = True else : re_filename_date_time = b"^\+\+\+ ([^\t]+)(?:\s([0-9-]+)\s([0-9:]+)|.*)" match = re . match ( re_filename_date_time , line ) if not match : warning ( "skipping invalid patch - no target filename at line %d" % ( lineno + 1 ) ) self . errors += 1 srcname = None # switch back to headscan state filenames = False headscan = True else : tgtname = match . group ( 1 ) . strip ( ) date = match . group ( 2 ) time = match . group ( 3 ) if ( date == b'1970-01-01' or date == b'1969-12-31' ) and time . split ( b':' , 1 ) [ 1 ] == b'00:00' : tgtname = b'/dev/null' if p : # for the first run p is None self . items . append ( p ) p = Patch ( ) p . source = srcname srcname = None p . target = tgtname tgtname = None p . header = header header = [ ] # switch to hunkhead state filenames = False hunkhead = True nexthunkno = 0 p . hunkends = lineends . copy ( ) continue if hunkhead : match = re . match ( b"^@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@(.*)" , line ) if not match : if not p . hunks : warning ( "skipping invalid patch with no hunks for file %s" % p . source ) self . errors += 1 # XXX review switch # switch to headscan state hunkhead = False headscan = True continue else : # TODO review condition case # switch to headscan state hunkhead = False headscan = True else : hunk = Hunk ( ) hunk . startsrc = int ( match . group ( 1 ) ) hunk . linessrc = 1 if match . group ( 3 ) : hunk . linessrc = int ( match . group ( 3 ) ) hunk . starttgt = int ( match . group ( 4 ) ) hunk . linestgt = 1 if match . group ( 6 ) : hunk . linestgt = int ( match . group ( 6 ) ) hunk . invalid = False hunk . desc = match . group ( 7 ) [ 1 : ] . rstrip ( ) hunk . text = [ ] hunkactual [ "linessrc" ] = hunkactual [ "linestgt" ] = 0 # switch to hunkbody state hunkhead = False hunkbody = True nexthunkno += 1 continue # /while fe.next() if p : self . items . append ( p ) if not hunkparsed : if hunkskip : warning ( "warning: finished with errors, some hunks may be invalid" ) elif headscan : if len ( self . items ) == 0 : warning ( "error: no patch data found!" ) return False else : # extra data at the end of file pass else : warning ( "error: patch stream is incomplete!" ) self . errors += 1 if len ( self . items ) == 0 : return False if debugmode and len ( self . items ) > 0 : debug ( "- %2d hunks for %s" % ( len ( p . hunks ) , p . source ) ) # XXX fix total hunks calculation debug ( "total files: %d  total hunks: %d" % ( len ( self . items ) , sum ( len ( p . hunks ) for p in self . items ) ) ) # ---- detect patch and patchset types ---- for idx , p in enumerate ( self . items ) : self . items [ idx ] . type = self . _detect_type ( p ) types = set ( [ p . type for p in self . items ] ) if len ( types ) > 1 : self . type = MIXED else : self . type = types . pop ( ) # -------- self . _normalize_filenames ( ) return ( self . errors == 0 )