text: # Copyright (c) 2010-2012 OpenStack Foundation
  #
  # Licensed 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 six.moves.cPickle as pickle
  import mock
  import os
  import unittest
  from contextlib import closing
  from gzip import GzipFile
  from shutil import rmtree
  from tempfile import mkdtemp
  from test.unit import FakeLogger
  
  from eventlet import spawn, Timeout, listen
  
  from swift.common import utils
  from swift.container import updater as container_updater
  from swift.container.backend import ContainerBroker, DATADIR
  from swift.common.ring import RingData
  from swift.common.utils import normalize_timestamp
  
  
  class TestContainerUpdater(unittest.TestCase):
  
      def setUp(self):
          utils.HASH_PATH_SUFFIX = 'endcap'
          utils.HASH_PATH_PREFIX = 'startcap'
          self.testdir = os.path.join(mkdtemp(), 'tmp_test_container_updater')
          rmtree(self.testdir, ignore_errors=1)
          os.mkdir(self.testdir)
          ring_file = os.path.join(self.testdir, 'account.ring.gz')
          with closing(GzipFile(ring_file, 'wb')) as f:
              pickle.dump(
                  RingData([[0, 1, 0, 1], [1, 0, 1, 0]],
                           [{'id': 0, 'ip': '127.0.0.1', 'port': 12345,
                             'device': 'sda1', 'zone': 0},
                            {'id': 1, 'ip': '127.0.0.1', 'port': 12345,
                             'device': 'sda1', 'zone': 2}], 30),
                  f)
          self.devices_dir = os.path.join(self.testdir, 'devices')
          os.mkdir(self.devices_dir)
          self.sda1 = os.path.join(self.devices_dir, 'sda1')
          os.mkdir(self.sda1)
  
      def tearDown(self):
          rmtree(os.path.dirname(self.testdir), ignore_errors=1)
  
      def _get_container_updater(self, conf_updates=None):
          conf = {
              'devices': self.devices_dir,
              'mount_check': 'false',
              'swift_dir': self.testdir,
              'interval': '1',
              'concurrency': '1',
              'node_timeout': '15',
              'account_suppression_time': 0
          }
          if conf_updates:
              conf.update(conf_updates)
          return container_updater.ContainerUpdater(conf)
  
      def test_creation(self):
          cu = self._get_container_updater({'concurrency': '2',
                                            'node_timeout': '5.5'})
          self.assertTrue(hasattr(cu, 'logger'))
          self.assertTrue(cu.logger is not None)
          self.assertEqual(cu.devices, self.devices_dir)
          self.assertEqual(cu.interval, 1)
          self.assertEqual(cu.concurrency, 2)
          self.assertEqual(cu.node_timeout, 5.5)
          self.assertEqual(cu.account_suppression_time, 0)
          self.assertTrue(cu.get_account_ring() is not None)
  
      @mock.patch.object(container_updater, 'ismount')
      @mock.patch.object(container_updater.ContainerUpdater, 'container_sweep')
      def test_run_once_with_device_unmounted(self, mock_sweep, mock_ismount):
  
          mock_ismount.return_value = False
          cu = self._get_container_updater()
          containers_dir = os.path.join(self.sda1, DATADIR)
          os.mkdir(containers_dir)
          partition_dir = os.path.join(containers_dir, "a")
          os.mkdir(partition_dir)
  
          cu.run_once()
          self.assertTrue(os.path.exists(containers_dir))  # sanity check
  
          # only called if a partition dir exists
          self.assertTrue(mock_sweep.called)
  
          mock_sweep.reset_mock()
          cu = self._get_container_updater({'mount_check': 'true'})
          cu.logger = FakeLogger()
          cu.run_once()
          log_lines = cu.logger.get_lines_for_level('warning')
          self.assertGreater(len(log_lines), 0)
          msg = 'sda1 is not mounted'
          self.assertEqual(log_lines[0], msg)
          # Ensure that the container_sweep did not run
          self.assertFalse(mock_sweep.called)
  
      def test_run_once(self):
          cu = self._get_container_updater()
          cu.run_once()
          containers_dir = os.path.join(self.sda1, DATADIR)
          os.mkdir(containers_dir)
          cu.run_once()
          self.assertTrue(os.path.exists(containers_dir))
          subdir = os.path.join(containers_dir, 'subdir')
          os.mkdir(subdir)
          cb = ContainerBroker(os.path.join(subdir, 'hash.db'), account='a',
                               container='c')
          cb.initialize(normalize_timestamp(1), 0)
          cu.run_once()
          info = cb.get_info()
          self.assertEqual(info['object_count'], 0)
          self.assertEqual(info['bytes_used'], 0)
          self.assertEqual(info['reported_object_count'], 0)
          self.assertEqual(info['reported_bytes_used'], 0)
  
          cb.put_object('o', normalize_timestamp(2), 3, 'text/plain',
                        '68b329da9893e34099c7d8ad5cb9c940')
          cu.run_once()
          info = cb.get_info()
          self.assertEqual(info['object_count'], 1)
          self.assertEqual(info['bytes_used'], 3)
          self.assertEqual(info['reported_object_count'], 0)
          self.assertEqual(info['reported_bytes_used'], 0)
  
          def accept(sock, addr, return_code):
              try:
                  with Timeout(3):
                      inc = sock.makefile('rb')
                      out = sock.makefile('wb')
                      out.write('HTTP/1.1 %d OK\r\nContent-Length: 0\r\n\r\n' %
                                return_code)
                      out.flush()
                      self.assertEqual(inc.readline(),
                                       'PUT /sda1/0/a/c HTTP/1.1\r\n')
                      headers = {}
                      line = inc.readline()
                      while line and line != '\r\n':
                          headers[line.split(':')[0].lower()] = \
                              line.split(':')[1].strip()
                          line = inc.readline()
                      self.assertTrue('x-put-timestamp' in headers)
                      self.assertTrue('x-delete-timestamp' in headers)
                      self.assertTrue('x-object-count' in headers)
                      self.assertTrue('x-bytes-used' in headers)
              except BaseException as err:
                  import traceback
                  traceback.print_exc()
                  return err
              return None
          bindsock = listen(('127.0.0.1', 0))
  
          def spawn_accepts():
              events = []
              for _junk in range(2):
                  sock, addr = bindsock.accept()
                  events.append(spawn(accept, sock, addr, 201))
              return events
  
          spawned = spawn(spawn_accepts)
          for dev in cu.get_account_ring().devs:
              if dev is not None:
                  dev['port'] = bindsock.getsockname()[1]
          cu.run_once()
          for event in spawned.wait():
              err = event.wait()
              if err:
                  raise err
          info = cb.get_info()
          self.assertEqual(info['object_count'], 1)
          self.assertEqual(info['bytes_used'], 3)
          self.assertEqual(info['reported_object_count'], 1)
          self.assertEqual(info['reported_bytes_used'], 3)
  
      @mock.patch('os.listdir')
      def test_listdir_with_exception(self, mock_listdir):
          e = OSError('permission_denied')
          mock_listdir.side_effect = e
          cu = self._get_container_updater()
          cu.logger = FakeLogger()
          paths = cu.get_paths()
          self.assertEqual(paths, [])
          log_lines = cu.logger.get_lines_for_level('error')
          msg = ('ERROR:  Failed to get paths to drive partitions: '
                 'permission_denied')
          self.assertEqual(log_lines[0], msg)
  
      @mock.patch('os.listdir', return_value=['foo', 'bar'])
      def test_listdir_without_exception(self, mock_listdir):
          cu = self._get_container_updater()
          cu.logger = FakeLogger()
          path = cu._listdir('foo/bar/')
          self.assertEqual(path, ['foo', 'bar'])
          log_lines = cu.logger.get_lines_for_level('error')
          self.assertEqual(len(log_lines), 0)
  
      def test_unicode(self):
          cu = self._get_container_updater()
          containers_dir = os.path.join(self.sda1, DATADIR)
          os.mkdir(containers_dir)
          subdir = os.path.join(containers_dir, 'subdir')
          os.mkdir(subdir)
          cb = ContainerBroker(os.path.join(subdir, 'hash.db'), account='a',
                               container='\xce\xa9')
          cb.initialize(normalize_timestamp(1), 0)
          cb.put_object('\xce\xa9', normalize_timestamp(2), 3, 'text/plain',
                        '68b329da9893e34099c7d8ad5cb9c940')
  
          def accept(sock, addr):
              try:
                  with Timeout(3):
                      inc = sock.makefile('rb')
                      out = sock.makefile('wb')
                      out.write('HTTP/1.1 201 OK\r\nContent-Length: 0\r\n\r\n')
                      out.flush()
                      inc.read()
              except BaseException as err:
                  import traceback
                  traceback.print_exc()
                  return err
              return None
  
          bindsock = listen(('127.0.0.1', 0))
  
          def spawn_accepts():
              events = []
              for _junk in range(2):
                  with Timeout(3):
                      sock, addr = bindsock.accept()
                      events.append(spawn(accept, sock, addr))
              return events
  
          spawned = spawn(spawn_accepts)
          for dev in cu.get_account_ring().devs:
              if dev is not None:
                  dev['port'] = bindsock.getsockname()[1]
          cu.run_once()
          for event in spawned.wait():
              err = event.wait()
              if err:
                  raise err
          info = cb.get_info()
          self.assertEqual(info['object_count'], 1)
          self.assertEqual(info['bytes_used'], 3)
          self.assertEqual(info['reported_object_count'], 1)
          self.assertEqual(info['reported_bytes_used'], 3)
  
  if __name__ == '__main__':
      unittest.main()
  