Commit 8686738f authored by Rafael Monnerat's avatar Rafael Monnerat

slapos.package: Include much more tests

  Include more tests realeted to signature and upgrade.
parent c52e5405
......@@ -69,11 +69,15 @@ class Promise(BasePromise):
raise ValueError(
"You need upgrade and/or reboot when invoke fixConsistency!")
signature.load()
self.log("Retrying after fixConsistency....\n\n")
return self.checkConsistency(fixit=0, **kw)
def checkConsistency(self, fixit=0, **kw):
# Get configuration
signature = self.getSignature()
self.log("Expected Reboot early them %s" % signature.reboot)
self.log("Expected Upgrade early them %s" % signature.upgrade)
self.log("Last reboot : %s" % signature.last_reboot)
......@@ -81,24 +85,32 @@ class Promise(BasePromise):
if signature.upgrade > datetime.date.today():
self.log("Upgrade will happens on %s" % signature.upgrade)
return
# It is consistent for now
return True
# Check if run for first time
if signature.last_reboot is None:
if fixit:
# Purge repositories list and add new ones
self.fixConsistency(upgrade=1, boot=1)
return self.fixConsistency(upgrade=1, boot=1)
return False
else:
is_ok = True
if signature.last_upgrade < signature.upgrade:
# Purge repositories list and add new ones
self.log('Upgrade is required.')
if fixit:
self.fixConsistency(upgrade=1)
is_ok = self.fixConsistency(upgrade=1)
else:
is_ok = False
else:
self.log("Your system is up to date")
if signature.last_reboot < signature.reboot:
if not self.config.dry_run:
self.fixConsistency(reboot=1)
else:
self.log("Dry run: Rebooting required.")
self.log("Rebooting is required.")
if fixit:
return self.fixConsistency(reboot=1)
return False
else:
self.log("No need to reboot.")
return is_ok
......@@ -170,7 +170,6 @@ class Signature:
"""
info, path = tempfile.mkstemp()
if not self._download(path) == False:
print open(path).read()
return path
else:
raise ValueError("No result from shacache")
......
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2012-2014 Vifib SARL and Contributors.
# All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# 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 3
# 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., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
def _fake_call(self, *args, **kw):
self.last_call = (args, kw)
CONFIGURATION_FILE = """
[networkcache]
download-binary-cache-url = http://www.shacache.org/shacache
download-cache-url = https://www.shacache.org/shacache
download-binary-dir-url = http://www.shacache.org/shadir
signature-certificate-list =
-----BEGIN CERTIFICATE-----
MIIB8DCCAVmgAwIBAgIJAPFf61p8y809MA0GCSqGSIb3DQEBBQUAMBAxDjAMBgNV
BAMMBUNPTVAtMCAXDTE0MDIxNzE2NDgxN1oYDzIxMTQwMTI0MTY0ODE3WjAQMQ4w
DAYDVQQDDAVDT01QLTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAsiqCyuv1
HO9FmtwnMbEa1/u8Dn7T0k7hVKYXVQYof+59Ltbb3cA3nLjFSJDr/wQT6N89MccS
PneRzkWqZKL06Kmj+N+XJfRyVaTz1qQtNzjdbYkO6RgQq+fvq2CO0+PSnL6NttLU
/a9nQMcVm7wZ8kmY+AG5LbVo8lmxDD16Wq0CAwEAAaNQME4wHQYDVR0OBBYEFEVi
YyWHF3W7/O4NaTjn4lElLpp7MB8GA1UdIwQYMBaAFEViYyWHF3W7/O4NaTjn4lEl
Lpp7MAwGA1UdEwQFMAMBAf8wDQYJKoZIhvcNAQEFBQADgYEAgIPGoxhUa16AgjZx
Jr1kUrs8Fg3ig8eRFQlBSLYfANIUxcQ2ScFAkmsvwXY3Md7uaSvMJsEl2jcjdmdi
eSreNkx85j9GtMLY/2cv0kF4yAQNRtibtDkbg6fRNkmUopDosJNVf79l1GKX8JFL
zZBOFdOaLYY/6dLRwiTUKHU6su8=
-----END CERTIFICATE-----
[slapupdate]
upgrade_key = slapos-upgrade-testing-key-with-config-file
"""
UPGRADE_KEY = """[debian-default]
repository-list =
main = http://ftp.fr.debian.org/debian/ wheezy main
main-src = http://ftp.fr.debian.org/debian/ wheezy main
update = http://ftp.fr.debian.org/debian/ wheezy-updates main
update-src = http://ftp.fr.debian.org/debian/ wheezy-updates main
slapos = http://download.opensuse.org/repositories/home:/VIFIBnexedi:/branches:/home:/VIFIBnexedi/Debian_7.0/ ./
re6stnet = http://git.erp5.org/dist/deb ./
key-list =
slapos = http://git.erp5.org/gitweb/slapos.package.git/blob_plain/HEAD:/debian-preseed/slapos.openbuildservice.key
re6st = http://git.erp5.org/gitweb/slapos.package.git/blob_plain/HEAD:/debian-preseed/git.erp5.org.key
filter-package-list =
ntp
openvpn
slapos.node
re6stnet
filter-promise-list =
core
signature-list =
debian+++jessie/sid+++
debian+++7.4+++
debian+++7.5+++
debian+++7.3+++
debian+++7+++
[opensuse-legacy]
repository-list =
suse = http://download.opensuse.org/distribution/12.1/repo/oss/
slapos = http://download.opensuse.org/repositories/home:/VIFIBnexedi:/branches:/home:/VIFIBnexedi/openSUSE_12.1/
re6st = http://git.erp5.org/dist/rpm
key-list =
filter-promise-list =
core
filter-package-list =
ntp
openvpn
slapos.node
re6stnet
signature-list =
opensuse+++12.1+++x86_64
[system]
reboot = 2011-10-10
upgrade = 2014-06-04
"""
......@@ -28,16 +28,44 @@
##############################################################################
from slapos.package.base_promise import BasePromise
from slapos.package.test.base import CONFIGURATION_FILE, UPGRADE_KEY, _fake_call
import os
from slapos.package.signature import NetworkCache
from optparse import Values
import unittest
def _fake_call(self, *args, **kw):
self.last_call = args
FAKE_CALL_COUNTER = 0
def _fake_signature_download(self, path, *args, **kwargs):
global FAKE_CALL_COUNTER
FAKE_CALL_COUNTER += 1
with open(path, 'w') as upgrade_signature:
upgrade_signature.write(UPGRADE_KEY)
return True
class testBasePromiseCase(unittest.TestCase):
def setUp(self):
BasePromise._call = _fake_call
NetworkCache.download = _fake_signature_download
global FAKE_CALL_COUNTER
FAKE_CALL_COUNTER = 0
self.config_dict = {
"slapos_configuration": self._createConfigurationFile(),
"srv_file": "/tmp/test_base_promise_slapupdate",
"dry_run": False,
"verbose": False
}
def _createConfigurationFile(self):
with open("/tmp/test_base_promise_configuration.cfg", "w") as configuration_file:
configuration_file.write(CONFIGURATION_FILE)
return "/tmp/test_base_promise_configuration.cfg"
def testIsSystemd(self):
promise = BasePromise()
......@@ -61,7 +89,8 @@ class testBasePromiseCase(unittest.TestCase):
promise._service("service_name", "service_action")
self.assertEqual(promise.last_call,
(['systemctl', 'service_action', 'service_name'],))
((['systemctl', 'service_action', 'service_name'],),
{'dry_run': False, 'stderr': None, 'stdout': None}))
def testService(self):
promise = BasePromise()
......@@ -73,5 +102,88 @@ class testBasePromiseCase(unittest.TestCase):
promise._service("service_name", "service_action")
self.assertEqual(promise.last_call,
(['service', 'service_name', 'service_action'],))
((['service', 'service_name', 'service_action'],),
{'dry_run': False, 'stderr': None, 'stdout': None}))
def testGetSignature(self):
global FAKE_CALL_COUNTER
promise = BasePromise(Values(self.config_dict))
self.assertEquals(FAKE_CALL_COUNTER, 0)
signature = promise.getSignature()
self.assertNotEquals(signature, None)
self.assertEquals(FAKE_CALL_COUNTER, 1)
# Make sure is already loaded.
self.assertNotEquals(signature.current_state, None)
self.assertEquals(signature.reboot.strftime("%Y-%m-%d"), "2011-10-10" )
# Make sure it do not download things again.
signature = promise.getSignature()
self.assertEquals(FAKE_CALL_COUNTER, 1)
def testGetSlapOSConfiguration(self):
promise = BasePromise(Values(self.config_dict))
self.assertEquals(promise.getSlapOSConfigurationDict("slapupdate"),
{'upgrade_key': 'slapos-upgrade-testing-key-with-config-file'})
def testGetPromiseSectionDict(self):
promise = BasePromise(Values(self.config_dict))
def fake_debian_getOSSignature():
return "debian+++7.4+++"
def fake_opensuse_getOSSignature():
return "opensuse+++12.1+++x86_64"
def fake_unsupported_getOSSignature():
return "readhat+++1+++"
promise.getOSSignature = fake_unsupported_getOSSignature
self.assertEquals(promise.getPromiseSectionDict(), None)
promise.getOSSignature = fake_opensuse_getOSSignature
opensuse_dict = {'filter-promise-list': ['core'],
'filter-package-list': ['ntp', 'openvpn', 'slapos.node', 're6stnet'],
'key-list': [],
'repository-list': [
'suse = http://download.opensuse.org/distribution/12.1/repo/oss/',
'slapos = http://download.opensuse.org/repositories/home:/VIFIBnexedi:/branches:/home:/VIFIBnexedi/openSUSE_12.1/',
're6st = http://git.erp5.org/dist/rpm'],
'signature-list': ['opensuse+++12.1+++x86_64']}
self.assertEquals(promise.getPromiseSectionDict(), opensuse_dict)
promise.getOSSignature = fake_debian_getOSSignature
debian_dict = {'filter-promise-list': ['core'],
'filter-package-list': ['ntp', 'openvpn', 'slapos.node', 're6stnet'],
'key-list': [
'slapos = http://git.erp5.org/gitweb/slapos.package.git/blob_plain/HEAD:/debian-preseed/slapos.openbuildservice.key',
're6st = http://git.erp5.org/gitweb/slapos.package.git/blob_plain/HEAD:/debian-preseed/git.erp5.org.key'],
'repository-list': [
'main = http://ftp.fr.debian.org/debian/ wheezy main',
'main-src = http://ftp.fr.debian.org/debian/ wheezy main',
'update = http://ftp.fr.debian.org/debian/ wheezy-updates main',
'update-src = http://ftp.fr.debian.org/debian/ wheezy-updates main',
'slapos = http://download.opensuse.org/repositories/home:/VIFIBnexedi:/branches:/home:/VIFIBnexedi/Debian_7.0/ ./',
're6stnet = http://git.erp5.org/dist/deb ./'],
'signature-list': ['debian+++jessie/sid+++', 'debian+++7.4+++', 'debian+++7.5+++',
'debian+++7.3+++', 'debian+++7+++']}
self.assertEquals(promise.getPromiseSectionDict(), debian_dict)
def testIsAplicable(self):
from slapos.package.promise import core, limits
def fake_debian_getOSSignature():
return "debian+++7.4+++"
promise = core.Promise(Values(self.config_dict))
promise.getOSSignature = fake_debian_getOSSignature
self.assertEquals(promise.isApplicable(), True)
promise = limits.Promise(Values(self.config_dict))
promise.getOSSignature = fake_debian_getOSSignature
self.assertEquals(promise.isApplicable(), False)
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2012-2014 Vifib SARL and Contributors.
# All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# 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 3
# 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., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from slapos.package.promise import core
from slapos.package.test.base import CONFIGURATION_FILE, UPGRADE_KEY, _fake_call
from slapos.package.signature import NetworkCache
from optparse import Values
import os
import unittest
def _fake_update(self, repository_list=[], package_list=[], key_list=[]):
self._fake_update_call = (repository_list, package_list, key_list)
def fake_debian_getOSSignature():
return "debian+++7.4+++"
def _log_message_list(self, message):
if getattr(self, "_message_list", None) is None:
self._message_list = []
self._message_list.append(message)
class TestCoreCase(unittest.TestCase):
def setUp(self):
core.Promise._call = _fake_call
core.Promise.log = _log_message_list
core.Promise.update = _fake_update
self.original_network_cache_download = NetworkCache.download
# Patch Download
def _fake_signature_download(self, path, *args, **kwargs):
with open(path, 'w') as upgrade_signature:
upgrade_signature.write(UPGRADE_KEY)
return True
NetworkCache.download = _fake_signature_download
self.config_dict = {
"slapos_configuration": self._createConfigurationFile(),
"srv_file": "/tmp/test_base_promise_slapupdate",
"dry_run": False,
"verbose": False
}
def _createConfigurationFile(self):
with open("/tmp/test_base_promise_configuration.cfg", "w") as configuration_file:
configuration_file.write(CONFIGURATION_FILE)
return "/tmp/test_base_promise_configuration.cfg"
def testCheckConsistencyFirstTime(self):
modified_config_dict = self.config_dict.copy()
modified_config_dict["srv_file"] = "/tmp/test_promise_core_which_do_not_exist"
promise = core.Promise(Values(modified_config_dict))
self.assertEquals(promise.checkConsistency(), False)
self.assertEquals(promise._message_list[-2], "Last reboot : None")
self.assertEquals(promise._message_list[-1], "Last upgrade : None")
def testCheckConsistencyAlreadyUpgraded(self):
modified_config_dict = self.config_dict.copy()
slapupdate_path = "/tmp/test_promise_core_testCheckConsistencyAlreadyUpgraded"
with open(slapupdate_path, 'w') as slapupdate:
slapupdate.write("""[system]
upgrade = 2014-11-11
reboot = 2015-11-11
""")
modified_config_dict["srv_file"] = slapupdate_path
promise = core.Promise(Values(modified_config_dict))
self.assertEquals(promise.checkConsistency(), True)
self.assertEquals(promise._message_list[-2], "Your system is up to date")
self.assertEquals(promise._message_list[-1], 'No need to reboot.')
def testCheckConsistencyRebootIsRequired(self):
modified_config_dict = self.config_dict.copy()
slapupdate_path = "/tmp/test_promise_core_testCheckConsistencyRebootIsRequired"
with open(slapupdate_path, 'w') as slapupdate:
slapupdate.write("""[system]
upgrade = 2000-11-11
reboot = 2009-11-11
""")
modified_config_dict["srv_file"] = slapupdate_path
promise = core.Promise(Values(modified_config_dict))
self.assertEquals(promise.checkConsistency(), False)
self.assertEquals(promise._message_list[-1], "Rebooting is required.")
def testCheckConsistencyUpgradeIsRequired(self):
modified_config_dict = self.config_dict.copy()
slapupdate_path = "/tmp/testCheckConsistencyUpgradeIsRequired"
with open(slapupdate_path, 'w') as slapupdate:
slapupdate.write("""[system]
upgrade = 2000-11-11
reboot = 2100-11-11
""")
modified_config_dict["srv_file"] = slapupdate_path
promise = core.Promise(Values(modified_config_dict))
self.assertEquals(promise.checkConsistency(), False)
self.assertEquals(promise._message_list[-1], 'No need to reboot.')
self.assertEquals(promise._message_list[-2], "Upgrade is required.")
def testCheckConsistencyUpgradeInFuture(self):
# Patch Download
def _fake_signature_download(self, path, *args, **kwargs):
with open(path, 'w') as upgrade_signature:
modified_upgrade_key = UPGRADE_KEY.replace("upgrade = 2014-06-04",
"upgrade = 2100-01-01")
upgrade_signature.write(modified_upgrade_key)
return True
NetworkCache.download = _fake_signature_download
modified_config_dict = self.config_dict.copy()
modified_config_dict["srv_file"] = "/tmp/test_promise_core_which_do_not_exist"
promise = core.Promise(Values(modified_config_dict))
self.assertEquals(promise.checkConsistency(), True)
self.assertEquals(promise._message_list[-1], 'Upgrade will happens on 2100-01-01')
def testFixConsistency(self):
modified_config_dict = self.config_dict.copy()
def _fake_update(self, repository_list=[], package_list=[], key_list=[]):
self._fake_update_call = (repository_list, package_list, key_list)
slapupdate_path = "/tmp/testFixConsistencyUpgrade"
with open(slapupdate_path, 'w') as slapupdate:
slapupdate.write("""[system]
upgrade = 2000-11-11
reboot = 2100-11-11
""")
modified_config_dict["srv_file"] = slapupdate_path
promise = core.Promise(Values(modified_config_dict))
self.assertEquals(promise.checkConsistency(fixit=1), True)
expected_message_list = [
'Expected Reboot early them 2011-10-10',
'Expected Upgrade early them 2014-06-04',
'Last reboot : 2100-11-11',
'Last upgrade : 2000-11-11',
'Upgrade is required.',
'Retrying after fixConsistency....\n\n',
'Expected Reboot early them 2011-10-10',
'Expected Upgrade early them 2014-06-04',
'Last reboot : 2100-11-11',
'Last upgrade : 2014-06-14',
'Your system is up to date',
'No need to reboot.',
'No need to reboot.']
self.assertEquals(promise._message_list, expected_message_list)
repository_list = [
('main', 'http://ftp.fr.debian.org/debian/ wheezy main'),
('main-src', 'http://ftp.fr.debian.org/debian/ wheezy main'),
('update', 'http://ftp.fr.debian.org/debian/ wheezy-updates main'),
('update-src', 'http://ftp.fr.debian.org/debian/ wheezy-updates main'),
('slapos', 'http://download.opensuse.org/repositories/home:/VIFIBnexedi:/branches:/home:/VIFIBnexedi/Debian_7.0/ ./'),
('re6stnet', 'http://git.erp5.org/dist/deb ./')]
filter_package_list = ['ntp', 'openvpn', 'slapos.node', 're6stnet']
key_list = [
('slapos', 'http://git.erp5.org/gitweb/slapos.package.git/blob_plain/HEAD:/debian-preseed/slapos.openbuildservice.key'),
('re6st', 'http://git.erp5.org/gitweb/slapos.package.git/blob_plain/HEAD:/debian-preseed/git.erp5.org.key')]
self.assertEquals(promise._fake_update_call[0], repository_list)
self.assertEquals(promise._fake_update_call[1], filter_package_list)
self.assertEquals(promise._fake_update_call[2], key_list)
......@@ -29,6 +29,7 @@
from slapos.package import update, signature
from slapos.libnetworkcache import NetworkcacheClient
from optparse import Values
import slapos.signature
import time
import difflib
......@@ -133,6 +134,8 @@ upgrade = 2014-06-04
"""
def _fake_upload(self, *args, **kwargs):
return True
......@@ -141,6 +144,20 @@ class NetworkCacheTestCase(unittest.TestCase):
def setUp(self):
NetworkcacheClient.upload = _fake_upload
self.config_dict = {
"slapos_configuration": self._createConfigurationFile(),
"srv_file": "/tmp/test_base_promise_slapupdate",
"dry_run": False,
"verbose": False
}
def _createConfigurationFile(self):
with open("/tmp/test_signature_000000_configuration.cfg", "w") as configuration_file:
configuration_file.write(VALID_UPDATE_CFG_DATA)
return "/tmp/test_signature_000000_configuration.cfg"
def test_basic_configuration(self):
info, self.configuration_file_path = tempfile.mkstemp()
open(self.configuration_file_path, 'w').write(UPDATE_CFG_DATA)
......@@ -279,3 +296,82 @@ shadir-key-file = %(tempfile)s
signature.strategy(entry_list),
{'timestamp': 123824.0})
def testLoadSignature(self):
sig = signature.Signature(Values(self.config_dict))
self.assertNotEquals(getattr(sig, "current_state", "COUSOUS!!!"),
"COUSCOUS!!!")
sig.load()
self.assertNotEquals(sig.current_state, None)
self.assertEquals(sig.reboot.strftime("%Y-%m-%d"), "2011-10-10" )
self.assertEquals(sig.upgrade.strftime("%Y-%m-%d"), "2014-06-04" )
self.assertEquals(sig.last_reboot, None )
self.assertEquals(sig.last_upgrade, None )
def testLoadSignatureWithLast(self):
modified_config_dict = self.config_dict.copy()
info, slapupdate_path = tempfile.mkstemp()
with open(slapupdate_path, "w") as slapupdate_file:
slapupdate_file.write("""
[system]
reboot = 2010-10-10
upgrade = 2010-06-04
""")
modified_config_dict["srv_file"] = slapupdate_path
sig = signature.Signature(Values(modified_config_dict))
self.assertNotEquals(getattr(sig, "current_state", "COUSOUS!!!"),
"COUSCOUS!!!")
sig.load()
self.assertNotEquals(sig.current_state, None)
self.assertEquals(sig.reboot.strftime("%Y-%m-%d"), "2011-10-10" )
self.assertEquals(sig.upgrade.strftime("%Y-%m-%d"), "2014-06-04" )
self.assertEquals(sig.last_reboot.strftime("%Y-%m-%d"), "2010-10-10" )
self.assertEquals(sig.last_upgrade.strftime("%Y-%m-%d"), "2010-06-04" )
def testSignatureUpdate(self):
modified_config_dict = self.config_dict.copy()
info, slapupdate_path = tempfile.mkstemp()
with open(slapupdate_path, "w") as slapupdate_file:
slapupdate_file.write("""
[system]
reboot = 2010-10-10
upgrade = 2010-06-04
""")
modified_config_dict["srv_file"] = slapupdate_path
sig = signature.Signature(Values(modified_config_dict))
sig.load()
self.assertNotEquals(sig.current_state, None)
self.assertEquals(sig.reboot.strftime("%Y-%m-%d"), "2011-10-10" )
self.assertEquals(sig.upgrade.strftime("%Y-%m-%d"), "2014-06-04" )
self.assertEquals(sig.last_reboot.strftime("%Y-%m-%d"), "2010-10-10" )
self.assertEquals(sig.last_upgrade.strftime("%Y-%m-%d"), "2010-06-04" )
sig.update(reboot="2014-07-01")
self.assertEquals(open(slapupdate_path, "r").read().strip(), """[system]
reboot = 2014-07-01
upgrade = 2010-06-04""")
sig.update(upgrade="2014-07-03")
self.assertEquals(open(slapupdate_path, "r").read().strip(), """[system]
reboot = 2014-07-01
upgrade = 2014-07-03""")
sig.update(reboot="2014-08-10", upgrade="2014-08-11")
self.assertEquals(open(slapupdate_path, "r").read().strip(), """[system]
reboot = 2014-08-10
upgrade = 2014-08-11""")
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment