From 985ed2669a7aae0acd6e6bd6f7b136251a96d328 Mon Sep 17 00:00:00 2001 From: Steven Armstrong Date: Wed, 12 Oct 2011 15:17:06 +0200 Subject: [PATCH] local code execution and tests Signed-off-by: Steven Armstrong --- lib/cdist/exec/local.py | 118 ++++++++++++++++++++++++++++++ lib/cdist/test/exec/test_local.py | 102 ++++++++++++++++++++++++++ 2 files changed, 220 insertions(+) create mode 100644 lib/cdist/exec/local.py create mode 100644 lib/cdist/test/exec/test_local.py diff --git a/lib/cdist/exec/local.py b/lib/cdist/exec/local.py new file mode 100644 index 00000000..b9c511f4 --- /dev/null +++ b/lib/cdist/exec/local.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +# +# 2011 Steven Armstrong (steven-cdist at armstrong.cc) +# 2011 Nico Schottelius (nico-cdist at schottelius.org) +# +# This file is part of cdist. +# +# cdist 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. +# +# cdist 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 cdist. If not, see . +# +# + +import io +import os +import sys +import subprocess +import shutil +import logging + +import cdist + + +class LocalScriptError(cdist.Error): + def __init__(self, script, command, script_content): + self.script = script + self.command = command + self.script_content = script_content + + def __str__(self): + return "Remote script execution failed: %s %s" % (self.script, self.command) + + +class Local(object): + """Execute commands locally. + + All interaction with the local side should be done through this class. + Directly accessing the local side from python code is a bug. + + """ + def __init__(self, target_host, local_base_path, out_path): + self.target_host = target_host + self.base_path = local_base_path + + # Local input + self.cache_path = os.path.join(self.base_path, "cache") + self.conf_path = os.path.join(self.base_path, "conf") + self.global_explorer_path = os.path.join(self.conf_path, "explorer") + self.manifest_path = os.path.join(self.conf_path, "manifest") + self.type_base_path = os.path.join(self.conf_path, "type") + # FIXME: should not be needed anywhere + self.lib_path = os.path.join(self.base_path, "lib") + + # Local output + self.out_path = out_path + self.bin_path = os.path.join(self.out_path, "bin") + self.global_explorer_out_path = os.path.join(self.out_path, "explorer") + self.object_base_path = os.path.join(self.out_path, "object") + + self.log = logging.getLogger(self.target_host) + + def create_directories(self): + self.mkdir(self.out_path) + self.mkdir(self.bin_path) + + def rmdir(self, path): + """Remove directory on the local side.""" + self.log.debug("Local rmdir: %s", path) + shutil.rmtree(path) + + def mkdir(self, path): + """Create directory on the local side.""" + self.log.debug("Local mkdir: %s", path) + os.makedirs(path, mode=0o700, exist_ok=True) + + def run(self, command, env=None): + """Run the given command with the given environment. + Return the output as a string. + + """ + assert isinstance(command, (list, tuple)), "list or tuple argument expected, got: %s" % command + self.log.debug("Local run: %s", command) + try: + return subprocess.check_output(command, env=env) + except subprocess.CalledProcessError: + raise cdist.Error("Command failed: " + " ".join(command)) + except OSError as error: + raise cdist.Error(" ".join(*args) + ": " + error.args[1]) + + def run_script(self, script, env=None): + """Run the given script with the given environment. + Return the output as a string. + + """ + command = ["/bin/sh", "-e"] + command.append(script) + + self.log.debug("Local run script: %s", command) + if env: + self.log.debug("Local run script env: %s", env) + + try: + return subprocess.check_output(command, env=env) + except subprocess.CalledProcessError as error: + script_content = self.run(["cat", script]) + self.log.error("Code that raised the error:\n%s", script_content) + raise LocalScriptError(script, command, script_content) + except EnvironmentError as error: + raise cdist.Error(" ".join(command) + ": " + error.args[1]) diff --git a/lib/cdist/test/exec/test_local.py b/lib/cdist/test/exec/test_local.py new file mode 100644 index 00000000..2c021404 --- /dev/null +++ b/lib/cdist/test/exec/test_local.py @@ -0,0 +1,102 @@ +# -*- coding: utf-8 -*- +# +# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc) +# +# This file is part of cdist. +# +# cdist 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. +# +# cdist 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 cdist. If not, see . +# +# + +import unittest +import os +import tempfile +import getpass +import shutil +import string +import random + +#import logging +#logging.basicConfig(level=logging.DEBUG, format='%(levelname)s: %(message)s') + +import cdist +from cdist.exec import local + +import os.path as op +my_dir = op.abspath(op.dirname(__file__)) +fixtures = op.join(my_dir, 'fixtures') +local_base_path = fixtures + + +class LocalTestCase(unittest.TestCase): + + def mkdtemp(self, **kwargs): + return tempfile.mkdtemp(prefix='tmp.cdist.test.', **kwargs) + + def mkstemp(self, **kwargs): + return tempfile.mkstemp(prefix='tmp.cdist.test.', **kwargs) + + def setUp(self): + self.temp_dir = self.mkdtemp() + target_host = 'localhost' + out_path = self.temp_dir + self.local = local.Local(target_host, local_base_path, out_path) + + def tearDown(self): + #shutil.rmtree(self.temp_dir) + pass + + def test_run_success(self): + self.local.run(['/bin/true']) + + def test_run_fail(self): + self.assertRaises(cdist.Error, self.local.run, ['/bin/false']) + + def test_run_script_success(self): + handle, script = self.mkstemp(dir=self.temp_dir) + fd = open(script, "w") + fd.writelines(["#!/bin/sh\n", "/bin/true"]) + fd.close() + self.local.run_script(script) + + def test_run_script_fail(self): + handle, script = self.mkstemp(dir=self.temp_dir) + fd = open(script, "w") + fd.writelines(["#!/bin/sh\n", "/bin/false"]) + fd.close() + self.assertRaises(local.LocalScriptError, self.local.run_script, script) + + def test_run_script_get_output(self): + handle, script = self.mkstemp(dir=self.temp_dir) + fd = open(script, "w") + fd.writelines(["#!/bin/sh\n", "echo foobar"]) + fd.close() + self.assertEqual(self.local.run_script(script), b"foobar\n") + + def test_mkdir(self): + temp_dir = self.mkdtemp(dir=self.temp_dir) + os.rmdir(temp_dir) + self.local.mkdir(temp_dir) + self.assertTrue(os.path.isdir(temp_dir)) + + def test_rmdir(self): + temp_dir = self.mkdtemp(dir=self.temp_dir) + self.local.rmdir(temp_dir) + self.assertFalse(os.path.isdir(temp_dir)) + + def test_create_directories(self): + self.local.create_directories() + print('self.local.bin_path: %s' % self.local.bin_path) + self.assertTrue(os.path.isdir(self.local.out_path)) + self.assertTrue(os.path.isdir(self.local.bin_path))