<commit_before>#!/usr/bin/python
#

# Copyright (C) 2006, 2007 Google 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.


"""Script for unittesting the utils module"""

import unittest
import os
import time
import tempfile
import os.path
import md5
import socket
import shutil
import re

import ganeti
from ganeti import constants
from ganeti import utils
from ganeti.utils import IsProcessAlive, Lock, Unlock, RunCmd, \
     RemoveFile, CheckDict, MatchNameComponent, FormatUnit, \
     ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \
     ShellQuote, ShellQuoteArgs, TcpPing, ListVisibleFiles, \
     SetEtcHostsEntry, RemoveEtcHostsEntry
from ganeti.errors import LockError, UnitParseError


class GanetiTestCase(unittest.TestCase):
  def assertFileContent(self, file_name, content):
    """Checks the content of a file.

    """
    handle = open(file_name, 'r')
    try:
      self.assertEqual(handle.read(), content)
    finally:
      handle.close()


class TestIsProcessAlive(unittest.TestCase):
  """Testing case for IsProcessAlive"""
  def setUp(self):
    # create a zombie and a (hopefully) non-existing process id
    self.pid_zombie = os.fork()
    if self.pid_zombie == 0:
      os._exit(0)
    elif self.pid_zombie < 0:
      raise SystemError("can't fork")
    self.pid_non_existing = os.fork()
    if self.pid_non_existing == 0:
      os._exit(0)
    elif self.pid_non_existing > 0:
      os.waitpid(self.pid_non_existing, 0)
    else:
      raise SystemError("can't fork")


  def testExists(self):
    mypid = os.getpid()
    self.assert_(IsProcessAlive(mypid),
                 "can't find myself running")

  def testZombie(self):
    self.assert_(not IsProcessAlive(self.pid_zombie),
                 "zombie not detected as zombie")


  def testNotExisting(self):
    self.assert_(not IsProcessAlive(self.pid_non_existing),
                 "noexisting process detected")


class TestLocking(unittest.TestCase):
  """Testing case for the Lock/Unlock functions"""
  def clean_lock(self, name):
    try:
      ganeti.utils.Unlock("unittest")
    except LockError:
      pass


  def testLock(self):
    self.clean_lock("unittest")
    self.assertEqual(None, Lock("unittest"))


  def testUnlock(self):
    self.clean_lock("unittest")
    ganeti.utils.Lock("unittest")
    self.assertEqual(None, Unlock("unittest"))


  def testDoubleLock(self):
    self.clean_lock("unittest")
    ganeti.utils.Lock("unittest")
    self.assertRaises(LockError, Lock, "unittest")


class TestRunCmd(unittest.TestCase):
  """Testing case for the RunCmd function"""

  def setUp(self):
    self.magic = time.ctime() + " ganeti test"

  def testOk(self):
    """Test successful exit code"""
    result = RunCmd("/bin/sh -c 'exit 0'")
    self.assertEqual(result.exit_code, 0)

  def testFail(self):
    """Test fail exit code"""
    result = RunCmd("/bin/sh -c 'exit 1'")
    self.assertEqual(result.exit_code, 1)<commit_msg>Fix the unittest for locking.<commit_after>#!/usr/bin/python
#

# Copyright (C) 2006, 2007 Google 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.


"""Script for unittesting the utils module"""

import unittest
import os
import time
import tempfile
import os.path
import os
import md5
import socket
import shutil
import re

import ganeti
from ganeti import constants
from ganeti import utils
from ganeti.utils import IsProcessAlive, Lock, Unlock, RunCmd, \
     RemoveFile, CheckDict, MatchNameComponent, FormatUnit, \
     ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \
     ShellQuote, ShellQuoteArgs, TcpPing, ListVisibleFiles, \
     SetEtcHostsEntry, RemoveEtcHostsEntry
from ganeti.errors import LockError, UnitParseError


class GanetiTestCase(unittest.TestCase):
  def assertFileContent(self, file_name, content):
    """Checks the content of a file.

    """
    handle = open(file_name, 'r')
    try:
      self.assertEqual(handle.read(), content)
    finally:
      handle.close()


class TestIsProcessAlive(unittest.TestCase):
  """Testing case for IsProcessAlive"""
  def setUp(self):
    # create a zombie and a (hopefully) non-existing process id
    self.pid_zombie = os.fork()
    if self.pid_zombie == 0:
      os._exit(0)
    elif self.pid_zombie < 0:
      raise SystemError("can't fork")
    self.pid_non_existing = os.fork()
    if self.pid_non_existing == 0:
      os._exit(0)
    elif self.pid_non_existing > 0:
      os.waitpid(self.pid_non_existing, 0)
    else:
      raise SystemError("can't fork")


  def testExists(self):
    mypid = os.getpid()
    self.assert_(IsProcessAlive(mypid),
                 "can't find myself running")

  def testZombie(self):
    self.assert_(not IsProcessAlive(self.pid_zombie),
                 "zombie not detected as zombie")


  def testNotExisting(self):
    self.assert_(not IsProcessAlive(self.pid_non_existing),
                 "noexisting process detected")


class TestLocking(unittest.TestCase):
  """Testing case for the Lock/Unlock functions"""

  def setUp(self):
    lock_dir = tempfile.mkdtemp(prefix="ganeti.unittest.",
                                suffix=".locking")
    self.old_lock_dir = constants.LOCK_DIR
    constants.LOCK_DIR = lock_dir

  def tearDown(self):
    try:
      ganeti.utils.Unlock("unittest")
    except LockError:
      pass
    shutil.rmtree(constants.LOCK_DIR, ignore_errors=True)
    constants.LOCK_DIR = self.old_lock_dir

  def clean_lock(self, name):
    try:
      ganeti.utils.Unlock("unittest")
    except LockError:
      pass


  def testLock(self):
    self.clean_lock("unittest")
    self.assertEqual(None, Lock("unittest"))


  def testUnlock(self):
    self.clean_lock("unittest")
    ganeti.utils.Lock("unittest")
    self.assertEqual(None, Unlock("unittest"))

  def testDoubleLock(self):
    self.clean_lock("unittest")
    ganeti.utils.Lock("unittest")
    self.assertRaises(LockError, Lock, "unittest")


class TestRunCmd(unittest.TestCase):
  """Testing case for the RunCmd function"""

  def setUp(self):
    self.magic = time.ctime() + " ganeti test"

  def testOk(self):
    """Test successful exit code"""
    result = RunCmd("/bin/sh -c 'exit 0'")
    self.assertEqual(result.exit_code, 0)

  def testFail(self):
    """Test fail exit code"""
    result = RunCmd("/bin/sh -c 'exit 1'")
    self.assertEqual(result.exit_code, 1)
{'fraction_non_alphanumeric': 0.10242622950819671, 'fraction_numerical': 0.00839344262295082, 'mean_word_length': 3.6022933011466507, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1757874', 'n_tokens_mistral': 2419, 'n_tokens_neox': 2298, 'n_words': 709}