1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
"""Test schemas modules."""
import json
import logging
import subprocess
import sys
import urllib
from pathlib import Path
from time import sleep
from typing import Any
from unittest.mock import DEFAULT, MagicMock, patch
import pytest
import spdx.config
from ansiblelint.file_utils import Lintable
from ansiblelint.schemas import __file__ as schema_module
from ansiblelint.schemas.__main__ import refresh_schemas
from ansiblelint.schemas.main import validate_file_schema
schema_path = Path(schema_module).parent
spdx_config_path = Path(spdx.config.__file__).parent
def test_refresh_schemas() -> None:
"""Test for schema update skip."""
# This is written as a single test in order to avoid concurrency issues,
# which caused random issues on CI when the two tests run in parallel
# and or in different order.
assert refresh_schemas(min_age_seconds=3600 * 24 * 365 * 10) == 0
sleep(1)
# this should disable the cache and force an update
assert refresh_schemas(min_age_seconds=0) == 1
sleep(1)
# should be cached now
assert refresh_schemas(min_age_seconds=10) == 0
def urlopen_side_effect(*_args: Any, **kwargs: Any) -> DEFAULT:
"""Actual test that timeout parameter is defined."""
assert "timeout" in kwargs
assert kwargs["timeout"] > 0
return DEFAULT
@patch("urllib.request")
def test_requests_uses_timeout(mock_request: MagicMock) -> None:
"""Test that schema refresh uses timeout."""
mock_request.urlopen.side_effect = urlopen_side_effect
refresh_schemas(min_age_seconds=0)
mock_request.urlopen.assert_called()
@patch("urllib.request")
def test_request_timeouterror_handling(
mock_request: MagicMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that schema refresh can handle time out errors."""
error_msg = "Simulating handshake operation time out."
mock_request.urlopen.side_effect = urllib.error.URLError(TimeoutError(error_msg))
with caplog.at_level(logging.DEBUG):
assert refresh_schemas(min_age_seconds=0) == 1
mock_request.urlopen.assert_called()
assert "Skipped schema refresh due to unexpected exception: " in caplog.text
assert error_msg in caplog.text
def test_schema_refresh_cli() -> None:
"""Ensure that we test the cli schema refresh command."""
proc = subprocess.run(
[sys.executable, "-m", "ansiblelint.schemas"],
check=False,
capture_output=True,
text=True,
)
assert proc.returncode == 0
def test_validate_file_schema() -> None:
"""Test file schema validation failure on unknown file kind."""
lintable = Lintable("foo.bar", kind="")
result = validate_file_schema(lintable)
assert len(result) == 1, result
assert "Unable to find JSON Schema" in result[0]
def test_spdx() -> None:
"""Test that SPDX license identifiers are in sync."""
_licenses = spdx_config_path / "licenses.json"
license_ids = set()
with _licenses.open(encoding="utf-8") as license_fh:
licenses = json.load(license_fh)
for lic in licenses["licenses"]:
if lic.get("isDeprecatedLicenseId"):
continue
license_ids.add(lic["licenseId"])
galaxy_json = schema_path / "galaxy.json"
with galaxy_json.open(encoding="utf-8") as f:
schema = json.load(f)
spx_enum = schema["$defs"]["SPDXLicenseEnum"]["enum"]
if set(spx_enum) != license_ids:
with galaxy_json.open("w", encoding="utf-8") as f:
schema["$defs"]["SPDXLicenseEnum"]["enum"] = sorted(license_ids)
json.dump(schema, f, indent=2)
pytest.fail(
"SPDX license list inside galaxy.json JSON Schema file was updated.",
)
|
