sdks/python/apache_beam/testing/test_utils_test.py

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

"""Unittest for testing utilities,"""

# pytype: skip-file

import importlib
import logging
import os
import sys
import tempfile
import unittest

import mock

from apache_beam.io.filesystem import BeamIOError
from apache_beam.io.filesystems import FileSystems
from apache_beam.testing import test_utils as utils


class TestUtilsTest(unittest.TestCase):
  def setUp(self):
    utils.patch_retry(self, utils)
    self.tmpdir = tempfile.mkdtemp()

  def test_delete_files_succeeds(self):
    path = os.path.join(self.tmpdir, 'f1')

    with open(path, 'a') as f:
      f.write('test')

    assert FileSystems.exists(path)
    utils.delete_files([path])
    assert not FileSystems.exists(path)

  def test_delete_files_fails_with_io_error(self):
    path = os.path.join(self.tmpdir, 'f2')

    with self.assertRaises(BeamIOError) as error:
      utils.delete_files([path])
    self.assertTrue(
        error.exception.args[0].startswith('Delete operation failed'))
    self.assertEqual(list(error.exception.exception_details.keys()), [path])

  def test_delete_files_fails_with_invalid_arg(self):
    with self.assertRaises(RuntimeError):
      utils.delete_files([])

  def test_temp_dir_removes_files(self):
    with utils.TempDir() as tempdir:
      dir_path = tempdir.get_path()
      file_path = tempdir.create_temp_file()
      self.assertTrue(os.path.exists(dir_path))
      self.assertTrue(os.path.exists(file_path))

    self.assertFalse(os.path.exists(dir_path))
    self.assertFalse(os.path.exists(file_path))

  def test_temp_file_field_correct(self):
    with utils.TempDir() as tempdir:
      filename = tempdir.create_temp_file(
          suffix='.txt', lines=[b'line1\n', b'line2\n', b'line3\n'])
      self.assertTrue(filename.endswith('.txt'))

      with open(filename, 'rb') as f:
        self.assertEqual(f.readline(), b'line1\n')
        self.assertEqual(f.readline(), b'line2\n')
        self.assertEqual(f.readline(), b'line3\n')

  def test_cleanup_subscriptions(self):
    sub_client = mock.Mock()
    sub = mock.Mock()
    sub.name = 'test_sub'
    utils.cleanup_subscriptions(sub_client, [sub])
    sub_client.delete_subscription.assert_called_with(subscription=sub.name)

  def test_cleanup_topics(self):
    pub_client = mock.Mock()
    topic = mock.Mock()
    topic.name = 'test_topic'
    utils.cleanup_topics(pub_client, [topic])
    pub_client.delete_topic.assert_called_with(topic=topic.name)


class BlockImportsTest(unittest.TestCase):
  def test_block_imports_blocks_and_records_imports(self):
    with utils.block_imports('colorsys') as attempted_imports:
      with self.assertRaises(ImportError):
        importlib.import_module('colorsys')
    self.assertEqual(attempted_imports, ['colorsys'])
    # The module can be imported again after exiting the context.
    importlib.import_module('colorsys')

  def test_block_imports_hides_and_restores_imported_modules(self):
    wsgiref_util = importlib.import_module('wsgiref.util')
    wsgiref = sys.modules['wsgiref']
    with utils.block_imports('wsgiref') as attempted_imports:
      self.assertNotIn('wsgiref', sys.modules)
      self.assertNotIn('wsgiref.util', sys.modules)
      with self.assertRaises(ImportError):
        importlib.import_module('wsgiref.util')
    # Importing a submodule first imports its parent package.
    self.assertEqual(attempted_imports, ['wsgiref'])
    self.assertIs(sys.modules['wsgiref'], wsgiref)
    self.assertIs(sys.modules['wsgiref.util'], wsgiref_util)

  def test_block_imports_only_blocks_given_packages(self):
    with utils.block_imports('wsgi', 'colorsys.foo') as attempted_imports:
      importlib.import_module('wsgiref')
      importlib.import_module('colorsys')
    self.assertEqual(attempted_imports, [])


class RequireLinuxDockerTest(unittest.TestCase):
  def _fake_docker(self, os_type=None, error=None):
    fake_docker = mock.MagicMock()
    client = fake_docker.from_env.return_value
    if error:
      client.info.side_effect = error
    else:
      client.info.return_value = {'OSType': os_type}
    return fake_docker, client

  def test_linux_docker_does_not_skip(self):
    fake_docker, client = self._fake_docker(os_type='linux')
    with mock.patch.dict(sys.modules, {'docker': fake_docker}):
      utils.require_linux_docker()
    client.close.assert_called_once()

  def test_windows_docker_skips(self):
    fake_docker, client = self._fake_docker(os_type='windows')
    with mock.patch.dict(sys.modules, {'docker': fake_docker}):
      with self.assertRaisesRegex(unittest.SkipTest, 'windows containers'):
        utils.require_linux_docker()
    client.close.assert_called_once()

  def test_unreachable_docker_skips(self):
    fake_docker, client = self._fake_docker(
        error=ConnectionError('no such file'))
    with mock.patch.dict(sys.modules, {'docker': fake_docker}):
      with self.assertRaisesRegex(unittest.SkipTest, 'Docker is not available'):
        utils.require_linux_docker()
    client.close.assert_called_once()

  def test_missing_docker_sdk_skips(self):
    with utils.block_imports('docker'):
      with self.assertRaisesRegex(unittest.SkipTest, 'Docker is not available'):
        utils.require_linux_docker()


class LCGeneratorTest(unittest.TestCase):
  Generator = utils.LCGenerator

  def test_generator_seed_results(self):
    generator = self.Generator()
    generator.seed(0)
    self.assertEqual(generator.next_int(), 4232237)

    generator.seed(1)
    self.assertEqual(generator.next_int(), -1151252339)
    self.assertEqual(generator.next_uint(), 3745583449)
    self.assertAlmostEqual(generator.random_sample(), 0.375548, delta=1e-6)
    self.assertEqual(generator.randbytes(10), b'\xa6\x8fW\xcb\xb1\xa88]dP')

  def test_generator_seed_jdk_results(self):
    generator = self.Generator()
    generator.seed_jdk(0)
    self.assertEqual(generator.next_int(), -1155484576)

    generator.seed_jdk(1)
    # the first next_int after seed_jdk(1) is close to seed_jdk(0)
    self.assertEqual(generator.next_int(), -1155869325)
    self.assertEqual(generator.next_uint(), 431529176)
    self.assertAlmostEqual(generator.random_sample(), 0.410081, delta=1e-6)
    self.assertEqual(generator.randbytes(10), b'\x92\xf9Mh\xfc\xcc,5\xf0\xb8')


try:
  # pylint: disable=wrong-import-position
  from .fast_test_utils import LCGenerator as FastLCGenerator

  class FastLCGeneratorTest(LCGeneratorTest):
    Generator = FastLCGenerator

except ImportError:
  pass

if __name__ == '__main__':
  logging.getLogger().setLevel(logging.INFO)
  unittest.main()