unittests/framework/backends/test_compression.py

   1 # coding=utf-8
   2 # Copyright (c) 2015-2016, 2019 Intel Corporation
   3
   4 # Permission is hereby granted, free of charge, to any person obtaining a copy
   5 # of this software and associated documentation files (the "Software"), to deal
   6 # in the Software without restriction, including without limitation the rights
   7 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8 # copies of the Software, and to permit persons to whom the Software is
   9 # furnished to do so, subject to the following conditions:
  10
  11 # The above copyright notice and this permission notice shall be included in
  12 # all copies or substantial portions of the Software.
  13
  14 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  20 # SOFTWARE.
  21
  22 """Tests for compression in file backends."""
  23
  24 import itertools
  25 import os
  26 import subprocess
  27 try:
  28     import mock
  29 except ImportError:
  30     from unittest import mock
  31
  32 import pytest
  33
  34 from framework import core
  35 from framework.backends import abstract
  36 from framework.backends import compression
  37
  38 from .. import skip
  39
  40 # pylint: disable=no-self-use,redefined-outer-name
  41
  42 # Helpers
  43
  44
  45 def _has_lzma():
  46     """Try to import backports.lzma, and return True if it exists, False if it
  47     doesn't.
  48     """
  49     try:
  50         import backports.lzma  # pylint: disable=unused-variable
  51     except ImportError:
  52         return False
  53     return True
  54
  55
  56 def _has_xz_bin():
  57     """Check for an xz binary."""
  58     try:
  59         subprocess.check_call(['xz', '--help'])
  60     except subprocess.CalledProcessError:
  61         return False
  62     except OSError as e:  # pylint: disable=undefined-variable
  63         if e.errno != 2:
  64             raise
  65         return False
  66     return True
  67
  68
  69 @pytest.fixture
  70 def env():
  71     with mock.patch.dict('os.environ'):
  72         yield os.environ
  73
  74 @pytest.fixture
  75 def config():
  76     with mock.patch('framework.core.PIGLIT_CONFIG',
  77                     new_callable=core.PiglitConfig) as conf:
  78         conf.add_section('core')
  79         yield conf
  80
  81 @pytest.fixture
  82 def compressor():
  83     """Workaround for yield_fixture not working with classes."""
  84     class Compressor(object):
  85         """Fixture to control the compressor/decompressor objects."""
  86
  87         def __init__(self):
  88             self._mock_c = mock.patch.dict(
  89                 'framework.backends.compression.COMPRESSORS', clear=False)
  90             self._mock_d = mock.patch.dict(
  91                 'framework.backends.compression.DECOMPRESSORS', clear=False)
  92
  93         def __enter__(self):
  94             self._mock_c.start()
  95             self._mock_d.start()
  96             return self
  97
  98         def __exit__(self, type_, value, traceback):
  99             self._mock_c.stop()
 100             self._mock_d.stop()
 101
 102         def add(self, name, func):
 103             assert name not in compression.COMPRESSORS
 104             assert name not in compression.DECOMPRESSORS
 105             compression.COMPRESSORS[name] = func
 106             compression.DECOMPRESSORS[name] = func
 107
 108         def rm(self, name):  # pylint: disable=invalid-name
 109             assert name in compression.COMPRESSORS
 110             assert name in compression.DECOMPRESSORS
 111             del compression.COMPRESSORS[name]
 112             del compression.DECOMPRESSORS[name]
 113
 114     with Compressor() as c:
 115         yield c
 116
 117
 118 # Tests
 119
 120
 121 @pytest.mark.parametrize("mode", ['none', 'bz2', 'gz', 'xz'])
 122 def test_compress(mode, tmpdir):
 123     """Test that each compressor that we want works.
 124
 125     These only check using modules, that means on python2 this test will skip
 126     unless backports.lzma is available, and it will not test the the xz binary
 127     path.
 128     """
 129     func = compression.COMPRESSORS[mode]
 130     testfile = tmpdir.join('test')
 131     with func(str(testfile)) as f:
 132         f.write('foo')
 133
 134
 135 @pytest.mark.parametrize("mode", ['none', 'bz2', 'gz', 'xz'])
 136 def test_decompress(mode, tmpdir):
 137     """Test that each supported decompressor works.
 138
 139     See the docstring in test_compress.
 140     """
 141     comp = compression.COMPRESSORS[mode]
 142     dec = compression.DECOMPRESSORS[mode]
 143     testfile = tmpdir.join('test')
 144
 145     with comp(str(testfile)) as f:
 146         f.write('foo')
 147
 148     with dec(str(testfile)) as f:
 149         actual = f.read()
 150
 151     assert actual == 'foo'
 152
 153
 154 @skip.posix
 155 class TestXZBin(object):
 156     """Tests for the xz bin path on python2.x."""
 157
 158     def test_compress_xz_bin(self, tmpdir):
 159         """Test python2 xz compression using the xz binary."""
 160         func = compression.COMPRESSORS['xz']
 161         testfile = tmpdir.join('test')
 162         with func(str(testfile)) as f:
 163             f.write('foo')
 164
 165     def test_decompress_xz_bin(self, tmpdir):
 166         """Test python2 xz decompression using the xz binary."""
 167         comp = compression.COMPRESSORS['xz']
 168         dec = compression.DECOMPRESSORS['xz']
 169         testfile = tmpdir.join('test')
 170
 171         with comp(str(testfile)) as f:
 172             f.write('foo')
 173
 174         with dec(str(testfile)) as f:
 175             actual = f.read()
 176
 177         assert actual == 'foo'
 178
 179
 180 class TestGetMode(object):
 181     """Tests for the compression.get_mode function."""
 182
 183     def test_default(self, env, config):  # pylint: disable=unused-argument
 184         """When neither the config file nor the environment sets a value for
 185         compression the default value should be used.
 186         """
 187         env.clear()
 188
 189         assert compression.get_mode() == compression.DEFAULT
 190
 191     def test_env(self, env, config, compressor):
 192         """Test that when env doesn't have a PIGLIT_COMPRESSION environment
 193         variable set, but the configuration has a compression method set that
 194         it is used.
 195         """
 196         compressor.add('foo', None)
 197         compressor.add('bar', None)
 198         config.set('core', 'compression', 'foo')
 199         env['PIGLIT_COMPRESSION'] = 'bar'
 200
 201         assert compression.get_mode() == 'bar'
 202
 203     def test_piglit_conf(self, env, config, compressor):
 204         """Test that when env doesn't have a PIGLIT_COMPRESSION environment
 205         variable set, but the configuration has a compression method set that
 206         it is used.
 207         """
 208         compressor.add('foobar', None)
 209         config.set('core', 'compression', 'foobar')
 210         env.clear()
 211
 212         assert compression.get_mode() == 'foobar'
 213
 214
 215 @pytest.mark.parametrize("extension", ['bz2', 'gz', 'xz'])
 216 def test_duplicate_extensions(extension, tmpdir, config):
 217     """Tests that exercises a bug that caused the compressed extension to be
 218     duplicated in some cases.
 219     """
 220     tmpdir.chdir()
 221     config.set('core', 'compression', extension)
 222     expected = 'results.txt.' + extension
 223
 224     with abstract.write_compressed(expected) as f:
 225         f.write('foo')
 226
 227     assert expected in os.listdir('.')
 228
 229
 230 @pytest.mark.parametrize("orig,new", itertools.permutations(
 231     ['bz2', 'gz', 'xz'], 2))
 232 def test_changed_extension(orig, new, tmpdir, config):
 233     """Tests that exercises a bug that caused two extensions to be present if
 234     the compression method changed.
 235     """
 236     tmpdir.chdir()
 237     config.set('core', 'compression', str(new))
 238
 239     with abstract.write_compressed('results.txt.' + orig) as f:
 240         f.write('foo')
 241
 242     assert 'results.txt.' + new in os.listdir('.')
 243     assert 'results.txt.' + orig not in os.listdir('.')
 244     assert 'results.txt.{}.{}'.format(orig, new) not in os.listdir('.')