1# coding=utf-8
2# Copyright (c) 2014, 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 the backend package """
23
24from __future__ import (
25    absolute_import, division, print_function, unicode_literals
26)
27import os
28try:
29    import simplejson as json
30except ImportError:
31    import json
32try:
33    import mock
34except ImportError:
35    from unittest import mock
36
37import jsonschema
38import pytest
39
40from framework import backends
41from framework import exceptions
42from framework import grouptools
43from framework import results
44
45from . import shared
46
47# pylint: disable=no-self-use,protected-access
48
49SCHEMA = os.path.join(os.path.dirname(__file__), 'schema',
50                      'piglit-{}.json'.format(backends.json.CURRENT_JSON_VERSION))
51
52
53@pytest.yield_fixture(scope='module', autouse=True)
54def mock_compression():
55    with mock.patch.dict(backends.json.compression.os.environ,
56                         {'PIGLIT_COMPRESSION': 'none'}):
57        yield
58
59
60class TestJSONBackend(object):
61    """Tests for the JSONBackend class."""
62
63    class TestInitialize(object):
64        """Tests for the initialize method."""
65
66        def test_metadata_file_created(self, tmpdir):
67            p = str(tmpdir)
68            test = backends.json.JSONBackend(p)
69            test.initialize(shared.INITIAL_METADATA)
70            assert os.path.exists(os.path.join(p, 'metadata.json'))
71
72    class TestWriteTest(object):
73        """Tests for the write_test method."""
74
75        def test_write(self, tmpdir):
76            """The write method should create a file."""
77            p = str(tmpdir)
78            test = backends.json.JSONBackend(p)
79            test.initialize(shared.INITIAL_METADATA)
80
81            with test.write_test('bar') as t:
82                t(results.TestResult())
83
84            assert tmpdir.join('tests/0.json').check()
85
86        def test_load(self, tmpdir):
87            """Test that the written JSON can be loaded.
88
89            This doesn't attempt to validate the schema of the code (That is
90            handled elsewhere), instead it just attempts a touch test of "can
91            this be read as JSON".
92            """
93            p = str(tmpdir)
94            test = backends.json.JSONBackend(p)
95            test.initialize(shared.INITIAL_METADATA)
96
97            with test.write_test('bar') as t:
98                t(results.TestResult())
99
100            with tmpdir.join('tests/0.json').open('r') as f:
101                json.load(f)
102
103    class TestFinalize(object):
104        """Tests for the finalize method."""
105
106        name = grouptools.join('a', 'test', 'group', 'test1')
107        result = results.TestResult('pass')
108
109        @pytest.fixture(scope='class')
110        def result_dir(self, tmpdir_factory):
111            directory = tmpdir_factory.mktemp('main')
112            test = backends.json.JSONBackend(str(directory))
113            test.initialize(shared.INITIAL_METADATA)
114            with test.write_test(self.name) as t:
115                t(self.result)
116            test.finalize(
117                {'time_elapsed':
118                    results.TimeAttribute(start=0.0, end=1.0).to_json()})
119
120            return directory
121
122        def test_metadata_removed(self, result_dir):
123            assert not result_dir.join('metadata.json').check()
124
125        def test_tests_directory_removed(self, result_dir):
126            assert not result_dir.join('tests').check()
127
128        def test_results_file_created(self, result_dir):
129            # Normally this would also have a compression extension, but this
130            # module has a setup fixture that forces the compression to None.
131            assert result_dir.join('results.json').check()
132
133        def test_results_are_json(self, result_dir):
134            # This only checks that the output is valid JSON, not that the
135            # schema is correct
136            with result_dir.join('results.json').open('r') as f:
137                json.load(f)
138
139        def test_results_are_valid(self, result_dir):
140            """Test that the values produced are valid."""
141            with result_dir.join('results.json').open('r') as f:
142                json_ = json.load(f)
143
144            with open(SCHEMA, 'r') as f:
145                schema = json.load(f)
146
147            jsonschema.validate(json_, schema)
148
149        def test_ignores_invalid(self, tmpdir):
150            test = backends.json.JSONBackend(str(tmpdir))
151            test.initialize(shared.INITIAL_METADATA)
152            with test.write_test(self.name) as t:
153                t(self.result)
154            tmpdir.join('tests/2.json.tmp').write('{}')
155            test.finalize(
156                {'time_elapsed':
157                    results.TimeAttribute(start=0.0, end=1.0).to_json()})
158
159
160class TestUpdateResults(object):
161    """Test for the _update_results function."""
162
163    def test_current_version(self, tmpdir, mocker):
164        """backends.json.update_results(): returns early when the
165        results_version is current.
166        """
167        class Sentinel(Exception):
168            pass
169
170        mocker.patch('framework.backends.json.os.rename',
171                     mocker.Mock(side_effect=Sentinel))
172        p = tmpdir.join('results.json')
173        p.write(json.dumps(shared.JSON))
174
175        with p.open('r') as f:
176            base = backends.json._load(f)
177        backends.json._update_results(base, str(p))
178
179
180class TestResume(object):
181    """tests for the resume function."""
182
183    def test_load_file(self, tmpdir):
184        p = tmpdir.join('results.json')
185        p.write('')
186
187        with pytest.raises(AssertionError):
188            backends.json._resume(str(p))
189
190    def test_load_valid_folder(self, tmpdir):
191        """backends.json._resume: loads valid results."""
192        backend = backends.json.JSONBackend(str(tmpdir))
193        backend.initialize(shared.INITIAL_METADATA)
194        with backend.write_test("group1/test1") as t:
195            t(results.TestResult('fail'))
196        with backend.write_test("group1/test2") as t:
197            t(results.TestResult('pass'))
198        with backend.write_test("group2/test3") as t:
199            t(results.TestResult('fail'))
200        test = backends.json._resume(str(tmpdir))
201
202        assert set(test.tests.keys()) == \
203            {'group1/test1', 'group1/test2', 'group2/test3'}
204
205    def test_load_invalid_ext(self, tmpdir):
206        """backends.json._resume: ignores invalid results extensions.
207
208        This gets triggered by an incomplete atomic write
209        """
210        f = str(tmpdir)
211        backend = backends.json.JSONBackend(f)
212        backend.initialize(shared.INITIAL_METADATA)
213        with backend.write_test("group1/test1") as t:
214            t(results.TestResult('fail'))
215        with backend.write_test("group1/test2") as t:
216            t(results.TestResult('pass'))
217        with backend.write_test("group2/test3") as t:
218            t(results.TestResult('fail'))
219        with open(os.path.join(f, 'tests', '3.json.tmp'), 'w') as w:
220            w.write('foo')
221        test = backends.json._resume(f)
222
223        assert set(test.tests.keys()) == \
224            {'group1/test1', 'group1/test2', 'group2/test3'}
225
226
227    def test_load_incomplete(self, tmpdir):
228        """backends.json._resume: loads incomplete results.
229
230        Because resume, aggregate, and summary all use the function called
231        _resume we can't remove incomplete tests here. It's probably worth
232        doing a refactor to split some code out and allow this to be done in
233        the resume path.
234        """
235        f = str(tmpdir)
236        backend = backends.json.JSONBackend(f)
237        backend.initialize(shared.INITIAL_METADATA)
238        with backend.write_test("group1/test1") as t:
239            t(results.TestResult('fail'))
240        with backend.write_test("group1/test2") as t:
241            t(results.TestResult('pass'))
242        with backend.write_test("group2/test3") as t:
243            t(results.TestResult('crash'))
244        with backend.write_test("group2/test4") as t:
245            t(results.TestResult('incomplete'))
246        test = backends.json._resume(f)
247
248        assert set(test.tests.keys()) == \
249            {'group1/test1', 'group1/test2', 'group2/test3', 'group2/test4'}
250
251
252class TestLoadResults(object):
253    """Tests for the load_results function."""
254
255    def test_folder_with_results_json(self, tmpdir):
256        """backends.json.load_results: takes a folder with a file named
257        results.json.
258        """
259        p = tmpdir.join('results.json')
260        with p.open('w') as f:
261            f.write(json.dumps(shared.JSON))
262        backends.json.load_results(str(tmpdir), 'none')
263
264    def test_load_file(self, tmpdir):
265        """backends.json.load_results: Loads a file passed by name"""
266        p = tmpdir.join('my file')
267        with p.open('w') as f:
268            f.write(json.dumps(shared.JSON))
269        backends.json.load_results(str(p), 'none')
270
271    def test_inst(self, tmpdir):
272        p = tmpdir.join('my file')
273        with p.open('w') as f:
274            f.write(json.dumps(shared.JSON))
275        assert isinstance(backends.json.load_results(str(p), 'none'),
276                          results.TestrunResult)
277
278
279class TestLoad(object):
280    """Tests for the _load function."""
281
282    def test_load_bad_json(self, tmpdir):
283        """backends.json._load: Raises fatal error if json is corrupt"""
284        p = tmpdir.join('foo')
285        p.write('{"bad json": }')
286        with p.open('r') as f:
287            with pytest.raises(exceptions.PiglitFatalError):
288                backends.json._load(f)
289