nsscache 0.33-3 / usr / lib / python2.7 / dist-packages / nss_cache / command_test.py

# Copyright 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.

"""Unit tests for nss_cache/command.py."""

__author__ = ('jaq@google.com (Jamie Wilkinson)',
              'vasilios@google.com (Vasilios Hoffman)')

import grp
import os
import pwd
import shutil
import StringIO
import sys
import tempfile
import time
import unittest

import mox

from nss_cache import command
from nss_cache import config
from nss_cache import error
from nss_cache import lock
from nss_cache import nss

from nss_cache.caches import caches
from nss_cache.caches import cache_factory
from nss_cache.maps import automount
from nss_cache.maps import passwd
from nss_cache.sources import source
from nss_cache.sources import source_factory

from nss_cache.update import updater
from nss_cache.update import files_updater
from nss_cache.update import map_updater


class TestCommand(mox.MoxTestBase):
  """Unit tests for the Command class."""

  def testRunCommand(self):
    c = command.Command()
    self.assertRaises(NotImplementedError, c.Run, [], {})

  def testLock(self):
    self.mox.StubOutClassWithMocks(lock, 'PidFile')
    mock_lock = lock.PidFile(filename=None)
    mock_lock.Lock(force=False).AndReturn('LOCK')
    mock_lock.Lock(force=False).AndReturn('MORLOCK')
    mock_lock.Locked().AndReturn(True)
    mock_lock.Unlock()

    self.mox.ReplayAll()

    c = command.Command()

    # First test that we create a lock and lock it.
    self.assertEquals('LOCK', c._Lock())

    # Then we test that we lock the existing one a second time.
    self.assertEquals('MORLOCK', c._Lock())

  def testForceLock(self):
    self.mox.StubOutClassWithMocks(lock, 'PidFile')
    mock_lock = lock.PidFile(filename=None)
    mock_lock.Lock(force=True).AndReturn('LOCK')
    mock_lock.Locked().AndReturn(True)
    mock_lock.Unlock()

    self.mox.ReplayAll()
    c = command.Command()

    self.assertEquals('LOCK', c._Lock(force=True))


  def testUnlock(self):
    self.mox.StubOutClassWithMocks(lock, 'PidFile')
    mock_lock = lock.PidFile(filename=None)
    mock_lock.Lock(force=False).AndReturn(True)
    mock_lock.Locked().AndReturn(True)
    mock_lock.Unlock()
    mock_lock.Locked().AndReturn(False) # destructor

    self.mox.ReplayAll()
    c = command.Command()
    c._Lock()
    c._Unlock()


  def testCommandHelp(self):
    c = command.Command()
    self.assertNotEqual(None, c)
    self.assertEqual(None, c.Help())

  def testDummyCommand(self):

    class Dummy(command.Command):
      """Dummy docstring for dummy command."""

      def Run(self):
        return 0

    c = Dummy()
    self.assertTrue(isinstance(c, command.Command))
    self.failIfEqual(None, c.Help())


class TestUpdateCommand(mox.MoxTestBase):
  """Unit tests for the Update command class."""

  def setUp(self):
    super(TestUpdateCommand, self).setUp()
    self.workdir = tempfile.mkdtemp()

    class DummyConfig(object):
      pass

    self.conf = DummyConfig()
    self.conf.options = {config.MAP_PASSWORD: config.MapOptions(),
                         config.MAP_AUTOMOUNT: config.MapOptions()}
    self.conf.options[config.MAP_PASSWORD].cache = {'name': 'dummy',
                                                    'dir': self.workdir}
    self.conf.options[config.MAP_PASSWORD].source = {'name': 'dummy'}
    self.conf.options[config.MAP_AUTOMOUNT].cache = {'name': 'dummy',
                                                     'dir': self.workdir}
    self.conf.options[config.MAP_AUTOMOUNT].source = {'name': 'dummy'}
    self.conf.timestamp_dir = self.workdir
    self.conf.lockfile = None

  def tearDown(self):
    super(TestUpdateCommand, self).tearDown()
    shutil.rmtree(self.workdir)

  def testConstructor(self):
    c = command.Update()
    self.failIfEqual(None, c)

  def testHelp(self):
    c = command.Update()
    self.failIfEqual(None, c.Help())

  def testRunWithNoParameters(self):
    c = command.Update()

    self.mox.StubOutWithMock(c, 'UpdateMaps')
    c.UpdateMaps(self.conf, incremental=True, force_lock=False, force_write=False).AndReturn(0)
    self.mox.ReplayAll()

    self.assertEquals(0, c.Run(self.conf, []))

  def testRunWithBadParameters(self):
    c = command.Update()
    # Trap stderr so the unit test runs clean,
    # since unit test status is printed on stderr.
    dev_null = StringIO.StringIO()
    stderr = sys.stderr
    sys.stderr = dev_null
    self.assertEquals(2, c.Run(None, ['--invalid']))
    sys.stderr = stderr

  def testRunWithFlags(self):
    c = command.Update()

    self.mox.StubOutWithMock(c, 'UpdateMaps')
    c.UpdateMaps(self.conf, incremental=False, force_lock=True, force_write=True).AndReturn(0)
    self.mox.ReplayAll()

    self.assertEquals(0, c.Run(self.conf,
                               ['-m', config.MAP_PASSWORD, '-f',
                                '--force-write', '--force-lock']))
    self.assertEqual(['passwd'], self.conf.maps)

  def testUpdateSingleMaps(self):
    self.mox.StubOutClassWithMocks(lock, 'PidFile')
    lock_mock = lock.PidFile(filename=None)
    lock_mock.Lock(force=False).AndReturn(True)
    lock_mock.Locked().AndReturn(True)
    lock_mock.Unlock()

    self.conf.maps = [config.MAP_PASSWORD]
    self.conf.cache = 'dummy'

    modify_stamp = 1
    map_entry = passwd.PasswdMapEntry({'name': 'foo', 'uid': 10, 'gid': 10})
    passwd_map = passwd.PasswdMap([map_entry])
    passwd_map.SetModifyTimestamp(modify_stamp)

    source_mock = self.mox.CreateMock(source.Source)
    source_mock.GetMap(config.MAP_PASSWORD, location=None).AndReturn(passwd_map)

    self.mox.StubOutWithMock(source_factory, 'Create')
    source_factory.Create(self.conf.options[config.MAP_PASSWORD].source).AndReturn(source_mock)

    cache_mock = self.mox.CreateMock(caches.Cache)
    cache_mock.WriteMap(map_data=passwd_map).AndReturn(0)

    self.mox.StubOutWithMock(cache_factory, 'Create')
    cache_factory.Create(self.conf.options[config.MAP_PASSWORD].cache, config.MAP_PASSWORD).AndReturn(cache_mock)

    self.mox.ReplayAll()
    c = command.Update()
    self.assertEquals(0, c.UpdateMaps(self.conf,
                                      incremental=True, force_write=False))

  def testUpdateAutomounts(self):
    self.mox.StubOutClassWithMocks(lock, 'PidFile')
    lock_mock = lock.PidFile(filename=None)
    lock_mock.Lock(force=False).AndReturn(True)
    lock_mock.Locked().AndReturn(True)
    lock_mock.Unlock()

    self.conf.maps = [config.MAP_AUTOMOUNT]
    self.conf.cache = 'dummy'

    modify_stamp = 1
    map_entry = automount.AutomountMapEntry()
    map_entry.key = '/home'
    map_entry.location = 'foo'
    automount_map = automount.AutomountMap([map_entry])
    automount_map.SetModifyTimestamp(modify_stamp)

    source_mock = self.mox.CreateMock(source.Source)
    source_mock.GetAutomountMasterMap().AndReturn(automount_map)
    source_mock.GetMap(config.MAP_AUTOMOUNT, location='foo').AndReturn(automount_map)

    self.mox.StubOutWithMock(source_factory, 'Create')
    source_factory.Create(self.conf.options[config.MAP_PASSWORD].source).AndReturn(source_mock)

    cache_mock = self.mox.CreateMock(caches.Cache)
    cache_mock.GetMapLocation().AndReturn('home')
    cache_mock.WriteMap(map_data=automount_map).AndReturn(0)
    cache_mock.WriteMap(map_data=automount_map).AndReturn(0)

    self.mox.StubOutWithMock(cache_factory, 'Create')
    cache_factory.Create(self.conf.options[config.MAP_AUTOMOUNT].cache, config.MAP_AUTOMOUNT, automount_mountpoint='/home').AndReturn(cache_mock)
    cache_factory.Create(self.conf.options[config.MAP_AUTOMOUNT].cache, config.MAP_AUTOMOUNT, automount_mountpoint=None).AndReturn(cache_mock)

    self.mox.ReplayAll()

    c = command.Update()
    self.assertEquals(0, c.UpdateMaps(self.conf,
                                      incremental=True, force_write=False))

  def testUpdateMapsTrapsPermissionDenied(self):
    self.mox.StubOutWithMock(map_updater.MapUpdater, 'UpdateFromSource')
    map_updater.MapUpdater.UpdateFromSource(mox.IgnoreArg(), incremental=True, force_write=False).AndRaise(error.PermissionDenied)

    self.mox.StubOutClassWithMocks(lock, 'PidFile')
    lock_mock = lock.PidFile(filename=None)
    lock_mock.Lock(force=False).AndReturn(True)
    lock_mock.Locked().AndReturn(True)
    lock_mock.Unlock()
    
    self.conf.maps = [config.MAP_PASSWORD]
    self.conf.cache = 'dummy'
    modify_stamp = 1
    map_entry = passwd.PasswdMapEntry({'name': 'foo', 'uid': 10, 'gid': 10})
    passwd_map = passwd.PasswdMap([map_entry])
    passwd_map.SetModifyTimestamp(modify_stamp)

    source_mock = self.mox.CreateMock(source.Source)

    self.mox.StubOutWithMock(source_factory, 'Create')
    source_factory.Create(self.conf.options[config.MAP_PASSWORD].source).AndReturn(source_mock)

    cache_mock = self.mox.CreateMock(caches.Cache)

    self.mox.StubOutWithMock(cache_factory, 'Create')

    self.mox.ReplayAll()

    c = command.Update()
    self.assertEquals(1, c.UpdateMaps(self.conf, incremental=True,
                                      force_write=False))

  def testUpdateMapsCanForceLock(self):
    self.mox.StubOutClassWithMocks(lock, 'PidFile')
    lock_mock = lock.PidFile(filename=None)
    lock_mock.Lock(force=True).AndReturn(False)
    lock_mock.Locked().AndReturn(True)
    lock_mock.Unlock()

    self.mox.ReplayAll()

    c = command.Update()
    self.assertEquals(c.UpdateMaps(self.conf, False, force_lock=True),
                      c.ERR_LOCK)

  def testSleep(self):
    self.mox.StubOutWithMock(time, 'sleep')
    time.sleep(1)

    c = command.Update()
    self.mox.StubOutWithMock(c, 'UpdateMaps')
    c.UpdateMaps(self.conf, incremental=True, force_lock=mox.IgnoreArg(), force_write=mox.IgnoreArg()).AndReturn(0)
    self.mox.ReplayAll()

    c.Run(self.conf, ['-s', '1'])

  def testForceWriteFlag(self):
    c = command.Update()
    (options, _) = c.parser.parse_args([])
    self.assertEqual(False, options.force_write)
    (options, _) = c.parser.parse_args(['--force-write'])
    self.assertEqual(True, options.force_write)

  def testForceLockFlag(self):
    c = command.Update()
    (options, _) = c.parser.parse_args([])
    self.assertEqual(False, options.force_lock)
    (options, _) = c.parser.parse_args(['--force-lock'])
    self.assertEqual(True, options.force_lock)

  def testForceWriteFlagCallsUpdateMapsWithForceWriteTrue(self):
    c = command.Update()

    self.mox.StubOutWithMock(c, 'UpdateMaps')
    c.UpdateMaps(self.conf, incremental=mox.IgnoreArg(), force_lock=mox.IgnoreArg(), force_write=True).AndReturn(0)
    self.mox.ReplayAll()

    self.assertEqual(0, c.Run(self.conf, ['--force-write']))

  def testForceLockFlagCallsUpdateMapsWithForceLockTrue(self):
    c = command.Update()

    self.mox.StubOutWithMock(c, 'UpdateMaps')
    c.UpdateMaps(self.conf, incremental=mox.IgnoreArg(), force_lock=True, force_write=mox.IgnoreArg()).AndReturn(0)
    self.mox.ReplayAll()

    self.assertEqual(0, c.Run(self.conf, ['--force-lock']))

  def testUpdateMapsWithBadMapName(self):
    c = command.Update()
    self.mox.StubOutWithMock(c, '_Lock')
    c._Lock(force=False, path=None).AndReturn(True)
    self.mox.ReplayAll()
    # Create an invalid map name.
    self.assertEqual(1, c.Run(self.conf, ['-m', config.MAP_PASSWORD + 'invalid']))


class TestVerifyCommand(mox.MoxTestBase):

  def setUp(self):
    super(TestVerifyCommand, self).setUp()

    class DummyConfig(object):
      pass

    class DummySource(source.Source):
      name = 'dummy'

      def Verify(self):
        return 0

    # Instead of a DummyCache, we will override cache_factory.Create so
    # we can return a pmock cache object.
    self.original_caches_create = cache_factory.Create
    self.original_sources_create = source_factory.Create

    # Add dummy source to the set if implementations of sources.
    source_factory.RegisterImplementation(DummySource)

    # Create a config with a section for a passwd map.
    self.conf = DummyConfig()
    self.conf.options = {config.MAP_PASSWORD: config.MapOptions()}
    self.conf.options[config.MAP_PASSWORD].cache = {'name': 'dummy'}
    self.conf.options[config.MAP_PASSWORD].source = {'name': 'dummy'}

    self.original_verify_configuration = config.VerifyConfiguration
    self.original_getmap = nss.GetMap
    self.original_getpwall = pwd.getpwall
    self.original_getgrall = grp.getgrall

    # Setup maps used by VerifyMap testing.
    big_map = passwd.PasswdMap()
    map_entry1 = passwd.PasswdMapEntry()
    map_entry1.name = 'foo'
    map_entry1.uid = 10
    map_entry1.gid = 10
    big_map.Add(map_entry1)
    map_entry2 = passwd.PasswdMapEntry()
    map_entry2.name = 'bar'
    map_entry2.uid = 20
    map_entry2.gid = 20
    big_map.Add(map_entry2)

    small_map = passwd.PasswdMap()
    small_map.Add(map_entry1)

    self.big_map = big_map
    self.small_map = small_map

  def tearDown(self):
    super(TestVerifyCommand, self).tearDown()
    config.VerifyConfiguration = self.original_verify_configuration
    cache_factory.Create = self.original_caches_create
    nss.getmap = self.original_getmap
    source_factory.Create = self.original_sources_create
    pwd.getpwall = self.original_getpwall
    grp.getgrall = self.original_getgrall

  def testConstructor(self):
    c = command.Verify()
    self.assertTrue(isinstance(c, command.Verify))

  def testHelp(self):
    c = command.Verify()
    self.failIfEqual(None, c.Help())

  def testRunWithNoParameters(self):

    def FakeVerifyConfiguration(conf):
      """Assert that we call VerifyConfiguration correctly."""
      self.assertEquals(conf, self.conf)
      return (0, 0)

    def FakeVerifyMaps(conf):
      """Assert that VerifyMaps is called with a config object."""
      self.assertEquals(conf, self.conf)
      return 0

    config.VerifyConfiguration = FakeVerifyConfiguration

    c = command.Verify()
    c.VerifyMaps = FakeVerifyMaps

    self.conf.maps = []

    self.assertEquals(1, c.Run(self.conf, []))

  def testRunWithBadParameters(self):
    c = command.Verify()
    # Trap stderr so the unit test runs clean,
    # since unit test status is printed on stderr.
    dev_null = StringIO.StringIO()
    stderr = sys.stderr
    sys.stderr = dev_null
    self.assertEquals(2, c.Run(None, ['--invalid']))
    sys.stderr = stderr

  def testRunWithParameters(self):

    def FakeVerifyConfiguration(conf):
      """Assert that we call VerifyConfiguration correctly."""
      self.assertEquals(conf, self.conf)
      return (0, 0)

    def FakeVerifyMaps(conf):
      """Assert that VerifyMaps is called with a config object."""
      self.assertEquals(conf, self.conf)
      return 0

    config.VerifyConfiguration = FakeVerifyConfiguration

    c = command.Verify()
    c.VerifyMaps = FakeVerifyMaps

    self.assertEquals(0, c.Run(self.conf, ['-m',
                                           config.MAP_PASSWORD]))

  def testVerifyMapsSucceedsOnGoodMaps(self):
    cache_mock = self.mox.CreateMock(caches.Cache)
    cache_mock.GetMap().AndReturn(self.small_map)

    self.mox.StubOutWithMock(cache_factory, 'Create')
    cache_factory.Create(self.conf.options[config.MAP_PASSWORD].cache, config.MAP_PASSWORD).AndReturn(cache_mock)

    self.conf.maps = [config.MAP_PASSWORD]

    self.mox.StubOutWithMock(nss, 'GetMap')
    nss.GetMap(config.MAP_PASSWORD).AndReturn(self.big_map)

    self.mox.ReplayAll()

    c = command.Verify()

    self.assertEquals(0, c.VerifyMaps(self.conf))

  def testVerifyMapsBad(self):
    cache_mock = self.mox.CreateMock(caches.Cache)
    cache_mock.GetMap().AndReturn(self.big_map)

    self.mox.StubOutWithMock(cache_factory, 'Create')
    cache_factory.Create(self.conf.options[config.MAP_PASSWORD].cache, config.MAP_PASSWORD).AndReturn(cache_mock)

    self.conf.maps = [config.MAP_PASSWORD]

    self.mox.StubOutWithMock(nss, 'GetMap')
    nss.GetMap(config.MAP_PASSWORD).AndReturn(self.small_map)

    self.mox.ReplayAll()

    c = command.Verify()

    self.assertEquals(1, c.VerifyMaps(self.conf))

  def testVerifyMapsException(self):
    cache_mock = self.mox.CreateMock(caches.Cache)
    cache_mock.GetMap().AndRaise(error.CacheNotFound)

    self.mox.StubOutWithMock(cache_factory, 'Create')
    cache_factory.Create(self.conf.options[config.MAP_PASSWORD].cache, config.MAP_PASSWORD).AndReturn(cache_mock)

    self.conf.maps = [config.MAP_PASSWORD]

    self.mox.StubOutWithMock(nss, 'GetMap')
    nss.GetMap(config.MAP_PASSWORD).AndReturn(self.small_map)

    self.mox.ReplayAll()

    c = command.Verify()

    self.assertEquals(1, c.VerifyMaps(self.conf))

  def testVerifyMapsSkipsNetgroups(self):
    self.mox.StubOutWithMock(cache_factory, 'Create')

    self.conf.maps = [config.MAP_NETGROUP]

    self.mox.StubOutWithMock(nss, 'GetMap')

    self.mox.ReplayAll()

    c = command.Verify()

    self.assertEquals(0, c.VerifyMaps(self.conf))

  def testVerifySourcesGood(self):
    source_mock = self.mox.CreateMock(source.Source)
    source_mock.Verify().AndReturn(0)

    self.mox.StubOutWithMock(source_factory, 'Create')
    source_factory.Create(mox.IgnoreArg()).AndReturn(source_mock)
    self.conf.maps = [config.MAP_PASSWORD]

    self.mox.ReplayAll()
    self.assertEquals(0, command.Verify().VerifySources(self.conf))

  def testVerifySourcesBad(self):

    self.conf.maps = []
    self.assertEquals(1, command.Verify().VerifySources(self.conf))

    source_mock = self.mox.CreateMock(source.Source)
    source_mock.Verify().AndReturn(1)

    self.mox.StubOutWithMock(source_factory, 'Create')
    source_factory.Create(self.conf.options[config.MAP_PASSWORD].cache).AndReturn(source_mock)

    self.conf.maps = [config.MAP_PASSWORD]

    self.mox.ReplayAll()

    self.assertEquals(1, command.Verify().VerifySources(self.conf))

  def testVerifySourcesTrapsSourceUnavailable(self):
    self.conf.maps = []
    self.assertEquals(1, command.Verify().VerifySources(self.conf))

    def FakeCreate(conf):
      """Stub routine returning a pmock to test VerifySources."""
      self.assertEquals(conf,
                        self.conf.options[config.MAP_PASSWORD].source)
      raise error.SourceUnavailable

    old_source_base_create = source_factory.Create
    source_factory.Create = FakeCreate
    self.conf.maps = [config.MAP_PASSWORD]

    self.assertEquals(1, command.Verify().VerifySources(self.conf))

    source_factory.Create = old_source_base_create


class TestRepairCommand(unittest.TestCase):

  def setUp(self):

    class DummyConfig(object):
      pass

    class DummySource(source.Source):
      name = 'dummy'

      def Verify(self):
        return 0

    # Add dummy source to the set if implementations of sources
    source_factory.RegisterImplementation(DummySource)

    self.conf = DummyConfig()
    self.conf.options = {config.MAP_PASSWORD: config.MapOptions()}
    self.conf.options[config.MAP_PASSWORD].cache = {'name': 'dummy'}
    self.conf.options[config.MAP_PASSWORD].source = {'name': 'dummy'}

    self.original_verify_configuration = config.VerifyConfiguration

  def tearDown(self):
    config.VerifyConfiguration = self.original_verify_configuration

  def testCreate(self):
    c = command.Repair()
    self.assertTrue(isinstance(c, command.Repair))

  def testHelp(self):
    c = command.Repair()
    self.failIfEqual(None, c.Help())

  def testRunWithNoParameters(self):
    c = command.Repair()

    def FakeVerifyConfiguration(conf):
      """Assert that we call VerifyConfiguration correctly."""
      self.assertEquals(conf, self.conf)
      return (0, 1)

    config.VerifyConfiguration = FakeVerifyConfiguration

    self.conf.maps = []

    self.assertEquals(1, c.Run(self.conf, []))

  def testRunWithBadParameters(self):
    c = command.Repair()
    # Trap stderr so the unit test runs clean,
    # since unit test status is printed on stderr.
    dev_null = StringIO.StringIO()
    stderr = sys.stderr
    sys.stderr = dev_null
    self.assertEquals(2, c.Run(None, ['--invalid']))
    sys.stderr = stderr

  def testRunWithParameters(self):

    def FakeVerifyConfiguration(conf):
      """Assert that we call VerifyConfiguration correctly."""
      self.assertEquals(conf, self.conf)
      return (0, 1)

    config.VerifyConfiguration = FakeVerifyConfiguration

    c = command.Repair()

    self.assertEquals(1, c.Run(self.conf, ['-m',
                                           config.MAP_PASSWORD]))


class TestHelpCommand(unittest.TestCase):

  def setUp(self):
    self.stdout = sys.stdout
    sys.stdout = StringIO.StringIO()

  def tearDown(self):
    sys.stdout = self.stdout

  def testHelp(self):
    c = command.Help()
    self.failIfEqual(None, c.Help())

  def testRunWithNoParameters(self):
    c = command.Help()
    self.assertEquals(0, c.Run(None, []))

  def testRunHelpHelp(self):
    c = command.Help()
    self.assertEquals(0, c.Run(None, ['help']))


class TestStatusCommand(mox.MoxTestBase):

  def setUp(self):
    super(TestStatusCommand, self).setUp()

    class DummyConfig(object):
      pass

    class DummySource(source.Source):
      name = 'dummy'

      def Verify(self):
        return 0

    # stub out parts of update.MapUpdater
    class DummyUpdater(map_updater.MapUpdater):
      def GetModifyTimestamp(self):
        return 1

      def GetUpdateTimestamp(self):
        return 2

    # Add dummy source to the set if implementations of sources
    source_factory.RegisterImplementation(DummySource)

    self.conf = DummyConfig()
    self.conf.timestamp_dir = 'TEST_DIR'
    self.conf.options = {config.MAP_PASSWORD: config.MapOptions(),
                         config.MAP_AUTOMOUNT: config.MapOptions()}
    self.conf.options[config.MAP_PASSWORD].cache = {'name': 'dummy'}
    self.conf.options[config.MAP_PASSWORD].source = {'name': 'dummy'}
    self.conf.options[config.MAP_AUTOMOUNT].cache = {'name': 'dummy'}
    self.conf.options[config.MAP_AUTOMOUNT].source = {'name': 'dummy'}

    self.original_verify_configuration = config.VerifyConfiguration
    self.original_create = cache_factory.Create
    self.original_updater = map_updater.MapUpdater

    # stub this out for all tests
    map_updater.MapUpdater = DummyUpdater

  def tearDown(self):
    super(TestStatusCommand, self).tearDown()
    config.VerifyConfiguration = self.original_verify_configuration
    cache_factory.Create = self.original_create
    map_updater.MapUpdater = self.original_updater

  def testHelp(self):
    c = command.Status()
    self.failIfEqual(None, c.Help())

  def testRunWithNoParameters(self):
    c = command.Status()
    self.conf.maps = []
    self.assertEquals(0, c.Run(self.conf, []))

  def testRunWithBadParameters(self):
    c = command.Status()
    # Trap stderr so the unit test runs clean,
    # since unit test status is printed on stderr.
    dev_null = StringIO.StringIO()
    stderr = sys.stderr
    sys.stderr = dev_null
    self.assertEquals(2, c.Run(None, ['--invalid']))
    sys.stderr = stderr

  def testEpochFormatParameter(self):
    c = command.Status()
    (options, args) = c.parser.parse_args([])
    self.assertEqual(False, options.epoch)
    self.assertEqual([], args)

  def testObeysMapsFlag(self):
    stdout_buffer = StringIO.StringIO()

    old_stdout = sys.stdout
    sys.stdout = stdout_buffer

    c = command.Status()
    self.assertEqual(0, c.Run(self.conf, ['-m', 'passwd']))
    sys.stdout = old_stdout

    self.failIfEqual(0, len(stdout_buffer.getvalue()))
    self.failIf(stdout_buffer.getvalue().find('group') >= 0)

  def testGetSingleMapMetadata(self):
    # test both automount and non-automount maps.

    # cache mock is returned by FakeCreate() for automount maps
    cache_mock = self.mox.CreateMock(caches.Cache)
    cache_mock.GetMapLocation().AndReturn('/etc/auto.master')

    self.mox.StubOutWithMock(cache_factory, 'Create')
    cache_factory.Create(self.conf.options[config.MAP_AUTOMOUNT].cache, config.MAP_AUTOMOUNT, automount_mountpoint='automount_mountpoint').AndReturn(cache_mock)

    self.mox.ReplayAll()

    c = command.Status()

    values = c.GetSingleMapMetadata(config.MAP_PASSWORD, self.conf)
    self.failUnless('map' in values[0])
    self.failUnless('key' in values[0])
    self.failUnless('value' in values[0])

    values = c.GetSingleMapMetadata(
        config.MAP_AUTOMOUNT, self.conf,
        automount_mountpoint='automount_mountpoint')

    self.failUnless('map' in values[0])
    self.failUnless('key' in values[0])
    self.failUnless('value' in values[0])
    self.failUnless('automount' in values[0])

  def testGetSingleMapMetadataTimestampEpoch(self):
    c = command.Status()
    values = c.GetSingleMapMetadata(config.MAP_PASSWORD, self.conf,
                                    epoch=True)
    self.failUnless('map' in values[0])
    self.failUnless('key' in values[0])
    self.failUnless('value' in values[0])
    # values below are returned by dummyupdater
    self.assertEqual(1, values[0]['value'])
    self.assertEqual(2, values[1]['value'])

  def testGetSingleMapMetadataTimestampEpochFalse(self):
    # set the timezone so we get a consistent return value
    os.environ['TZ'] = 'US/Pacific'
    time.tzset()

    c = command.Status()
    values = c.GetSingleMapMetadata(config.MAP_PASSWORD, self.conf,
                                    epoch=False)
    self.failUnlessEqual('Wed Dec 31 16:00:02 1969',
                         values[1]['value'])

  def testGetAutomountMapMetadata(self):
    # need to stub out GetSingleMapMetadata (tested above) and then
    # stub out cache_factory.Create to return a cache mock that spits
    # out an iterable map for the function to use.

    # stub out GetSingleMapMetadata
    class DummyStatus(command.Status):
      def GetSingleMapMetadata(self, unused_map_name, unused_conf,
                               automount_mountpoint=None, epoch=False):
        return {'map': 'map_name', 'last-modify-timestamp': 'foo',
                'last-update-timestamp': 'bar'}

    # the master map to loop over
    master_map = automount.AutomountMap()
    master_map.Add(automount.AutomountMapEntry({'key': '/home',
                                           'location': '/etc/auto.home'}))
    master_map.Add(automount.AutomountMapEntry({'key': '/auto',
                                           'location': '/etc/auto.auto'}))

    # mock out a cache to return the master map
    cache_mock = self.mox.CreateMock(caches.Cache)
    cache_mock.GetMap().AndReturn(master_map)

    self.mox.StubOutWithMock(cache_factory, 'Create')
    cache_factory.Create(self.conf.options[config.MAP_AUTOMOUNT].cache, config.MAP_AUTOMOUNT, automount_mountpoint=None).AndReturn(cache_mock)

    self.mox.ReplayAll()

    c = DummyStatus()
    value_list = c.GetAutomountMapMetadata(self.conf)

    self.assertEqual(9, len(value_list))


if __name__ == '__main__':
  unittest.main()