# -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013, Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import shutil
import tarfile
import tempfile
from unittest import TestCase
from ..test_base import PackstackTestCaseMixin
from packstack.installer.core import drones
class SshTarballTransferMixinTestCase(PackstackTestCaseMixin, TestCase):
def setUp(self):
# Creating a temp directory that can be used by tests
self.tempdir = tempfile.mkdtemp()
# prepare resource files
res1path = os.path.join(self.tempdir, 'res1.txt')
with open(res1path, 'w') as f:
f.write('resource one')
resdir = os.path.join(self.tempdir, 'resdir')
os.mkdir(resdir)
res2path = os.path.join(resdir, 'res2.txt')
with open(res2path, 'w') as f:
f.write('resource two')
# prepare recipe files
rec1path = os.path.join(self.tempdir, 'rec1.pp')
with open(rec1path, 'w') as f:
f.write('recipe one')
recdir = os.path.join(self.tempdir, 'recdir')
os.mkdir(recdir)
rec2path = os.path.join(recdir, 'rec2.pp')
with open(rec2path, 'w') as f:
f.write('recipe two')
# prepare class
self.mixin = drones.SshTarballTransferMixin()
self.mixin.node = '127.0.0.1'
self.mixin.resource_dir = os.path.join(self.tempdir, 'remote')
self.mixin.recipe_dir = os.path.join(self.tempdir, 'remote',
'recipes')
self.mixin.local_tmpdir = os.path.join(self.tempdir, 'loctmp')
self.mixin.remote_tmpdir = os.path.join(self.tempdir, 'remtmp')
self.mixin._resources = [(res1path, 'resources'),
(resdir, 'resources')]
self.mixin._recipes = {'one': [rec1path], 'two': [rec2path]}
for i in (self.mixin.resource_dir, self.mixin.recipe_dir,
self.mixin.local_tmpdir, self.mixin.remote_tmpdir):
os.mkdir(i)
def tearDown(self):
# remove the temp directory
# shutil.rmtree(self.tempdir)
pass
def test_tarball_packing(self):
"""
Tests packing of resources and recipes
"""
# pack
res_path = self.mixin._pack_resources()
rec_path = self.mixin._pack_recipes()
# unpack
for pack_path in (res_path, rec_path):
tarball = tarfile.open(pack_path)
tarball.extractall(path=self.tempdir)
# check content of files
for path, content in [('resources/res1.txt', 'resource one'),
('resources/resdir/res2.txt', 'resource two'),
('recipes/rec1.pp', 'recipe one'),
('recipes/rec2.pp', 'recipe two')]:
with open(os.path.join(self.tempdir, path)) as f:
fcont = f.read()
self.assertEqual(fcont, content)
# clean for next test
shutil.rmtree(os.path.join(self.tempdir, 'resources'))
shutil.rmtree(os.path.join(self.tempdir, 'recipes'))
'''
# uncomment this test only on local machines
def test_transfer(self):
"""
Tests resources transfer if sshd to 127.0.0.1 is enabled.
"""
remtmp = self.mixin.resource_dir
server = utils.ScriptRunner('127.0.0.1')
try:
server.append('echo "test"')
server.execute()
except ScripRuntimeError:
return
# transfer
self.mixin._copy_resources()
self.mixin._copy_recipes()
# check content of files
for path, content in \
[('resources/res1.txt', 'resource one'),
('resources/resdir/res2.txt', 'resource two'),
('recipes/rec1.pp', 'recipe one'),
('recipes/rec2.pp', 'recipe two')]:
with open(os.path.join(remtmp, path)) as f:
fcont = f.read()
self.assertEqual(fcont, content)
# clean for next test
server = utils.ScriptRunner('127.0.0.1')
server.append('rm -fr %s' % remtmp)
server.execute()
'''
class FakeDroneObserver(drones.DroneObserver):
def __init__(self, *args, **kwargs):
super(FakeDroneObserver, self).__init__(*args, **kwargs)
self.log = []
def applying(self, drone, recipe):
"""
Drone is calling this method when it starts applying recipe.
"""
self.log.append('applying:%s' % recipe)
def checking(self, drone, recipe):
"""
Drone is calling this method when it starts checking if recipe
has been applied.
"""
self.log.append('checking:%s' % recipe)
def finished(self, drone, recipe):
"""
Drone is calling this method when it's finished with recipe
application.
"""
# subclass must implement this method
self.log.append('finished:%s' % recipe)
class FakeDrone(drones.Drone):
def __init__(self, *args, **kwargs):
super(FakeDrone, self).__init__(*args, **kwargs)
self.log = []
def _apply(self, recipe):
self.log.append(recipe)
def _finished(self, recipe):
return True
class DroneTestCase(PackstackTestCaseMixin, TestCase):
def setUp(self):
super(DroneTestCase, self).setUp()
self.observer = FakeDroneObserver()
self.drone = FakeDrone('127.0.0.1')
self.drone.add_recipe('/some/recipe/path1.rec')
self.drone.add_recipe('/some/recipe/path2.rec')
self.drone.add_recipe('/some/recipe/path3a.rec', marker='test')
self.drone.add_recipe('/some/recipe/path3b.rec', marker='test')
self.drone.add_recipe('/some/recipe/path4a.rec', marker='next')
self.drone.add_recipe('/some/recipe/path4b.rec', marker='next')
self.drone.add_recipe('/some/recipe/path5.rec')
def test_drone_behavior(self):
"""
Tests Drone's recipe application order.
"""
self.drone.log = []
self.drone.apply()
rdir = self.drone.recipe_dir
recpaths = [os.path.join(rdir, os.path.basename(i))
for i in self.drone.recipes]
self.assertListEqual(self.drone.log, recpaths)
def test_observer_behavior(self):
"""
Tests DroneObserver calling.
"""
self.drone.set_observer(self.observer)
self.drone.apply()
rdir = self.drone.recipe_dir.rstrip('/')
first = ['applying:/some/recipe/path1.rec',
'checking:%s/path1.rec' % rdir,
'finished:%s/path1.rec' % rdir,
'applying:/some/recipe/path2.rec',
'checking:%s/path2.rec' % rdir,
'finished:%s/path2.rec' % rdir]
last = ['applying:/some/recipe/path5.rec',
'checking:%s/path5.rec' % rdir,
'finished:%s/path5.rec' % rdir]
self.assertListEqual(self.observer.log[:6], first)
self.assertListEqual(self.observer.log[-3:], last)
self.assertEqual(len(self.observer.log), 21)