Add function to perform delta encoding.

Author: fancycode

src/pylzma/pylzma.c

@@ -310,6 +310,46 @@ pylzma_delta_decode(PyObject *self, PyObject *args)
     return result;
 }
 
+const char
+doc_delta_encode[] =
+    "delta_encode(data, delta) -- Encode Delta streams.";
+
+static PyObject *
+pylzma_delta_encode(PyObject *self, PyObject *args)
+{
+    char *data;
+    PARSE_LENGTH_TYPE length;
+    unsigned int delta;
+    Byte state[DELTA_STATE_SIZE];
+    Byte *tmp;
+    PyObject *result;
+
+    if (!PyArg_ParseTuple(args, "s#I", &data, &length, &delta)) {
+        return NULL;
+    }
+
+    if (!delta) {
+        PyErr_SetString(PyExc_TypeError, "delta must be non-zero");
+        return NULL;
+    }
+
+    if (!length) {
+        return PyBytes_FromString("");
+    }
+
+    result = PyBytes_FromStringAndSize(data, length);
+    if (!result) {
+        return NULL;
+    }
+
+    Delta_Init(state);
+    tmp = (Byte *) PyBytes_AS_STRING(result);
+    Py_BEGIN_ALLOW_THREADS
+    Delta_Encode(state, delta, tmp, length);
+    Py_END_ALLOW_THREADS
+    return result;
+}
+
 PyMethodDef
 methods[] = {
     // exported functions
@@ -331,6 +371,7 @@ methods[] = {
     {"bcj2_decode", (PyCFunction)pylzma_bcj2_decode,   METH_VARARGS,   (char *)&doc_bcj2_decode},
     // Delta
     {"delta_decode", (PyCFunction)pylzma_delta_decode,   METH_VARARGS,   (char *)&doc_delta_decode},
+    {"delta_encode", (PyCFunction)pylzma_delta_encode,   METH_VARARGS,   (char *)&doc_delta_encode},
     {NULL, NULL},
 };
 

tests/test_pylzma.py

@@ -245,6 +245,18 @@ def test_github_10(self):
         # prevent regression of github #10
         self.assertRaises(ValueError, pylzma.compress, bytes("foo", 'ascii'), dictionary=100)
 
+    def test_delta(self):
+        for i in range(8, 18):
+            size = 1 << i
+            original = generate_random(size)
+            delta = random.randint(1, 255)
+            encoded = pylzma.delta_encode(original, delta)
+            self.assertEqual(len(encoded), size)
+            self.assertNotEqual(md5(original).hexdigest(), md5(encoded).hexdigest())
+            result = pylzma.delta_decode(encoded, delta)
+            self.assertEqual(len(result), size)
+            self.assertEqual(md5(original).hexdigest(), md5(result).hexdigest())
+
 def suite():
     suite = unittest.TestSuite()