cffi: pass C type and attribute names as str instead of bytes

mercurial/cffi/bdiff.py

@@ -1,88 +1,88
 # bdiff.py - CFFI implementation of bdiff.c
 #
 # Copyright 2016 Maciej Fijalkowski <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import struct
 
 from typing import (
     List,
     Tuple,
 )
 
 from ..pure.bdiff import *
 from . import _bdiff  # pytype: disable=import-error
 
 ffi = _bdiff.ffi
 lib = _bdiff.lib
 
 
 def blocks(sa: bytes, sb: bytes) -> List[Tuple[int, int, int, int]]:
-    a = ffi.new(b"struct bdiff_line**")
-    b = ffi.new(b"struct bdiff_line**")
-    ac = ffi.new(b"char[]", str(sa))
-    bc = ffi.new(b"char[]", str(sb))
-    l = ffi.new(b"struct bdiff_hunk*")
+    a = ffi.new("struct bdiff_line**")
+    b = ffi.new("struct bdiff_line**")
+    ac = ffi.new("char[]", str(sa))
+    bc = ffi.new("char[]", str(sb))
+    l = ffi.new("struct bdiff_hunk*")
     try:
         an = lib.bdiff_splitlines(ac, len(sa), a)
         bn = lib.bdiff_splitlines(bc, len(sb), b)
         if not a[0] or not b[0]:
             raise MemoryError
         count = lib.bdiff_diff(a[0], an, b[0], bn, l)
         if count < 0:
             raise MemoryError
         rl = [(0, 0, 0, 0)] * count
         h = l.next
         i = 0
         while h:
             rl[i] = (h.a1, h.a2, h.b1, h.b2)
             h = h.next
             i += 1
     finally:
         lib.free(a[0])
         lib.free(b[0])
         lib.bdiff_freehunks(l.next)
     return rl
 
 
 def bdiff(sa: bytes, sb: bytes) -> bytes:
-    a = ffi.new(b"struct bdiff_line**")
-    b = ffi.new(b"struct bdiff_line**")
-    ac = ffi.new(b"char[]", str(sa))
-    bc = ffi.new(b"char[]", str(sb))
-    l = ffi.new(b"struct bdiff_hunk*")
+    a = ffi.new("struct bdiff_line**")
+    b = ffi.new("struct bdiff_line**")
+    ac = ffi.new("char[]", str(sa))
+    bc = ffi.new("char[]", str(sb))
+    l = ffi.new("struct bdiff_hunk*")
     try:
         an = lib.bdiff_splitlines(ac, len(sa), a)
         bn = lib.bdiff_splitlines(bc, len(sb), b)
         if not a[0] or not b[0]:
             raise MemoryError
         count = lib.bdiff_diff(a[0], an, b[0], bn, l)
         if count < 0:
             raise MemoryError
         rl = []
         h = l.next
         la = lb = 0
         while h:
             if h.a1 != la or h.b1 != lb:
                 lgt = (b[0] + h.b1).l - (b[0] + lb).l
                 rl.append(
                     struct.pack(
                         b">lll",
                         (a[0] + la).l - a[0].l,
                         (a[0] + h.a1).l - a[0].l,
                         lgt,
                     )
                 )
                 rl.append(str(ffi.buffer((b[0] + lb).l, lgt)))
             la = h.a2
             lb = h.b2
             h = h.next
 
     finally:
         lib.free(a[0])
         lib.free(b[0])
         lib.bdiff_freehunks(l.next)
     return b"".join(rl)

mercurial/cffi/mpatch.py

@@ -1,50 +1,50
 # mpatch.py - CFFI implementation of mpatch.c
 #
 # Copyright 2016 Maciej Fijalkowski <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 from typing import List
 
 from ..pure.mpatch import *
 from ..pure.mpatch import mpatchError  # silence pyflakes
 from . import _mpatch  # pytype: disable=import-error
 
 ffi = _mpatch.ffi
 lib = _mpatch.lib
 
 
 @ffi.def_extern()
 def cffi_get_next_item(arg, pos):
     all, bins = ffi.from_handle(arg)
-    container = ffi.new(b"struct mpatch_flist*[1]")
-    to_pass = ffi.new(b"char[]", str(bins[pos]))
+    container = ffi.new("struct mpatch_flist*[1]")
+    to_pass = ffi.new("char[]", str(bins[pos]))
     all.append(to_pass)
     r = lib.mpatch_decode(to_pass, len(to_pass) - 1, container)
     if r < 0:
         return ffi.NULL
     return container[0]
 
 
 def patches(text: bytes, bins: List[bytes]) -> bytes:
     lgt = len(bins)
     all = []
     if not lgt:
         return text
     arg = (all, bins)
     patch = lib.mpatch_fold(ffi.new_handle(arg), lib.cffi_get_next_item, 0, lgt)
     if not patch:
         raise mpatchError(b"cannot decode chunk")
     outlen = lib.mpatch_calcsize(len(text), patch)
     if outlen < 0:
         lib.mpatch_lfree(patch)
         raise mpatchError(b"inconsistency detected")
-    buf = ffi.new(b"char[]", outlen)
+    buf = ffi.new("char[]", outlen)
     if lib.mpatch_apply(buf, text, len(text), patch) < 0:
         lib.mpatch_lfree(patch)
         raise mpatchError(b"error applying patches")
     res = ffi.buffer(buf, outlen)[:]
     lib.mpatch_lfree(patch)
     return res

mercurial/cffi/osutil.py

@@ -1,114 +1,114
 # osutil.py - CFFI version of osutil.c
 #
 # Copyright 2016 Maciej Fijalkowski <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import os
 import stat as statmod
 
 from ..pure.osutil import *
 
 from .. import pycompat
 
 if pycompat.isdarwin:
     from . import _osutil  # pytype: disable=import-error
 
     ffi = _osutil.ffi
     lib = _osutil.lib
 
     listdir_batch_size = 4096
     # tweakable number, only affects performance, which chunks
     # of bytes do we get back from getattrlistbulk
 
     attrkinds = [None] * 20  # we need the max no for enum VXXX, 20 is plenty
 
     attrkinds[lib.VREG] = statmod.S_IFREG
     attrkinds[lib.VDIR] = statmod.S_IFDIR
     attrkinds[lib.VLNK] = statmod.S_IFLNK
     attrkinds[lib.VBLK] = statmod.S_IFBLK
     attrkinds[lib.VCHR] = statmod.S_IFCHR
     attrkinds[lib.VFIFO] = statmod.S_IFIFO
     attrkinds[lib.VSOCK] = statmod.S_IFSOCK
 
     class stat_res:
         def __init__(self, st_mode, st_mtime, st_size):
             self.st_mode = st_mode
             self.st_mtime = st_mtime
             self.st_size = st_size
 
-    tv_sec_ofs = ffi.offsetof(b"struct timespec", b"tv_sec")
-    buf = ffi.new(b"char[]", listdir_batch_size)
+    tv_sec_ofs = ffi.offsetof("struct timespec", "tv_sec")
+    buf = ffi.new("char[]", listdir_batch_size)
 
     def listdirinternal(dfd, req, stat, skip):
         ret = []
         while True:
             r = lib.getattrlistbulk(dfd, req, buf, listdir_batch_size, 0)
             if r == 0:
                 break
             if r == -1:
                 raise OSError(ffi.errno, os.strerror(ffi.errno))
-            cur = ffi.cast(b"val_attrs_t*", buf)
+            cur = ffi.cast("val_attrs_t*", buf)
             for i in range(r):
                 lgt = cur.length
-                assert lgt == ffi.cast(b'uint32_t*', cur)[0]
+                assert lgt == ffi.cast('uint32_t*', cur)[0]
                 ofs = cur.name_info.attr_dataoffset
                 str_lgt = cur.name_info.attr_length
-                base_ofs = ffi.offsetof(b'val_attrs_t', b'name_info')
+                base_ofs = ffi.offsetof('val_attrs_t', 'name_info')
                 name = bytes(
                     ffi.buffer(
-                        ffi.cast(b"char*", cur) + base_ofs + ofs, str_lgt - 1
+                        ffi.cast("char*", cur) + base_ofs + ofs, str_lgt - 1
                     )
                 )
                 tp = attrkinds[cur.obj_type]
                 if name == b"." or name == b"..":
                     continue
                 if skip == name and tp == statmod.S_ISDIR:
                     return []
                 if stat:
                     mtime = cur.mtime.tv_sec
                     mode = (cur.accessmask & ~lib.S_IFMT) | tp
                     ret.append(
                         (
                             name,
                             tp,
                             stat_res(
                                 st_mode=mode,
                                 st_mtime=mtime,
                                 st_size=cur.datalength,
                             ),
                         )
                     )
                 else:
                     ret.append((name, tp))
                 cur = ffi.cast(
-                    b"val_attrs_t*", int(ffi.cast(b"intptr_t", cur)) + lgt
+                    "val_attrs_t*", int(ffi.cast("intptr_t", cur)) + lgt
                 )
         return ret
 
     def listdir(path, stat=False, skip=None):
-        req = ffi.new(b"struct attrlist*")
+        req = ffi.new("struct attrlist*")
         req.bitmapcount = lib.ATTR_BIT_MAP_COUNT
         req.commonattr = (
             lib.ATTR_CMN_RETURNED_ATTRS
             | lib.ATTR_CMN_NAME
             | lib.ATTR_CMN_OBJTYPE
             | lib.ATTR_CMN_ACCESSMASK
             | lib.ATTR_CMN_MODTIME
         )
         req.fileattr = lib.ATTR_FILE_DATALENGTH
         dfd = lib.open(path, lib.O_RDONLY, 0)
         if dfd == -1:
             raise OSError(ffi.errno, os.strerror(ffi.errno))
 
         try:
             ret = listdirinternal(dfd, req, stat, skip)
         finally:
             try:
                 lib.close(dfd)
             except BaseException:
                 pass  # we ignore all the errors from closing, not
                 # much we can do about that
         return ret