AMBARI-2217. Increase ambari-agent test coverage. (smohanty)

git-svn-id: https://svn.apache.org/repos/asf/incubator/ambari/trunk@1487536 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -904,6 +904,8 @@ Trunk (unreleased changes):
 
  BUG FIXES
 
+ AMBARI-2217. Increase ambari-agent test coverage. (smohanty)
+
  AMBARI-2211. YARN does not show any configurations. (yusaku) 
 
  AMBARI-2210. Hadoop 2 stack version check should be integer based instead of

ambari-agent/src/main/python/ambari_agent/shell.py

@@ -18,7 +18,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 '''
 
-from pwd import getpwnam
 import logging
 import subprocess
 import os
@@ -30,6 +29,10 @@ import time
 import traceback
 import AmbariConfig
 
+try:
+    import pwd
+except ImportError:
+    import winpwd as pwd
 
 global serverTracker
 serverTracker = {}
@@ -77,7 +80,7 @@ class shellRunner:
   def run(self, script, user=None):
     try:
       if user!=None:
-        user=getpwnam(user)[2]
+        user = pwd.getpwnam(user)[2]
       else:
         user = os.getuid()
       threadLocal.uid = user
@@ -93,202 +96,5 @@ class shellRunner:
     logger.debug("Exitcode for %s is %d" % (cmd,code))
     return {'exitCode': code, 'output': out, 'error': err}
 
-  # dispatch action types
-  def runAction(self, clusterId, component, role, 
-                user, command, cleanUpCommand, result):
-    oldDir = os.getcwd()
-    #TODO: handle this better. Don't like that it is doing a chdir for the main process
-    os.chdir(self.getWorkDir(clusterId, role))
-    try:
-      if user is not None:
-        user=getpwnam(user)[2]
-      else:
-        user = oldUid
-      threadLocal.uid = user
-    except Exception:
-      logger.warn("%s %s %s can not switch user for RUN_ACTION." 
-                  % (clusterId, component, role))
-    code = 0
-    cmd = sys.executable
-    tempfilename = tempfile.mktemp()
-    tmp = open(tempfilename, 'w')
-    tmp.write(command['script'])
-    tmp.close()
-    cmd = "%s %s %s" % (cmd, tempfilename, " ".join(command['param']))
-    commandResult = {}
-    p = subprocess.Popen(cmd, preexec_fn=changeUid, stdout=subprocess.PIPE,
-                          stderr=subprocess.PIPE, shell=True, close_fds=True)
-    out, err = p.communicate()
-    code = p.wait()
-    if code != 0:
-      commandResult['output'] = out
-      commandResult['error'] = err
-    commandResult['exitCode'] = code
-    result['commandResult'] = commandResult
-    os.unlink(tempfilename)
-    if code != 0:
-      tempfilename = tempfile.mktemp()
-      tmp = open(tempfilename, 'w')
-      tmp.write(command['script'])
-      tmp.close()
-      cmd = sys.executable
-      cmd = "%s %s %s" % (cmd, tempfilename, " ".join(cleanUpCommand['param']))
-      cleanUpCode = 0
-      cleanUpResult = {}
-      p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-                            shell=True, close_fds=True)
-      out, err = p.communicate()
-      cleanUpCode = p.wait()
-      if cleanUpCode != 0:
-        cleanUpResult['output'] = out
-        cleanUpResult['error'] = err
-      cleanUpResult['exitCode'] = cleanUpCode
-      result['cleanUpResult'] = cleanUpResult
-      os.unlink(tempfilename)
-      os._exit(1)
-    try:
-      os.chdir(oldDir)
-    except Exception:
-      logger.warn("%s %s %s can not restore environment for RUN_ACTION."
-                   % (clusterId, component, role))
-    return result
-
-  # Start a process and presist its state
-  def startProcess(self, clusterId, clusterDefinitionRevision, component,
-                    role, script, user, result):
-    global serverTracker
-    oldDir = os.getcwd()
-    try:
-      os.chdir(self.getWorkDir(clusterId,role))
-    except Exception:
-      logger.warn("%s %s %s can not switch dir for START_ACTION."
-                   % (clusterId, component, role))
-    oldUid = os.getuid()
-    try:
-      if user is not None:
-        user=getpwnam(user)[2]
-      else:
-        user = os.getuid()
-      threadLocal.uid = user
-    except Exception:
-      logger.warn("%s %s %s can not switch user for START_ACTION." 
-                  % (clusterId, component, role))
-    code = 0
-    commandResult = {}
-    process = self.getServerKey(clusterId,clusterDefinitionRevision,
-                                component,role)
-    if not process in serverTracker:
-      try:
-        plauncher = processlauncher(script,user)
-        plauncher.start()
-        plauncher.blockUntilProcessCreation()
-      except Exception:
-        traceback.print_exc()
-        logger.warn("Can not launch process for %s %s %s" 
-                    % (clusterId, component, role))
-        code = -1
-      serverTracker[process] = plauncher
-      commandResult['exitCode'] = code 
-      result['commandResult'] = commandResult
-    try:
-      os.chdir(oldDir)
-    except Exception:
-      logger.warn("%s %s %s can not restore environment for START_ACTION." \
-                   % (clusterId, component, role))
-    return result
-
-  # Stop a process and remove presisted state
-  def stopProcess(self, processKey):
-    global serverTracker
-    keyFragments = processKey.split('/')
-    process = self.getServerKey(keyFragments[0],keyFragments[1],
-                                keyFragments[2],keyFragments[3])
-    if process in serverTracker:
-      logger.info ("Sending %s with PID %d the SIGTERM signal"
-                    % (process,serverTracker[process].getpid()))
-      killprocessgrp(serverTracker[process].getpid())
-      del serverTracker[process]
-
   def getServerTracker(self):
-    return serverTracker
-
-  def getServerKey(self,clusterId, clusterDefinitionRevision, component, role):
-    return clusterId+"/"+str(clusterDefinitionRevision)+"/"+component+"/"+role
-
-  def getWorkDir(self, clusterId, role):
-    prefix = AmbariConfig.config.get('stack','installprefix')
-    return str(os.path.join(prefix, clusterId, role))
-
-
-class processlauncher(threading.Thread):
-  def __init__(self,script,uid):
-    threading.Thread.__init__(self)
-    self.script = script
-    self.serverpid = -1
-    self.uid = uid
-    self.out = None
-    self.err = None
-
-  def run(self):
-    try:
-      tempfilename = tempfile.mktemp()
-      noteTempFile(tempfilename)
-      pythoncmd = sys.executable
-      tmp = open(tempfilename, 'w')
-      tmp.write(self.script['script'])
-      tmp.close()
-      threadLocal.uid = self.uid
-      self.cmd = "%s %s %s" % (pythoncmd, tempfilename,
-                                " ".join(self.script['param']))
-      logger.info("Launching %s as uid %d" % (self.cmd,self.uid) )
-      p = subprocess.Popen(self.cmd,
-                            preexec_fn=self.changeUidAndSetSid, 
-                            stdout=subprocess.PIPE, 
-                            stderr=subprocess.PIPE, shell=True, close_fds=True)
-      logger.info("Launched %s; PID %d" % (self.cmd,p.pid))
-      self.serverpid = p.pid
-      self.out, self.err = p.communicate()
-      self.code = p.wait()
-      logger.info("%s; PID %d exited with code %d \nSTDOUT: %s\nSTDERR %s" % 
-                 (self.cmd,p.pid,self.code,self.out,self.err))
-    except:
-      logger.warn("Exception encountered while launching : " + self.cmd)
-      traceback.print_exc()
-
-    os.unlink(self.getpidfile())
-    os.unlink(tempfilename)
-
-  def blockUntilProcessCreation(self):
-    self.getpid()
- 
-  def getpid(self):
-    sleepCount = 1
-    while (self.serverpid == -1):
-      time.sleep(1)
-      logger.info("Waiting for process %s to start" % self.cmd)
-      if sleepCount > 10:
-        logger.warn("Couldn't start process %s even after %d seconds"
-                     % (self.cmd,sleepCount))
-        os._exit(1)
-    return self.serverpid
-
-  def getpidfile(self):
-    prefix = AmbariConfig.config.get('stack','installprefix')
-    pidfile = os.path.join(prefix,str(self.getpid())+".pid")
-    return pidfile
- 
-  def changeUidAndSetSid(self):
-    prefix = AmbariConfig.config.get('stack','installprefix')
-    pidfile = os.path.join(prefix,str(os.getpid())+".pid")
-    #TODO remove try/except (when there is a way to provide
-    #config files for testcases). The default config will want
-    #to create files in /var/ambari which may not exist unless
-    #specifically created.
-    #At that point add a testcase for the pid file management.
-    try: 
-      f = open(pidfile,'w')
-      f.close()
-    except:
-      logger.warn("Couldn't write pid file %s for %s" % (pidfile,self.cmd))
-    changeUid()
-    os.setsid() 
+    return serverTracker

ambari-agent/src/test/python/TestActionQueue.py

@@ -20,9 +20,12 @@ limitations under the License.
 
 from unittest import TestCase
 from ambari_agent.ActionQueue import ActionQueue
+import ambari_agent.ActionQueue as AQM
 from ambari_agent.AmbariConfig import AmbariConfig
 from ambari_agent.UpgradeExecutor import UpgradeExecutor
+from ambari_agent.PuppetExecutor import PuppetExecutor
 from ambari_agent.StackVersionsFileHandler import StackVersionsFileHandler
+from ambari_agent.ActualConfigHandler import ActualConfigHandler
 import os, errno, time, pprint, tempfile, threading
 import TestStackVersionsFileHandler
 
@@ -35,12 +38,7 @@ class TestActionQueue(TestCase):
     actionQueue.start()
     actionQueue.stop()
     actionQueue.join()
-    self.assertEqual(actionQueue.stopped(), True, 'Action queue is not stopped.') 
-
-#This feature is not yet implemented in ActionQueue
-  def test_RetryAction(self):
-    pass
-
+    self.assertEqual(actionQueue.stopped(), True, 'Action queue is not stopped.')
 
   def test_command_in_progress(self):
     config = AmbariConfig().getConfig()
@@ -191,7 +189,10 @@ class TestActionQueue(TestCase):
 
 
   @patch.object(UpgradeExecutor, "perform_stack_upgrade")
-  def test_upgradeCommand_executeCommand(self, perform_stack_upgrade_method):
+  @patch.object(PuppetExecutor, "runCommand")
+  @patch.object(ActualConfigHandler, "findRunDir")
+  def test_upgradeCommand_executeCommand(self, action_conf_handler_findRunDir_method,
+                                         puppet_executor_run_command_method, perform_stack_upgrade_method):
     queue = ActionQueue(config = MagicMock())
     command = {
       'commandId': 17,
@@ -212,11 +213,13 @@ class TestActionQueue(TestCase):
         'target_stack_version' : 'HDP-1.3.0'
       }
     }
-    perform_stack_upgrade_method.return_value = {
-      'exitcode' : 0,
-      'stdout'   : 'abc',
-      'stderr'   : 'def'
-    }
+
+    upgrade_method_return_value = {'exitcode' : 0,
+                                    'stdout'   : 'abc',
+                                    'stderr'   : 'def'}
+
+    perform_stack_upgrade_method.return_value = upgrade_method_return_value
+
     result = queue.executeCommand(command)
     expected_result = [{'actionId': 17,
                         'clusterName': 'clusterName',
@@ -230,6 +233,55 @@ class TestActionQueue(TestCase):
                         'roleCommand': 'UPGRADE'}]
     self.assertEquals(result, expected_result)
 
+    puppet_executor_run_command_method.return_value = {'exitcode' : 0,
+                                                       'stdout'   : 'abc',
+                                                       'stderr'   : 'def'}
+
+    command['roleCommand'] = 'START'
+    action_conf_handler_findRunDir_method.return_value = AmbariConfig().getConfig().get("stack", "installprefix")
+    expected_result[0]['configurationTags'] = None
+    expected_result[0]['roleCommand'] = 'START'
+    result = queue.executeCommand(command)
+    self.assertEquals(result, expected_result)
+
+    #--------------------------------------------
+    command['roleCommand'] = 'UPGRADE'
+
+    upgrade_method_return_value['exitcode'] = 1
+    upgrade_method_return_value['stdout'] = ''
+    upgrade_method_return_value['stderr'] = ''
+
+    perform_stack_upgrade_method.return_value = upgrade_method_return_value
+    result = queue.executeCommand(command)
+
+    expected_result[0]['roleCommand'] = 'UPGRADE'
+    del expected_result[0]['configurationTags']
+    expected_result[0]['exitCode'] = 1
+    expected_result[0]['stderr'] = 'None'
+    expected_result[0]['stdout'] = 'None'
+    expected_result[0]['status'] = 'FAILED'
+    self.assertEquals(result, expected_result)
+
+
+  @patch.object(ActionQueue, "stopped")
+  @patch.object(AQM.logger, "warn")
+  def test_run_unrecognized_command(self, logger_method, stopped_method):
+    config = AmbariConfig().getConfig()
+    actionQueue = ActionQueue(config)
+    command = {
+        "serviceName" : 'HDFS',
+        "commandType" : "SOME_UNRECOGNIZED_COMMAND",
+        "clusterName" : "",
+        "componentName" : "DATANODE",
+        'configurations':{}
+    }
+    actionQueue.commandQueue.put(command)
+    actionQueue.stopped = stopped_method
+    stopped_method.side_effect = [False, False, True, True, True]
+    actionQueue.IDLE_SLEEP_TIME = 0.001
+    actionQueue.run()
+    self.assertTrue(logger_method.call_args[0][0].startswith('Unrecognized command'))
+
 
   @patch.object(StackVersionsFileHandler, "read_stack_version")
   @patch.object(ActionQueue, "stopped")

ambari-agent/src/test/python/TestPuppetExecutor.py

@@ -25,7 +25,9 @@ from pprint import pformat
 import socket, threading, tempfile
 import os, time
 import sys
+import json
 from AmbariConfig import AmbariConfig
+from mock.mock import patch, MagicMock, call
 from threading import Thread
 
 class TestPuppetExecutor(TestCase):
@@ -40,6 +42,46 @@ class TestPuppetExecutor(TestCase):
     self.assertEquals("--detailed-exitcodes", command[3], "make sure output \
     correct")
 
+  @patch.object(PuppetExecutor, 'runPuppetFile')
+  def test_run_command(self, runPuppetFileMock):
+    tmpdir = AmbariConfig().getConfig().get("stack", "installprefix")
+    puppetInstance = PuppetExecutor("/tmp", "/x", "/y", tmpdir, AmbariConfig().getConfig())
+    jsonFile = open('../../main/python/ambari_agent/test.json', 'r')
+    jsonStr = jsonFile.read()
+    parsedJson = json.loads(jsonStr)
+    parsedJson["taskId"] = 1
+    def side_effect1(puppetFile, result, puppetEnv, tmpoutfile, tmperrfile):
+        result["exitcode"] = 0
+    runPuppetFileMock.side_effect = side_effect1
+    puppetInstance.reposInstalled = False
+    res = puppetInstance.runCommand(parsedJson, tmpdir + '/out.txt', tmpdir + '/err.txt')
+    self.assertEquals(res["exitcode"], 0)
+    self.assertTrue(puppetInstance.reposInstalled)
+
+    def side_effect2(puppetFile, result, puppetEnv, tmpoutfile, tmperrfile):
+        result["exitcode"] = 999
+    runPuppetFileMock.side_effect = side_effect2
+    puppetInstance.reposInstalled = False
+    res = puppetInstance.runCommand(parsedJson, tmpdir + '/out.txt', tmpdir + '/err.txt')
+    self.assertEquals(res["exitcode"], 999)
+    self.assertFalse(puppetInstance.reposInstalled)
+    os.unlink(tmpdir + os.sep + 'site-' + str(parsedJson["taskId"]) + '.pp')
+
+  @patch("os.path.exists")
+  def test_configure_environ(self, osPathExistsMock):
+    config = AmbariConfig().getConfig()
+    tmpdir = config.get("stack", "installprefix")
+    puppetInstance = PuppetExecutor("/tmp", "/x", "/y", tmpdir, config)
+    environ = puppetInstance.configureEnviron({})
+    self.assertEquals(environ, {})
+
+    config.set('puppet','ruby_home',"test/ruby_home")
+    puppetInstance = PuppetExecutor("/tmp", "/x", "/y", tmpdir, config)
+    osPathExistsMock.return_value = True
+    environ = puppetInstance.configureEnviron({"PATH" : "test_path"})
+    self.assertEquals(environ["PATH"], "test/ruby_home/bin:test_path")
+    self.assertEquals(environ["MY_RUBY_HOME"], "test/ruby_home")
+
   def test_condense_bad2(self):
     puppetexecutor = PuppetExecutor("/tmp", "/x", "/y", "/z", AmbariConfig().getConfig())
     grep = Grep()

ambari-agent/src/test/python/TestPythonExecutor.py

@@ -111,6 +111,19 @@ class TestPythonExecutor(TestCase):
     self.assertEquals(result, {'exitcode': 0, 'stderr': 'Dummy err', 'stdout': 'Dummy output'})
 
 
+  def test_is_successfull(self):
+    executor = PythonExecutor("/tmp", AmbariConfig().getConfig())
+
+    executor.python_process_has_been_killed = False
+    self.assertTrue(executor.isSuccessfull(0))
+    self.assertFalse(executor.isSuccessfull(1))
+
+    executor.python_process_has_been_killed = True
+    self.assertFalse(executor.isSuccessfull(0))
+    self.assertFalse(executor.isSuccessfull(1))
+
+
+
   class Subprocess_mockup():
     """
     It's not trivial to use PyMock instead of class here because we need state

ambari-agent/src/test/python/TestRepoInstaller.py

@@ -21,7 +21,7 @@ limitations under the License.
 from unittest import TestCase
 from ambari_agent.RepoInstaller import RepoInstaller
 import tempfile
-import json
+import json, os
 import shutil
 from ambari_agent.AmbariConfig import AmbariConfig
 from mock.mock import patch, MagicMock, call
@@ -42,6 +42,31 @@ class TestRepoInstaller(TestCase):
     pass
 
 
+  def test_prepare_repos_info(self):
+    localParsedJson = json.loads('{"hostLevelParams" : {"repo_info" : {"test" : "test"}}}')
+    localRepoInstaller = RepoInstaller(localParsedJson, self.dir, '../../main/puppet/modules', 1, self.config)
+    localRepoInstaller.prepareReposInfo()
+    self.assertEquals(localRepoInstaller.repoInfoList['test'], "test")
+
+    localParsedJson = json.loads('{"hostLevelParams" : {"repo_info" : "1"}}')
+    localRepoInstaller = RepoInstaller(localParsedJson, self.dir, '../../main/puppet/modules', 1, self.config)
+    localRepoInstaller.prepareReposInfo()
+    self.assertEquals(localRepoInstaller.repoInfoList, 1)
+
+    localParsedJson = json.loads('{"hostLevelParams" : {"repo_info" : ""}}')
+    localRepoInstaller = RepoInstaller(localParsedJson, self.dir, '../../main/puppet/modules', 1, self.config)
+    localRepoInstaller.prepareReposInfo()
+    self.assertEquals(localRepoInstaller.repoInfoList, [])
+
+
+  def test_generate_files(self):
+    localParsedJson = json.loads('{"hostLevelParams": { "repo_info" : [{"baseUrl":"http://public-repo-1.hortonworks.com/HDP-1.1.1.16/repos/centos5"\
+           ,"osType":"centos5","repoId":"HDP-1.1.1.16_TEST","repoName":"HDP_TEST", "mirrorsList":"http://mirrors.fedoraproject.org/mirrorlist"}]}}')
+    localRepoInstaller = RepoInstaller(localParsedJson, self.dir, '../../main/puppet/modules', 1, self.config)
+    localRepoInstaller.prepareReposInfo()
+    result = localRepoInstaller.generateFiles()
+    self.assertTrue(result[0].endswith("HDP-1.1.1.16_TEST-1.pp"))
+
   @patch.object(RepoInstaller, 'prepareReposInfo')
   @patch.object(RepoInstaller, 'generateFiles')
   def testInstallRepos(self, generateFilesMock, prepareReposInfoMock):

ambari-agent/src/test/python/TestShell.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python2.6
+# -*- coding: utf-8 -*-
+
+'''
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you 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 unittest
+import tempfile
+from mock.mock import patch, MagicMock, call
+from ambari_agent.AmbariConfig import AmbariConfig
+from ambari_agent import shell
+from shell import shellRunner
+
+
+
+class TestShell(unittest.TestCase):
+
+  @patch("os.killpg")
+  @patch("time.sleep")
+  def test_kill_stale_process(self, timeSleepMock, os_killPgMock):
+    temp_path = AmbariConfig().getConfig().get("stack", "installprefix") + '/9999.pid'
+    file = open(temp_path, 'w')
+    file.close()
+
+    shell.killstaleprocesses()
+    self.assertFalse(os.path.exists(temp_path))
+
+
+  @patch("os.setuid")
+  def test_changeUid(self, os_setUIDMock):
+    shell.threadLocal.uid = 9999
+    shell.changeUid()
+    self.assertTrue(os_setUIDMock.called)
+
+
+  @patch("pwd.getpwnam")
+  def test_shellRunner_run(self, getpwnamMock):
+    sh = shellRunner()
+    result = sh.run(['echo'])
+    self.assertEquals(result['exitCode'], 0)
+    self.assertEquals(result['error'], '')
+
+    getpwnamMock.return_value = [os.getuid(), os.getuid(), os.getuid()]
+    result = sh.run(['echo'], 'non_exist_user_name')
+    self.assertEquals(result['exitCode'], 0)
+    self.assertEquals(result['error'], '')

ambari-agent/src/test/python/winpwd.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python2.6
+
+'''
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you 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.
+'''
+
+
+
+
+
+def getpwnam(user):
+    print("YESSSSSSSS")
+
+
+
+