lorax/tests/lifted/test_queue.py

120 lines
4.9 KiB
Python
Raw Normal View History

#
# Copyright (C) 2019 Red Hat, Inc.
#
# 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, see <http://www.gnu.org/licenses/>.
#
import os
import shutil
import tempfile
import unittest
import lifted.config
from lifted.providers import list_providers
from lifted.queue import _write_callback, create_upload, get_all_uploads, get_upload, get_uploads
from lifted.queue import ready_upload, reset_upload, cancel_upload
import pylorax.api.config
from tests.lifted.profiles import test_profiles
class QueueTestCase(unittest.TestCase):
@classmethod
def setUpClass(self):
self.root_dir = tempfile.mkdtemp(prefix="lifted.test.")
self.config = pylorax.api.config.configure(root_dir=self.root_dir, test_config=True)
self.config.set("composer", "share_dir", os.path.realpath("./share/"))
lifted.config.configure(self.config)
self.upload_uuids = []
@classmethod
def tearDownClass(self):
shutil.rmtree(self.root_dir)
# This should run first, it writes uploads to the queue directory
def test_00_create_upload(self):
"""Test creating an upload for each provider"""
for p in list_providers(self.config["upload"]):
print(p)
upload = create_upload(self.config["upload"], p, "test-image", test_profiles[p][1])
summary = upload.summary()
self.assertEqual(summary["provider_name"], p)
self.assertEqual(summary["image_name"], "test-image")
self.assertTrue(summary["status"], "WAITING")
self.upload_uuids.append(summary["uuid"])
self.assertTrue(len(self.upload_uuids) > 0)
self.assertTrue(len(self.upload_uuids), len(list_providers(self.config["upload"])))
def test_01_get_all_uploads(self):
"""Test listing all the uploads"""
uploads = get_all_uploads(self.config["upload"])
# Should be one upload per provider
providers = sorted([u.provider_name for u in uploads])
self.assertEqual(providers, list_providers(self.config["upload"]))
def test_02_get_upload(self):
"""Test listing specific uploads by uuid"""
for uuid in self.upload_uuids:
upload = get_upload(self.config["upload"], uuid)
self.assertTrue(upload.uuid, uuid)
def test_02_get_upload_error(self):
"""Test listing an unknown upload uuid"""
with self.assertRaises(RuntimeError):
get_upload(self.config["upload"], "not-a-valid-uuid")
def test_03_get_uploads(self):
"""Test listing multiple uploads by uuid"""
uploads = get_uploads(self.config["upload"], self.upload_uuids)
uuids = sorted([u.uuid for u in uploads])
self.assertTrue(uuids, sorted(self.upload_uuids))
def test_04_ready_upload(self):
"""Test ready_upload"""
ready_upload(self.config["upload"], self.upload_uuids[0], "image-test-path")
upload = get_upload(self.config["upload"], self.upload_uuids[0])
self.assertEqual(upload.image_path, "image-test-path")
def test_05_reset_upload(self):
"""Test reset_upload"""
# Set the status to FAILED so it can be reset
upload = get_upload(self.config["upload"], self.upload_uuids[0])
upload.set_status("FAILED", _write_callback(self.config["upload"]))
reset_upload(self.config["upload"], self.upload_uuids[0])
upload = get_upload(self.config["upload"], self.upload_uuids[0])
self.assertEqual(upload.status, "READY")
def test_06_reset_upload_error(self):
"""Test reset_upload raising an error"""
with self.assertRaises(RuntimeError):
reset_upload(self.config["upload"], self.upload_uuids[0])
def test_07_cancel_upload(self):
"""Test cancel_upload"""
cancel_upload(self.config["upload"], self.upload_uuids[0])
upload = get_upload(self.config["upload"], self.upload_uuids[0])
self.assertEqual(upload.status, "CANCELLED")
def test_08_cancel_upload_error(self):
"""Test cancel_upload raises an error"""
# Set the status to CANCELED to make sure the cancel will fail
upload = get_upload(self.config["upload"], self.upload_uuids[0])
upload.set_status("CANCELLED", _write_callback(self.config["upload"]))
with self.assertRaises(RuntimeError):
cancel_upload(self.config["upload"], self.upload_uuids[0])
# TODO test execute