'''
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 platform

IS_WINDOWS = platform.system() == "Windows"

from unittest import TestCase
from mock.mock import patch, MagicMock, call

from resource_management.core.system import System
from resource_management.core.resources.system import Execute
from resource_management.core.environment import Environment

import subprocess
import logging
import os
from resource_management import Fail

if not IS_WINDOWS:
  import grp
  import pwd


@patch.object(System, "os_family", new='redhat')
class TestExecuteResource(TestCase):
  @patch.object(logging.Logger, "info")
  @patch.object(subprocess, "Popen")
  def test_attribute_logoutput(self, popen_mock, info_mock):
    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    subproc_mock.communicate.side_effect = [["1"], ["2"]]
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      Execute('echo "1"',
              logoutput=True)
      Execute('echo "2"',
              logoutput=False)

    info_mock.assert_called('1')
    self.assertTrue("call('2')" not in str(info_mock.mock_calls))
    
  @patch('subprocess.Popen.communicate')
  @patch('subprocess.Popen')
  def test_attribute_wait(self, popen_mock, proc_communicate_mock):
    with Environment("/") as env:
      Execute('echo "1"',
              wait_for_finish=False)
      Execute('echo "2"',
              wait_for_finish=False)
    
    self.assertTrue(popen_mock.called, 'subprocess.Popen should have been called!')
    self.assertFalse(proc_communicate_mock.called, 'proc.communicate should not have been called!')

  @patch('subprocess.Popen.communicate')
  @patch('subprocess.Popen')
  def test_attribute_wait_and_poll(self, popen_mock, proc_communicate_mock):
    with Environment("/") as env:
      try:
        Execute('echo "1"',
                wait_for_finish=False,
                poll_after = 5)
        self.assertTrue(False, "Should fail as process does not run for 5 seconds")
      except Fail as e:
        self.assertTrue("returned 1" in str(e))
        pass

    self.assertTrue(popen_mock.called, 'subprocess.Popen should have been called!')
    self.assertFalse(proc_communicate_mock.called, 'proc.communicate should not have been called!')

  if not IS_WINDOWS:
    @patch('subprocess.Popen.communicate')
    def test_attribute_wait_and_poll_and_success(self, proc_communicate_mock):
      with Environment("/") as env:
        Execute('sleep 6',
                wait_for_finish=False,
                poll_after = 2)

      self.assertFalse(proc_communicate_mock.called, 'proc.communicate should not have been called!')

  @patch.object(os.path, "exists")
  @patch.object(subprocess, "Popen")
  def test_attribute_creates(self, popen_mock, exists_mock):
    exists_mock.return_value = True

    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    subproc_mock.communicate.side_effect = [["1"]]
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      Execute('echo "1"',
              creates="/must/be/created")

    exists_mock.assert_called_with("/must/be/created")
    self.assertEqual(subproc_mock.call_count, 0)

  @patch.object(subprocess, "Popen")
  def test_attribute_path(self, popen_mock):
    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    subproc_mock.communicate.side_effect = [["1"]]
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      execute_resource = Execute('echo "1"',
                                 path=["/test/one", "test/two"]
      )

    if IS_WINDOWS:
      self.assertEqual(execute_resource.environment["PATH"], '/test/one;test/two')
    else:
      self.assertEqual(execute_resource.environment["PATH"], '/test/one:test/two')

  @patch('time.sleep')
  @patch.object(subprocess, "Popen")
  def test_attribute_try_sleep_tries(self, popen_mock, time_mock):
    expected_call = "call('Retrying after %d seconds. Reason: %s', 1, 'Fail')"

    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    subproc_mock.communicate.side_effect = [Fail("Fail"), ["1"]]
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      Execute('echo "1"',
              tries=2,
              try_sleep=10
      )
    pass

    time_mock.assert_called_once_with(10)

  if not IS_WINDOWS:
    @patch.object(pwd, "getpwnam")
    def test_attribute_group(self, getpwnam_mock):
      def error(argument):
        self.assertEqual(argument, "test_user")
        raise KeyError("fail")

      getpwnam_mock.side_effect = error
      try:
        with Environment("/") as env:
          Execute('echo "1"',
                  user="test_user",
          )
      except Fail as e:
        pass

    @patch.object(grp, "getgrnam")
    @patch.object(pwd, "getpwnam")
    def test_attribute_group(self, getpwnam_mock, getgrnam_mock):
      def error(argument):
        self.assertEqual(argument, "test_group")
        raise KeyError("fail")

      getpwnam_mock.side_effect = 1
      getgrnam_mock.side_effect = error
      try:
        with Environment("/") as env:
          Execute('echo "1"',
                  group="test_group",
          )
      except Fail as e:
        pass
  

  @patch.object(subprocess, "Popen")
  def test_attribute_environment(self, popen_mock):
    expected_dict = {"JAVA_HOME": "/test/java/home"}

    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    subproc_mock.communicate.side_effect = [["1"]]
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      Execute('echo "1"',
              environment=expected_dict
      )

    self.assertEqual(popen_mock.call_args_list[0][1]["env"], expected_dict)
    pass

  @patch.object(subprocess, "Popen")
  def test_attribute_cwd(self, popen_mock):
    expected_cwd = "/test/work/directory"

    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    subproc_mock.communicate.side_effect = [["1"]]
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      Execute('echo "1"',
              cwd=expected_cwd
      )

    self.assertEqual(popen_mock.call_args_list[0][1]["cwd"], expected_cwd)

  @patch.object(subprocess, "Popen")
  def test_attribute_command_escaping(self, popen_mock):
    expected_command0 = "arg1 arg2 'quoted arg'"
    expected_command1 = "arg1 arg2 'command \"arg\"'"
    expected_command2 = 'arg1 arg2 \'command \'"\'"\'arg\'"\'"\'\''
    expected_command3 = "arg1 arg2 'echo `ls /root`'"
    expected_command4 = "arg1 arg2 '$ROOT'"
    expected_command5 = "arg1 arg2 '`ls /root`'"

    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      Execute(('arg1', 'arg2', 'quoted arg'),
      )
      Execute(('arg1', 'arg2', 'command "arg"'),
      )
      Execute(('arg1', 'arg2', "command 'arg'"),
      )
      Execute(('arg1', 'arg2', "echo `ls /root`"),
      )
      Execute(('arg1', 'arg2', "$ROOT"),
      )
      Execute(('arg1', 'arg2', "`ls /root`"),
      )

    self.assertEqual(popen_mock.call_args_list[0][0][0][3], expected_command0)
    self.assertEqual(popen_mock.call_args_list[1][0][0][3], expected_command1)
    self.assertEqual(popen_mock.call_args_list[2][0][0][3], expected_command2)
    self.assertEqual(popen_mock.call_args_list[3][0][0][3], expected_command3)
    self.assertEqual(popen_mock.call_args_list[4][0][0][3], expected_command4)
    self.assertEqual(popen_mock.call_args_list[5][0][0][3], expected_command5)

  @patch.object(subprocess, "Popen")
  def test_attribute_command_one_line(self, popen_mock):
    expected_command = "rm -rf /somedir"

    subproc_mock = MagicMock()
    subproc_mock.returncode = 0
    popen_mock.return_value = subproc_mock

    with Environment("/") as env:
      Execute(expected_command)

    self.assertEqual(popen_mock.call_args_list[0][0][0][3], expected_command)