python: Add compression_suffix and detect_compression functions.

diff --git a/src/python/__init__.py b/src/python/__init__.py
index dd08405..0b872d4 100644 (file)
--- a/src/python/__init__.py
+++ b/src/python/__init__.py
@@ -172,3 +172,6 @@ checksum_type       = _createrepo_c.checksum_type
 
 def compress_file(src, dst, comtype, stat=None):
     return _createrepo_c.compress_file(src, dst, comtype, stat)
+
+compression_suffix  = _createrepo_c.compression_suffix
+detect_compression  = _createrepo_c.detect_compression

diff --git a/src/python/compression_wrapper-py.c b/src/python/compression_wrapper-py.c
index 02807e1..4a8a823 100644 (file)
--- a/src/python/compression_wrapper-py.c
+++ b/src/python/compression_wrapper-py.c
@@ -26,6 +26,49 @@
 #include "contentstat-py.h"
 #include "typeconversion.h"
 
+/*
+ * Module functions
+ */
+
+PyObject *
+py_compression_suffix(PyObject *self, PyObject *args)
+{
+    int type;
+
+    CR_UNUSED(self);
+
+    if (!PyArg_ParseTuple(args, "i:py_compression_suffix", &type))
+        return NULL;
+
+    return PyStringOrNone_FromString(cr_compression_suffix(type));
+}
+
+PyObject *
+py_detect_compression(PyObject *self, PyObject *args)
+{
+    long type;
+    char *filename;
+    GError *tmp_err = NULL;
+
+    CR_UNUSED(self);
+
+    if (!PyArg_ParseTuple(args, "s:py_detect_compression", &filename))
+        return NULL;
+
+    type = cr_detect_compression(filename, &tmp_err);
+    if (tmp_err) {
+        PyErr_Format(CrErr_Exception, "%s", tmp_err->message);
+        g_clear_error(&tmp_err);
+        return NULL;
+    }
+
+    return PyLong_FromLong(type);
+}
+
+/*
+ * CrFile object
+ */
+
 typedef struct {
     PyObject_HEAD
     CR_FILE *f;

diff --git a/src/python/compression_wrapper-py.h b/src/python/compression_wrapper-py.h
index 81ebe2f..7d75cab 100644 (file)
--- a/src/python/compression_wrapper-py.h
+++ b/src/python/compression_wrapper-py.h
@@ -26,4 +26,7 @@ extern PyTypeObject CrFile_Type;
 
 #define CrFileObject_Check(o)   PyObject_TypeCheck(o, &CrFile_Type)
 
+PyObject *py_compression_suffix(PyObject *self, PyObject *args);
+PyObject *py_detect_compression(PyObject *self, PyObject *args);
+
 #endif

diff --git a/src/python/createrepo_cmodule.c b/src/python/createrepo_cmodule.c
index 24d62a8..6e5e426 100644 (file)
--- a/src/python/createrepo_cmodule.c
+++ b/src/python/createrepo_cmodule.c
@@ -64,6 +64,10 @@ static struct PyMethodDef createrepo_c_methods[] = {
      METH_VARARGS, NULL},
     {"compress_file",           (PyCFunction)py_compress_file_with_stat,
      METH_VARARGS, NULL},
+    {"compression_suffix",      (PyCFunction)py_compression_suffix,
+     METH_VARARGS, NULL},
+    {"detect_compression",      (PyCFunction)py_detect_compression,
+     METH_VARARGS, NULL},
     { NULL }
 };
 

diff --git a/tests/python/tests/test_compression_wrapper.py b/tests/python/tests/test_compression_wrapper.py
new file mode 100644 (file)
index 0000000..d19569c
--- /dev/null
+++ b/tests/python/tests/test_compression_wrapper.py
@@ -0,0 +1,59 @@
+import unittest
+import os.path
+import createrepo_c as cr
+
+from fixtures import *
+
+class TestCaseCompressionWrapper(unittest.TestCase):
+    def test_compression_suffix(self):
+        self.assertEqual(cr.compression_suffix(cr.AUTO_DETECT_COMPRESSION), None)
+        self.assertEqual(cr.compression_suffix(cr.UNKNOWN_COMPRESSION), None)
+        self.assertEqual(cr.compression_suffix(cr.NO_COMPRESSION), None)
+        self.assertEqual(cr.compression_suffix(123), None)
+
+        self.assertEqual(cr.compression_suffix(cr.GZ), ".gz")
+        self.assertEqual(cr.compression_suffix(cr.BZ2), ".bz2")
+        self.assertEqual(cr.compression_suffix(cr.XZ), ".xz")
+
+    def test_detect_compression(self):
+
+        # no compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.txt")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.NO_COMPRESSION)
+
+        # gz compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.txt.gz")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.GZ)
+
+        # bz2 compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.txt.bz2")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.BZ2)
+
+        # xz compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.txt.xz")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.XZ)
+
+        # Bad suffix - no compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.foo0")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.NO_COMPRESSION)
+
+        # Bad suffix - gz compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.foo1")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.GZ)
+
+        # Bad suffix - bz2 compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.foo2")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.BZ2)
+
+        # Bad suffix - xz compression
+        path = os.path.join(COMPRESSED_FILES_PATH, "01_plain.foo3")
+        comtype = cr.detect_compression(path)
+        self.assertEqual(comtype, cr.XZ)
+