SET (craeterepo_cmodule_SRCS
checksum-py.c
+ compression_wrapper-py.c
contentstat-py.c
createrepo_cmodule.c
exception-py.c
SHA384 = _createrepo_c.SHA384
SHA512 = _createrepo_c.SHA512
+MODE_READ = _createrepo_c.MODE_READ
+MODE_WRITE = _createrepo_c.MODE_WRITE
+
AUTO_DETECT_COMPRESSION = _createrepo_c.AUTO_DETECT_COMPRESSION
UNKNOWN_COMPRESSION = _createrepo_c.UNKNOWN_COMPRESSION
NO_COMPRESSION = _createrepo_c.NO_COMPRESSION
ContentStat = _createrepo_c.ContentStat
+# CrFile class
+
+class CrFile(_createrepo_c.CrFile):
+ def __init__(self, filename, mode=MODE_READ,
+ comtype=NO_COMPRESSION, stat=None):
+ _createrepo_c.CrFile.__init__(self, filename, mode, comtype, stat)
+
# Metadata class
Metadata = _createrepo_c.Metadata
--- /dev/null
+/* createrepo_c - Library of routines for manipulation with repodata
+ * Copyright (C) 2013 Tomas Mlcoch
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
+ * USA.
+ */
+
+#include <Python.h>
+#include <assert.h>
+#include <stddef.h>
+
+#include "compression_wrapper-py.h"
+#include "exception-py.h"
+#include "contentstat-py.h"
+#include "typeconversion.h"
+
+typedef struct {
+ PyObject_HEAD
+ CR_FILE *f;
+ PyObject *py_stat;
+} _CrFileObject;
+
+static PyObject * py_close(_CrFileObject *self, void *nothing);
+
+static int
+check_CrFileStatus(const _CrFileObject *self)
+{
+ assert(self != NULL);
+ assert(CrFileObject_Check(self));
+ if (self->f == NULL) {
+ PyErr_SetString(CrErr_Exception,
+ "Improper createrepo_c CrFile object (Already closed file?).");
+ return -1;
+ }
+ return 0;
+}
+
+/* Function on the type */
+
+static PyObject *
+crfile_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
+{
+ CR_UNUSED(args);
+ CR_UNUSED(kwds);
+ _CrFileObject *self = (_CrFileObject *)type->tp_alloc(type, 0);
+ if (self) {
+ self->f = NULL;
+ self->py_stat = NULL;
+ }
+ return (PyObject *)self;
+}
+
+static int
+crfile_init(_CrFileObject *self, PyObject *args, PyObject *kwds)
+{
+ char *path;
+ int mode, comtype;
+ GError *err = NULL;
+ PyObject *py_stat, *ret;
+ cr_ContentStat *stat;
+
+ CR_UNUSED(kwds);
+
+ if (!PyArg_ParseTuple(args, "siiO|:crfile_init",
+ &path, &mode, &comtype, &py_stat))
+ return -1;
+
+ /* Check arguments */
+ if (mode != CR_CW_MODE_READ && mode != CR_CW_MODE_WRITE) {
+ PyErr_SetString(PyExc_ValueError, "Bad open mode");
+ return -1;
+ }
+
+ if (comtype < 0 || comtype >= CR_CW_COMPRESSION_SENTINEL) {
+ PyErr_SetString(PyExc_ValueError, "Unknown compression type");
+ return -1;
+ }
+
+ if (py_stat == Py_None) {
+ stat = NULL;
+ } else if (ContentStatObject_Check(py_stat)) {
+ stat = ContentStat_FromPyObject(py_stat);
+ } else {
+ PyErr_SetString(PyExc_ValueError, "Use ContentStat or None");
+ return -1;
+ }
+
+ /* Free all previous resources when reinitialization */
+ ret = py_close(self, NULL);
+ Py_XDECREF(ret);
+ Py_XDECREF(self->py_stat);
+ self->py_stat = NULL;
+ if (ret == NULL) {
+ // Error encountered!
+ return -1;
+ }
+
+ /* Init */
+ self->f = cr_open_with_stat(path, mode, comtype, stat, &err);
+ if (err) {
+ PyErr_Format(CrErr_Exception, "CrFile initialization failed: %s", err->message);
+ g_clear_error(&err);
+ return -1;
+ }
+
+ self->py_stat = py_stat;
+ Py_XINCREF(py_stat);
+
+ return 0;
+}
+
+static void
+crfile_dealloc(_CrFileObject *self)
+{
+ cr_close(self->f, NULL);
+ Py_XDECREF(self->py_stat);
+ Py_TYPE(self)->tp_free(self);
+}
+
+static PyObject *
+crfile_repr(_CrFileObject *self)
+{
+ char *mode;
+
+ switch (self->f->mode) {
+ case CR_CW_MODE_READ:
+ mode = "Read mode";
+ break;
+ case CR_CW_MODE_WRITE:
+ mode = "Write mode";
+ break;
+ default:
+ mode = "Unknown mode";
+ }
+
+ return PyString_FromFormat("<createrepo_c.CrFile %s object>", mode);
+}
+
+/* CrFile methods */
+
+static PyObject *
+py_write(_CrFileObject *self, PyObject *args)
+{
+ char *str;
+ Py_ssize_t len;
+ GError *tmp_err = NULL;
+
+ if (!PyArg_ParseTuple(args, "s#:set_num_of_pkgs", &str, &len))
+ return NULL;
+
+ if (check_CrFileStatus(self))
+ return NULL;
+
+ cr_write(self->f, str, len, &tmp_err);
+ if (tmp_err) {
+ PyErr_Format(CrErr_Exception, "%s", tmp_err->message);
+ g_clear_error(&tmp_err);
+ return NULL;
+ }
+
+ Py_RETURN_NONE;
+}
+
+static PyObject *
+py_close(_CrFileObject *self, void *nothing)
+{
+ GError *tmp_err = NULL;
+
+ CR_UNUSED(nothing);
+
+ if (self->f) {
+ cr_close(self->f, &tmp_err);
+ self->f = NULL;
+ }
+
+ Py_XDECREF(self->py_stat);
+ self->py_stat = NULL;
+
+ if (tmp_err) {
+ PyErr_Format(CrErr_Exception, "Close error: %s", tmp_err->message);
+ g_clear_error(&tmp_err);
+ return NULL;
+ }
+
+ Py_RETURN_NONE;
+}
+
+static struct PyMethodDef crfile_methods[] = {
+ {"write", (PyCFunction)py_write, METH_VARARGS, NULL},
+ {"close", (PyCFunction)py_close, METH_NOARGS, NULL},
+ {NULL} /* sentinel */
+};
+
+PyTypeObject CrFile_Type = {
+ PyObject_HEAD_INIT(NULL)
+ 0, /* ob_size */
+ "createrepo_c.CrFile", /* tp_name */
+ sizeof(_CrFileObject), /* tp_basicsize */
+ 0, /* tp_itemsize */
+ (destructor) crfile_dealloc, /* tp_dealloc */
+ 0, /* tp_print */
+ 0, /* tp_getattr */
+ 0, /* tp_setattr */
+ 0, /* tp_compare */
+ (reprfunc) crfile_repr, /* tp_repr */
+ 0, /* tp_as_number */
+ 0, /* tp_as_sequence */
+ 0, /* tp_as_mapping */
+ 0, /* tp_hash */
+ 0, /* tp_call */
+ 0, /* tp_str */
+ 0, /* tp_getattro */
+ 0, /* tp_setattro */
+ 0, /* tp_as_buffer */
+ Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, /* tp_flags */
+ "CrFile object", /* tp_doc */
+ 0, /* tp_traverse */
+ 0, /* tp_clear */
+ 0, /* tp_richcompare */
+ 0, /* tp_weaklistoffset */
+ PyObject_SelfIter, /* tp_iter */
+ 0, /* tp_iternext */
+ crfile_methods, /* tp_methods */
+ 0, /* tp_members */
+ 0, /* tp_getset */
+ 0, /* tp_base */
+ 0, /* tp_dict */
+ 0, /* tp_descr_get */
+ 0, /* tp_descr_set */
+ 0, /* tp_dictoffset */
+ (initproc) crfile_init, /* tp_init */
+ 0, /* tp_alloc */
+ crfile_new, /* tp_new */
+ 0, /* tp_free */
+ 0, /* tp_is_gc */
+};
--- /dev/null
+/* createrepo_c - Library of routines for manipulation with repodata
+ * Copyright (C) 2013 Tomas Mlcoch
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
+ * USA.
+ */
+
+#ifndef CR_COMPRESSION_WRAPPER_PY_H
+#define CR_COMPRESSION_WRAPPER_PY_H
+
+#include "src/createrepo_c.h"
+
+extern PyTypeObject CrFile_Type;
+
+#define CrFileObject_Check(o) PyObject_TypeCheck(o, &CrFile_Type)
+
+#endif
#include "src/createrepo_c.h"
#include "checksum-py.h"
+#include "compression_wrapper-py.h"
#include "contentstat-py.h"
#include "exception-py.h"
#include "load_metadata-py.h"
Py_INCREF(&ContentStat_Type);
PyModule_AddObject(m, "ContentStat", (PyObject *)&ContentStat_Type);
+ /* _createrepo_c.CrFile */
+ if (PyType_Ready(&CrFile_Type) < 0)
+ return;
+ Py_INCREF(&CrFile_Type);
+ PyModule_AddObject(m, "CrFile", (PyObject *)&CrFile_Type);
+
/* _createrepo_c.Package */
if (PyType_Ready(&Package_Type) < 0)
return;
PyModule_AddIntConstant(m, "SHA384", CR_CHECKSUM_SHA384);
PyModule_AddIntConstant(m, "SHA512", CR_CHECKSUM_SHA512);
+ /* File open modes */
+ PyModule_AddIntConstant(m, "MODE_READ", CR_CW_MODE_READ);
+ PyModule_AddIntConstant(m, "MODE_WRITE", CR_CW_MODE_WRITE);
+
/* Compression types */
PyModule_AddIntConstant(m, "AUTO_DETECT_COMPRESSION", CR_CW_AUTO_DETECT_COMPRESSION);
PyModule_AddIntConstant(m, "UNKNOWN_COMPRESSION", CR_CW_UNKNOWN_COMPRESSION);
self.assertEqual(cs.checksum, "67bc6282915fad80dc11f3d7c3210977a0bde"\
"05a762256d86083c2447d425776")
- def test_contentstat_2(self):
+ def test_contentstat_ref_in_xmlfile(self):
"""Test if reference is saved properly"""
pkg = cr.package_from_rpm(PKG_ARCHER_PATH)
f.close()
self.assertTrue(os.path.isfile(path))
+
+ def test_contentstat_ref_in_crfile(self):
+ """Test if reference is saved properly"""
+
+ cs = cr.ContentStat(cr.SHA256)
+ self.assertEqual(cs.size, 0)
+ self.assertEqual(cs.checksum_type, cr.SHA256)
+ self.assertEqual(cs.checksum, None)
+
+ path = os.path.join(self.tmpdir, "foofile.gz")
+ f = cr.CrFile(path, cr.MODE_WRITE, cr.GZ_COMPRESSION, cs)
+ self.assertTrue(f)
+ self.assertTrue(os.path.isfile(path))
+ del cs
+ f.write("foobar")
+ f.close()
+
+ self.assertTrue(os.path.isfile(path))
--- /dev/null
+import unittest
+import shutil
+import tempfile
+import os.path
+import createrepo_c as cr
+
+from fixtures import *
+
+class TestCaseCrFile(unittest.TestCase):
+
+ def setUp(self):
+ self.tmpdir = tempfile.mkdtemp(prefix="createrepo_ctest-")
+
+ def tearDown(self):
+ shutil.rmtree(self.tmpdir)
+
+ def test_crfile_basic_operations(self):
+ f = cr.CrFile(self.tmpdir+"/foo.gz",
+ cr.MODE_WRITE,
+ cr.GZ_COMPRESSION,
+ None)
+ self.assertTrue(f)
+ self.assertTrue(os.path.isfile(self.tmpdir+"/foo.gz"))
+
+ def test_crfile_operations_on_closed_file(self):
+ # Already closed file
+ path = os.path.join(self.tmpdir, "primary.xml.gz")
+ f = cr.CrFile(path, cr.MODE_WRITE, cr.GZ_COMPRESSION)
+ self.assertTrue(f)
+ self.assertTrue(os.path.isfile(path))
+ f.close()
+
+ self.assertRaises(cr.CreaterepoCError, f.write, "foobar")
+ f.close() # No error should be raised
+ del(f) # No error should be raised
+
+ def test_crfile_error_cases(self):
+ path = os.path.join(self.tmpdir, "foofile")
+ self.assertFalse(os.path.exists(path))
+
+ # Bad open mode
+ self.assertRaises(ValueError, cr.CrFile, path, 86,
+ cr.GZ_COMPRESSION, None)
+ self.assertFalse(os.path.exists(path))
+
+ # Bad compression type
+ self.assertRaises(ValueError, cr.CrFile, path,
+ cr.MODE_READ, 678, None)
+ self.assertFalse(os.path.exists(path))
+
+ # Bad contentstat object
+ self.assertRaises(ValueError, cr.XmlFile, path,
+ cr.MODE_READ, cr.GZ_COMPRESSION, "foo")
+ self.assertFalse(os.path.exists(path))
+
+ # Non existing path
+ self.assertRaises(cr.CreaterepoCError, cr.CrFile,
+ "foobar/foo/xxx/cvydmaticxuiowe")
+
+ def test_crfile_no_compression(self):
+ path = os.path.join(self.tmpdir, "foo")
+ f = cr.CrFile(path, cr.MODE_WRITE, cr.NO_COMPRESSION)
+ self.assertTrue(f)
+ self.assertTrue(os.path.isfile(path))
+ f.write("foobar")
+ f.close()
+
+ content = open(path).read()
+ self.assertEqual(content, "foobar")
+
+ def test_crfile_gz_compression(self):
+ path = os.path.join(self.tmpdir, "foo.gz")
+ f = cr.CrFile(path, cr.MODE_WRITE, cr.GZ_COMPRESSION)
+ self.assertTrue(f)
+ self.assertTrue(os.path.isfile(path))
+ f.write("foobar")
+ f.close()
+
+ import gzip
+ content = gzip.open(path).read()
+ self.assertEqual(content, "foobar")
+
+ def test_crfile_bz2_compression(self):
+ path = os.path.join(self.tmpdir, "foo.bz2")
+ f = cr.CrFile(path, cr.MODE_WRITE, cr.BZ2_COMPRESSION)
+ self.assertTrue(f)
+ self.assertTrue(os.path.isfile(path))
+ f.write("foobar")
+ f.close()
+
+ import bz2
+ content = bz2.decompress(open(path).read())
+ self.assertEqual(content, "foobar")
+
+ def test_crfile_xz_compression(self):
+ path = os.path.join(self.tmpdir, "foo.xz")
+ f = cr.CrFile(path, cr.MODE_WRITE, cr.XZ_COMPRESSION)
+ self.assertTrue(f)
+ self.assertTrue(os.path.isfile(path))
+ f.write("foobar")
+ f.close()
+
+ import subprocess
+ p = subprocess.Popen(["unxz", "--stdout", path], stdout=subprocess.PIPE)
+ content = p.stdout.read()
+ self.assertEqual(content, "foobar")
projects / services / createrepo_c.git / commitdiff