Adding upstream version 6.13.1.upstream/6.13.1upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 99 insertions, 0 deletions

diff --git a/test/test_schemas.py b/test/test_schemas.py
new file mode 100644
index 0000000..5c89c11
--- /dev/null
+++ b/test/test_schemas.py
@@ -0,0 +1,99 @@
+"""Test schemas modules."""
+import logging
+import subprocess
+import sys
+import urllib
+from time import sleep
+from typing import Any
+from unittest.mock import DEFAULT, MagicMock, patch
+
+import pytest
+
+from ansiblelint.file_utils import Lintable
+from ansiblelint.rules import RulesCollection
+from ansiblelint.runner import Runner
+from ansiblelint.schemas import refresh_schemas, validate_file_schema
+
+
+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
+
+
+@pytest.mark.parametrize(
+    ("file", "expected_tags"),
+    (
+        pytest.param(
+            "examples/changelogs/changelog.yaml", ["schema[changelog]"], id="changelog"
+        ),
+    ),
+)
+def test_schema(
+    default_rules_collection: RulesCollection,
+    file: str,
+    expected_tags: list[str],
+) -> None:
+    """Test that runner can go through any corner cases."""
+    runner = Runner(file, rules=default_rules_collection)
+    matches = runner.run()
+
+    assert len(matches) == len(expected_tags)
+    for i, match in enumerate(matches):
+        assert match.tag == expected_tags[i]
+
+
+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]