import pysqlite2
from storm.database import create_database
-from storm.locals import Bool, Int, Unicode
+from storm.locals import Bool, Int, Unicode, RawStr
from storm.store import Store
import time
"""A host in the buildfarm."""
def __init__(self, name, owner=None, owner_email=None, password=None, platform=None,
- ssh_access=False, last_update=None, fqdn=None):
+ ssh_access=False, last_update=None, fqdn=None, join_time=None, permission=None):
self.name = name
if owner:
self.owner = (owner, owner_email)
else:
self.owner = None
+ if join_time is None:
+ self.join_time = time.time()
+ else:
+ self.join_time = join_time
+ self.permission = permission
self.password = password
self.platform = platform
self.ssh_access = ssh_access
class StormHost(Host):
__storm_table__ = "host"
- name = Unicode()
- owner = Unicode()
+ name = Unicode(primary=True)
+ owner_name = Unicode(name="owner")
owner_email = Unicode()
password = Unicode()
ssh_access = Bool()
- fqdn = Unicode()
+ fqdn = RawStr()
platform = Unicode()
permission = Unicode()
last_dead_mail = Int()
join_time = Int()
+ def _set_owner(self, value):
+ if value is None:
+ self.owner_name = None
+ self.owner_email = None
+ else:
+ (self.owner_name, self.owner_email) = value
+
+ def _get_owner(self):
+ if self.owner_name is None:
+ return None
+ else:
+ return (self.owner_name, self.owner_email)
+
+ owner = property(_get_owner, _set_owner)
+
class HostDatabase(object):
"""Host database."""
self.store.flush()
def createhost(self, name, platform=None, owner=None, owner_email=None, password=None, permission=None):
+ newhost = StormHost(unicode(name), owner=owner, owner_email=owner_email, password=password, permission=permission, platform=platform)
try:
- self.store.execute("INSERT INTO host (name, platform, owner, owner_email, password, permission, join_time) VALUES (?,?,?,?,?,?,?)",
- (name, platform, owner, owner_email, password, permission, time.time()))
- except (pysqlite2.dbapi2.IntegrityError,):
+ self.store.add(newhost)
+ self.store.flush()
+ except pysqlite2.dbapi2.IntegrityError:
raise HostAlreadyExists(name)
- self.store.flush()
def deletehost(self, name):
- cursor = self.store.execute("DELETE FROM host WHERE name = ?", (name,))
- if cursor.rowcount == 0:
+ host = self.host(name)
+ if host is None:
raise NoSuchHost(name)
- self.store.flush()
+ self.store.remove(host)
def hosts(self):
- cursor = self.store.execute("SELECT name, owner, owner_email, password, platform, ssh_access, fqdn FROM host ORDER BY name")
- for row in cursor:
- yield Host(name=row[0], owner=row[1], owner_email=row[2], password=row[3], platform=row[4], ssh_access=bool(row[5]), fqdn=row[6])
+ return self.store.find(StormHost).order_by(StormHost.name)
def dead_hosts(self, age):
dead_time = time.time() - age
self.store.flush()
def host(self, name):
- for host in self.hosts():
- if host.name == name:
- return host
- return None
+ return self.hosts().find(StormHost.name==name).one()
def update_platform(self, name, new_platform):
- cursor = self.store.execute("UPDATE host SET platform = ? WHERE name = ?", (new_platform, name))
- if cursor.rowcount == 0:
+ host = self.host(unicode(name))
+ if host is None:
raise NoSuchHost(name)
- self.store.flush()
+ host.platform = new_platform
def update_owner(self, name, new_owner, new_owner_email):
cursor = self.store.execute(
"""Write out the web/"""
for host in self.hosts():
- yield "%s: %s\n" % (host.name, host.platform)
+ yield "%s: %s\n" % (host.name.encode("utf-8"), host.platform.encode("utf-8"))
class HostTests(unittest.TestCase):
def test_create_simple(self):
- host = hostdb.Host(name="foo")
+ host = hostdb.Host(name=u"foo")
self.assertEquals(None, host.owner)
self.assertEquals("foo", host.name)
def test_create_with_owner(self):
- host = hostdb.Host(name="foo", owner="Jelmer", owner_email="jelmer@samba.org")
- self.assertEquals(("Jelmer", "jelmer@samba.org"), host.owner)
- self.assertEquals("foo", host.name)
+ host = hostdb.Host(name="foo", owner=u"Jelmer", owner_email=u"jelmer@samba.org")
+ self.assertEquals((u"Jelmer", u"jelmer@samba.org"), host.owner)
+ self.assertEquals(u"foo", host.name)
class DatabaseTests(unittest.TestCase):
self.db = hostdb.HostDatabase()
def test_createhost(self):
- self.db.createhost("charis", "linux", "Jelmer", "jelmer@samba.org", "bla", "Pemrission?")
+ self.db.createhost(u"charis", u"linux", u"Jelmer", u"jelmer@samba.org", u"bla", u"Pemrission?")
hosts = list(self.db.hosts())
self.assertEquals(1, len(hosts))
self.assertEquals("charis", hosts[0].name)
def test_create_already_exists(self):
- host = self.db.createhost(name="foo", owner="Jelmer", owner_email="jelmer@samba.org")
- self.assertRaises(hostdb.HostAlreadyExists, self.db.createhost, name="foo",
- owner="Jelmer", owner_email="jelmer@samba.org")
+ host = self.db.createhost(name=u"foo", owner=u"Jelmer", owner_email=u"jelmer@samba.org")
+ self.assertRaises(hostdb.HostAlreadyExists, self.db.createhost, name=u"foo",
+ owner=u"Jelmer", owner_email=u"jelmer@samba.org")
def test_delete(self):
- host = self.db.createhost(name="foo", owner="Jelmer", owner_email="jelmer@samba.org")
- self.db.deletehost("foo")
+ host = self.db.createhost(name=u"foo", owner=u"Jelmer", owner_email=u"jelmer@samba.org")
+ self.db.deletehost(u"foo")
def test_delete_doesntexist(self):
- self.assertRaises(hostdb.NoSuchHost, self.db.deletehost, "foo")
+ self.assertRaises(hostdb.NoSuchHost, self.db.deletehost, u"foo")
def test_update_platform(self):
- host = self.db.createhost(name="foo", owner="Jelmer",
- owner_email="jelmer@samba.org")
- self.db.update_platform(name="foo", new_platform="Debian")
+ host = self.db.createhost(name=u"foo", owner=u"Jelmer",
+ owner_email=u"jelmer@samba.org")
+ self.db.update_platform(name=u"foo", new_platform=u"Debian")
def test_update_platform_doesnt_exist(self):
self.assertRaises(hostdb.NoSuchHost, self.db.update_platform, name="foo",
new_platform="Debian")
def test_update_owner(self):
- host = self.db.createhost(name="foo", owner="Jelmer", owner_email="jelmer@samba.org")
- self.db.update_owner(name="foo", new_owner="Matthieu",
- new_owner_email="mat@samba.org")
+ host = self.db.createhost(name=u"foo", owner=u"Jelmer", owner_email=u"jelmer@samba.org")
+ self.db.update_owner(name=u"foo", new_owner=u"Matthieu",
+ new_owner_email=u"mat@samba.org")
def test_update_owner_doesnt_exist(self):
self.assertRaises(hostdb.NoSuchHost, self.db.update_owner, name="foo",
new_owner="Mat", new_owner_email="mat@samba.org")
def test_create_hosts_list(self):
- self.db.createhost(name="foo", owner="Jelmer", owner_email="jelmer@samba.org",
- platform="Debian")
- self.db.createhost(name="bla", owner="Jelmer", owner_email="jelmer@samba.org",
- platform="Fedora")
+ self.db.createhost(name=u"foo", owner=u"Jelmer", owner_email=u"jelmer@samba.org",
+ platform=u"Debian")
+ self.db.createhost(name=u"bla", owner=u"Jelmer", owner_email=u"jelmer@samba.org",
+ platform=u"Fedora")
expected = [
"foo: Debian\n",
"bla: Fedora\n"]
self.assertEquals(expected, got)
def test_create_rsync_secrets(self):
- self.db.createhost(name="foo")
- self.db.createhost(name="bla", owner="Jelmer", owner_email="jelmer@samba.org",
- platform="Fedora", password="o")
+ self.db.createhost(name=u"foo")
+ self.db.createhost(name=u"bla", owner=u"Jelmer", owner_email=u"jelmer@samba.org",
+ platform=u"Fedora", password=u"o")
expected = [
"# rsyncd.secrets file\n",
"# automatically generated by textfiles.pl. DO NOT EDIT!\n\n",