Mercurial > treepkg

changeset 414:e0539b483b04 treepkg-status

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
moved data handling in publishdebianpackages into seperate module
author Bjoern Ricks <bricks@intevation.de>
date Mon, 26 Jul 2010 08:28:48 +0000
parents 94a6ae627b31
children 2d31eea39f70
files bin/publishdebianpackages.py test/test_info.py test/test_info_data.py treepkg/info/data.py
diffstat 4 files changed, 161 insertions(+), 123 deletions(-) [+]
line wrap: on
line diff
--- a/bin/publishdebianpackages.py	Fri Jul 23 16:40:38 2010 +0000
+++ b/bin/publishdebianpackages.py	Mon Jul 26 08:28:48 2010 +0000
@@ -12,7 +12,6 @@
 import os
 import sys
 import shlex
-import sqlite3
 
 from optparse import OptionParser
 from ConfigParser import SafeConfigParser
@@ -24,6 +23,8 @@
 from treepkg.publish import *
 from treepkg.util import md5sum
 from treepkg.info.status import TreepkgInfo
+from treepkg.info.data import Package
+from treepkg.info.data import CacheDb
 
 config_desc = ["distribution", "section", "num_newest",
                "build_user", "build_host", "build_listpackages",
@@ -35,90 +36,6 @@
                ("cachedir",
                 lambda s: expand_filename(remove_trailing_slashes(s)))]
 
-class Package:
-    
-    def __init__(self, filename, trackname, packagename, packagepath, arch, md5sum):
-        self.filename = filename
-        self.trackname = trackname
-        self.name = packagename
-        self.sourcepath = packagepath
-        self.arch = arch
-        self.md5sum = md5sum
-
-
-class CacheDb:
-
-
-    def __init__(self, file):
-        self.SELECT_PACKAGE_TMPL = """SELECT * FROM packages 
-                                   WHERE filename = ?"""
-        self.file = file
-        self.conn = sqlite3.connect(file)
-        self.cursor = self.conn.cursor()
-        self.init_db(file)
-
-    def init_db(self, file):
-        self.cursor.execute(
-               """CREATE TABLE IF NOT EXISTS packages (
-                  filename TEXT PRIMARY KEY,
-                  trackname TEXT,
-                  packagename TEXT,
-                  sourcepath TEXT,
-                  arch TEXT, 
-                  md5sum TEXT )""")
-        self.conn.commit()
-
-
-    def add_package(self, package):
-        self.cursor.execute(self.SELECT_PACKAGE_TMPL, (package.filename,))
-        row = self.cursor.fetchone()
-        if not row:
-            self.insert_package(package)
-        else:
-            self.update_package(package)
-
-    def insert_package(self, package):
-        INSERT_TMPL = """INSERT INTO packages VALUES (
-                         ?, ?, ? ,? ,?, ?)"""
-        self.cursor.execute(INSERT_TMPL, (package.filename,
-                            package.trackname,
-                            package.name,
-                            package.sourcepath,
-                            package.arch,
-                            package.md5sum))
-        self.conn.commit()
-
-    def update_package(self, package):
-        UPDATE_TMPL = """UPDATE packages set md5sum = '?' 
-                         WHERE filename in (?)"""
-        self.cursor.execute(UPDATE_TMPL, (package.md5sum, package.filename))
-        self.conn.commit()
-
-    def get_package(self, filename):
-        self.cursor.execute(self.SELECT_PACKAGE_TMPL, (filename,))
-        row = self.cursor.fetchone()
-        if not row:
-            return None
-        return Package(row[0], row[1], row[2], row[3], row[4], row[5]) 
-
-    def get_old_packages(self, newfiles):
-        SELECT_TMPL = """SELECT * FROM packages 
-                      WHERE filename not in (%s)"""
-        tmp = ", ".join(['?'] * len(newfiles))
-        self.cursor.execute(SELECT_TMPL % tmp, newfiles)
-        return [Package(*row) for row in self.cursor.fetchall()]
-
-    def remove_packages(self, files):
-        DELET_TMPL = """DELETE FROM packages 
-                     WHERE filename in (%s)"""
-        tmp = ", ".join(['?'] * len(files))
-        self.cursor.execute(DELET_TMPL % tmp, files)
-        self.conn.commit()
-
-    def get_packages(self):
-        SELECT_TMPL = "SELECT * FROM packages" 
-        self.cursor.execute(SELECT_TMPL)
-        return [Package(*row) for row in self.cursor.fetchall()]
 
 def read_config(filename):
     if not os.path.exists(filename):
@@ -155,7 +72,8 @@
     runremote = prefix_for_remote_command(variables["build_user"],
                                           variables["build_host"])
     xml = capture_output(cmdexpand("@runremote $build_listpackages"
-                                     " --newest=$num_newest",
+                                     " --newest=$num_newest"
+                                     " --only-successful",
                                      runremote=runremote,
                                      **variables))
     return TreepkgInfo.fromxml(xml)
--- a/test/test_info.py	Fri Jul 23 16:40:38 2010 +0000
+++ b/test/test_info.py	Mon Jul 26 08:28:48 2010 +0000
@@ -13,15 +13,12 @@
 
 test_dir = os.path.dirname(__file__)
 sys.path.append(os.path.join(test_dir, os.pardir))
-sys.path.append(os.path.join(test_dir, os.pardir, "bin"))
 
 from treepkg.info.status import *
 from treepkg.report import get_packager_group
 from filesupport import FileTestMixin
 
 from publishdebianpackages import get_binary_arch
-from publishdebianpackages import Package
-from publishdebianpackages import CacheDb
 
 class TreepkgInfoTest(unittest.TestCase, FileTestMixin):
     config_contents = """\
@@ -96,40 +93,6 @@
         binary_armel = get_binary_arch("binary-armel")
         self.assertEquals("binary-armel", binary_armel)
 
-    def test_cache_db(self):
-        tmpdir = self.create_test_specific_temp_dir()
-        package = Package("/tmp/abc/abc_0.1_i386.deb", "abc",
-            "abc_0.1_i386.deb", "/source/abc/abc_0.1_i386.deb", 
-            "binary-i386", "1234567")
-#        tmpfile = self.create_temp_file("cachedb1", "abc")
-#        with self.assertRaises(Exception):
-#            db = CacheDb(tmpfile)
-
-        dbfile = os.path.join(tmpdir, "cachedb2")
-        db = CacheDb(dbfile)
-        db.add_package(package)
-
-        # test get_package and get_timestamp
-        package2 = db.get_package(package.filename)
-        self.assertNotEquals(None, package2)
-        self.assertEquals(package.filename, package2.filename)
-        self.assertEquals(package.name, package2.name)
-        self.assertEquals(package.sourcepath, package2.sourcepath)
-        self.assertEquals(package.arch, package2.arch)
-        self.assertEquals(package.md5sum, package2.md5sum)
-
-        # test get_old_packages
-        package3 = Package("/tmp/foo/foo_0.2.i386.deb", "foo",
-            "foo_0.2_i386.deb", "/tmp/foo/foo_0.2.i386.deb",
-            "binary-i386", "987654321")
-        db.add_package(package3)
-        oldpackages = db.get_old_packages([package.filename])
-        self.assertEquals(1, len(oldpackages))
-        packages = db.get_packages()
-        self.assertEquals(2, len(packages))
-        db.remove_packages([package.filename for package in oldpackages])
-        packages = db.get_packages()
-        self.assertEquals(1, len(packages))
 
 if __name__ == '__main__':
     unittest.main()
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/test/test_info_data.py	Mon Jul 26 08:28:48 2010 +0000
@@ -0,0 +1,60 @@
+# Copyright (C) 2010 by Intevation GmbH
+# Authors:
+# Bjoern Ricks <bjoern.ricks@intevation.de>
+#
+# This program is free software under the GPL (>=v2)
+# Read the file COPYING coming with the software for details.
+
+"""Tests for treepkg.info.data"""
+
+import unittest
+import os.path
+import sys
+
+test_dir = os.path.dirname(__file__)
+sys.path.append(os.path.join(test_dir, os.pardir))
+
+from treepkg.info.data import Package
+from treepkg.info.data import CacheDb
+from filesupport import FileTestMixin
+
+class TestCacheDb(unittest.TestCase, FileTestMixin):
+
+    def test_cache_db(self):
+        tmpdir = self.create_test_specific_temp_dir()
+        package = Package("/tmp/abc/abc_0.1_i386.deb", "abc",
+            "abc_0.1_i386.deb", "/source/abc/abc_0.1_i386.deb", 
+            "binary-i386", "1234567")
+#        tmpfile = self.create_temp_file("cachedb1", "abc")
+#        with self.assertRaises(Exception):
+#            db = CacheDb(tmpfile)
+
+        dbfile = os.path.join(tmpdir, "cachedb2")
+        db = CacheDb(dbfile)
+        db.add_package(package)
+
+        # test get_package and get_timestamp
+        package2 = db.get_package(package.filename)
+        self.assertNotEquals(None, package2)
+        self.assertEquals(package.filename, package2.filename)
+        self.assertEquals(package.name, package2.name)
+        self.assertEquals(package.sourcepath, package2.sourcepath)
+        self.assertEquals(package.arch, package2.arch)
+        self.assertEquals(package.md5sum, package2.md5sum)
+
+        # test get_old_packages
+        package3 = Package("/tmp/foo/foo_0.2.i386.deb", "foo",
+            "foo_0.2_i386.deb", "/tmp/foo/foo_0.2.i386.deb",
+            "binary-i386", "987654321")
+        db.add_package(package3)
+        oldpackages = db.get_old_packages([package.filename])
+        self.assertEquals(1, len(oldpackages))
+        packages = db.get_packages()
+        self.assertEquals(2, len(packages))
+        db.remove_packages([package.filename for package in oldpackages])
+        packages = db.get_packages()
+        self.assertEquals(1, len(packages))
+
+if __name__ == '__main__':
+    unittest.main()
+
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/treepkg/info/data.py	Mon Jul 26 08:28:48 2010 +0000
@@ -0,0 +1,97 @@
+#! /usr/bin/python
+# Copyright (C) 2010 by Intevation GmbH
+# Authors:
+# Bjoern Ricks    <bjoern.ricks@intevation.de>
+#
+# This program is free software under the GPL (>=v2)
+# Read the file COPYING coming with the software for details.
+""" Classes for storing and handling data related to treepkg info  """
+
+import sqlite3
+
+class Package:
+    
+    def __init__(self, filename, trackname, packagename, packagepath,
+                    arch, md5sum):
+        self.filename = filename
+        self.trackname = trackname
+        self.name = packagename
+        self.sourcepath = packagepath
+        self.arch = arch
+        self.md5sum = md5sum
+
+
+class CacheDb:
+
+
+    def __init__(self, file):
+        self.SELECT_PACKAGE_TMPL = """SELECT * FROM packages 
+                                   WHERE filename = ?"""
+        self.file = file
+        self.conn = sqlite3.connect(file)
+        self.cursor = self.conn.cursor()
+        self.init_db()
+
+    def init_db(self):
+        self.cursor.execute(
+               """CREATE TABLE IF NOT EXISTS packages (
+                  filename TEXT PRIMARY KEY,
+                  trackname TEXT,
+                  packagename TEXT,
+                  sourcepath TEXT,
+                  arch TEXT, 
+                  md5sum TEXT )""")
+        self.conn.commit()
+
+
+    def add_package(self, package):
+        self.cursor.execute(self.SELECT_PACKAGE_TMPL, (package.filename,))
+        row = self.cursor.fetchone()
+        if not row:
+            self.insert_package(package)
+        else:
+            self.update_package(package)
+
+    def insert_package(self, package):
+        INSERT_TMPL = """INSERT INTO packages VALUES (
+                         ?, ?, ? ,? ,?, ?)"""
+        self.cursor.execute(INSERT_TMPL, (package.filename,
+                            package.trackname,
+                            package.name,
+                            package.sourcepath,
+                            package.arch,
+                            package.md5sum))
+        self.conn.commit()
+
+    def update_package(self, package):
+        UPDATE_TMPL = """UPDATE packages set md5sum = '?' 
+                         WHERE filename in (?)"""
+        self.cursor.execute(UPDATE_TMPL, (package.md5sum, package.filename))
+        self.conn.commit()
+
+    def get_package(self, filename):
+        self.cursor.execute(self.SELECT_PACKAGE_TMPL, (filename,))
+        row = self.cursor.fetchone()
+        if not row:
+            return None
+        return Package(row[0], row[1], row[2], row[3], row[4], row[5]) 
+
+    def get_old_packages(self, newfiles):
+        SELECT_TMPL = """SELECT * FROM packages 
+                      WHERE filename not in (%s)"""
+        tmp = ", ".join(['?'] * len(newfiles))
+        self.cursor.execute(SELECT_TMPL % tmp, newfiles)
+        return [Package(*row) for row in self.cursor.fetchall()]
+
+    def remove_packages(self, files):
+        DELET_TMPL = """DELETE FROM packages 
+                     WHERE filename in (%s)"""
+        tmp = ", ".join(['?'] * len(files))
+        self.cursor.execute(DELET_TMPL % tmp, files)
+        self.conn.commit()
+
+    def get_packages(self):
+        SELECT_TMPL = "SELECT * FROM packages" 
+        self.cursor.execute(SELECT_TMPL)
+        return [Package(*row) for row in self.cursor.fetchall()]
+
This site is hosted by Intevation GmbH (Datenschutzerklärung und Impressum | Privacy Policy and Imprint)