scripts/tests/twister/test_scl.py

#!/usr/bin/env python3
# Copyright (c) 2023 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
"""
Tests for scl.py functions
"""

import logging
import sys
from contextlib import nullcontext
from importlib import reload
from unittest import mock

import pytest
import scl
from pykwalify.errors import SchemaError
from yaml.scanner import ScannerError

TESTDATA_1 = [
    (False,),
    (True,),
]

@pytest.mark.parametrize(
    'fail_c',
    TESTDATA_1,
    ids=['C YAML', 'non-C YAML']
)
def test_yaml_imports(fail_c):
    class ImportRaiser:
        def find_spec(self, fullname, path, target=None):
            if fullname == 'yaml.CLoader' and fail_c:
                raise ImportError()
            if fullname == 'yaml.CSafeLoader' and fail_c:
                raise ImportError()
            if fullname == 'yaml.CDumper' and fail_c:
                raise ImportError()

    modules_mock = sys.modules.copy()

    if hasattr(modules_mock['yaml'], 'CLoader'):
        del modules_mock['yaml'].CLoader
        del modules_mock['yaml'].CSafeLoader
        del modules_mock['yaml'].CDumper

    cloader_mock = mock.Mock()
    loader_mock = mock.Mock()
    csafeloader_mock = mock.Mock()
    safeloader_mock = mock.Mock()
    cdumper_mock = mock.Mock()
    dumper_mock = mock.Mock()

    if not fail_c:
        modules_mock['yaml'].CLoader = cloader_mock
        modules_mock['yaml'].CSafeLoader = csafeloader_mock
        modules_mock['yaml'].CDumper = cdumper_mock

    modules_mock['yaml'].Loader = loader_mock
    modules_mock['yaml'].SafeLoader = safeloader_mock
    modules_mock['yaml'].Dumper = dumper_mock

    meta_path_mock = sys.meta_path[:]
    meta_path_mock.insert(0, ImportRaiser())

    with mock.patch.dict('sys.modules', modules_mock, clear=True), \
         mock.patch('sys.meta_path', meta_path_mock):
        reload(scl)

    assert sys.modules['scl'].Loader == loader_mock if fail_c else \
                                        cloader_mock

    assert sys.modules['scl'].SafeLoader == safeloader_mock if fail_c else \
                                        csafeloader_mock

    assert sys.modules['scl'].Dumper == dumper_mock if fail_c else \
                                        cdumper_mock

    import yaml
    reload(yaml)


TESTDATA_2 = [
    (False, logging.CRITICAL, []),
    (True, None, ['can\'t import pykwalify; won\'t validate YAML']),
]

@pytest.mark.parametrize(
    'fail_pykwalify, log_level, expected_logs',
    TESTDATA_2,
    ids=['pykwalify OK', 'no pykwalify']
)
def test_pykwalify_import(caplog, fail_pykwalify, log_level, expected_logs):
    class ImportRaiser:
        def find_spec(self, fullname, path, target=None):
            if fullname == 'pykwalify.core' and fail_pykwalify:
                raise ImportError()

    modules_mock = sys.modules.copy()
    modules_mock['pykwalify'] = None if fail_pykwalify else \
                                modules_mock['pykwalify']

    meta_path_mock = sys.meta_path[:]
    meta_path_mock.insert(0, ImportRaiser())

    with mock.patch.dict('sys.modules', modules_mock, clear=True), \
         mock.patch('sys.meta_path', meta_path_mock):
        reload(scl)

    if log_level:
        assert logging.getLogger('pykwalify.core').level == log_level

    assert all([log in caplog.text for log in expected_logs])

    if fail_pykwalify:
        assert scl._yaml_validate(None, None) is None
        assert scl._yaml_validate(mock.Mock(), mock.Mock()) is None

    reload(scl)


TESTDATA_3 = [
    (False),
    (True),
]

@pytest.mark.parametrize(
    'fail_parsing',
    TESTDATA_3,
    ids=['ok', 'parsing error']
)
def test_yaml_load(caplog, fail_parsing):
    result_mock = mock.Mock()

    def mock_load(*args, **kwargs):
        if fail_parsing:
            context_mark = mock.Mock()
            problem_mark = mock.Mock()
            type(context_mark).args = mock.PropertyMock(return_value=[])
            type(context_mark).name = 'dummy context mark'
            type(context_mark).line = 0
            type(context_mark).column = 0
            type(problem_mark).args = mock.PropertyMock(return_value=[])
            type(problem_mark).name = 'dummy problem mark'
            type(problem_mark).line = 0
            type(problem_mark).column = 0
            raise ScannerError(context='dummy context',
                               context_mark=context_mark, problem='dummy problem',
                               problem_mark=problem_mark, note='Dummy note')
        return result_mock

    filename = 'dummy/file.yaml'

    with mock.patch('yaml.load', side_effect=mock_load), \
         mock.patch('builtins.open', mock.mock_open()) as mock_file:
        with pytest.raises(ScannerError) if fail_parsing else nullcontext():
            result = scl.yaml_load(filename)

    mock_file.assert_called_with('dummy/file.yaml', 'r', encoding='utf-8')

    if not fail_parsing:
        assert result == result_mock
    else:
        assert 'dummy problem mark:0:0: error: dummy problem' \
               ' (note Dummy note context @dummy context mark:0:0' \
               ' dummy context)' in caplog.text



TESTDATA_4 = [
    (True, False, None),
    (False, False, SchemaError),
    (False, True, ScannerError),
]

@pytest.mark.parametrize(
    'validate, fail_load, expected_error',
    TESTDATA_4,
    ids=['successful validation', 'failed validation', 'failed load']
)
def test_yaml_load_verify(validate, fail_load, expected_error):
    filename = 'dummy/file.yaml'
    schema_mock = mock.Mock()
    data_mock = mock.Mock()

    def mock_load(file_name, *args, **kwargs):
        assert file_name == filename
        if fail_load:
            raise ScannerError
        return data_mock

    def mock_validate(data, schema, *args, **kwargs):
        assert data == data_mock
        assert schema == schema_mock
        if validate:
            return True
        raise SchemaError(u'Schema validation failed.')

    with mock.patch('scl.yaml_load', side_effect=mock_load), \
         mock.patch('scl._yaml_validate', side_effect=mock_validate), \
         pytest.raises(expected_error) if expected_error else nullcontext():
        res = scl.yaml_load_verify(filename, schema_mock)

    if validate:
        assert res == data_mock


TESTDATA_5 = [
    (True, True, None),
    (True, False, SchemaError),
    (False, None, None),
]

@pytest.mark.parametrize(
    'schema_exists, validate, expected_error',
    TESTDATA_5,
    ids=['successful validation', 'failed validation', 'no schema']
)
def test_yaml_validate(schema_exists, validate, expected_error):
    data_mock = mock.Mock()
    schema_mock = mock.Mock() if schema_exists else None

    def mock_validate(raise_exception, *args, **kwargs):
        assert raise_exception
        if validate:
            return True
        raise SchemaError(u'Schema validation failed.')

    def mock_core(source_data, schema_data, *args, **kwargs):
        assert source_data == data_mock
        assert schema_data == schema_mock
        return mock.Mock(validate=mock_validate)

    core_mock = mock.Mock(side_effect=mock_core)

    with mock.patch('pykwalify.core.Core', core_mock), \
         pytest.raises(expected_error) if expected_error else nullcontext():
        scl._yaml_validate(data_mock, schema_mock)

    if schema_exists:
        core_mock.assert_called_once()
    else:
        core_mock.assert_not_called()


def test_yaml_load_empty_file(tmp_path):
    quarantine_file = tmp_path / 'empty_quarantine.yml'
    quarantine_file.write_text("# yaml file without data")
    with pytest.raises(scl.EmptyYamlFileException):
        scl.yaml_load_verify(quarantine_file, None)