Merge pull request #136 from openSUSE/toms/complexity-process_file

Reduce complexity of `process_file`

Author: tomschr

changelog.d/135.refactor.rst

@@ -0,0 +1 @@
+Reduce complexity of :func:`~docbuild.cli.cmd_validate.process.process_file`. Split into smaller functions.

src/docbuild/cli/cmd_validate/process.py

@@ -2,6 +2,7 @@
 
 import asyncio
 from collections.abc import Iterator
+from dataclasses import dataclass
 from datetime import date
 import logging
 from pathlib import Path
@@ -215,6 +216,85 @@ async def run_python_checks(
     return check_results
 
 
+@dataclass
+class ValidationResult:
+    """Normalized result of RNG validation.
+
+    :ivar success: True when validation passed.
+    :ivar exit_code: Exit code to return when validation fails (0 for success).
+    :ivar message: Optional human-readable message describing the failure.
+    """
+
+    success: bool
+    exit_code: int
+    message: str = ""
+
+
+def build_shortname(filepath: Path | str) -> str:
+    """Return a shortened display name for ``filepath``.
+
+    :param filepath: Path-like object to shorten.
+    :returns: Shortened display name (last two path components or full path).
+    """
+    path_obj = Path(filepath)
+    return "/".join(path_obj.parts[-2:]) if len(path_obj.parts) >= 2 else str(filepath)
+
+
+async def run_validation(filepath: Path | str, method: str) -> ValidationResult:
+    """Run RNG validation using the selected method and normalize result.
+
+    :param filepath: Path to the XML file to validate.
+    :param method: Validation method name ("jing" or "lxml").
+    :returns: A :class:`ValidationResult` describing the outcome.
+    """
+    path_obj = Path(filepath)
+    if method == "lxml":
+        rng_success, rng_output = await asyncio.to_thread(
+            validate_rng_lxml, path_obj, XMLDATADIR / "product-config-schema.rng"
+        )
+        if rng_success:
+            return ValidationResult(True, 0, "")
+        return ValidationResult(False, 10, rng_output or "")
+
+    if method == "jing":
+        jing_result = await validate_rng(path_obj, idcheck=True)
+        if jing_result.returncode != 0:
+            output = (jing_result.stdout or "") + (jing_result.stderr or "")
+            return ValidationResult(False, 10, output.strip())
+        return ValidationResult(True, 0, "")
+
+    return ValidationResult(False, 11, f"Unknown validation method: {method}")
+
+
+async def parse_tree(filepath: Path | str) -> etree._ElementTree:
+    """Parse XML file using lxml in a background thread.
+
+    Exceptions from :func:`lxml.etree.parse` (for example
+    :class:`lxml.etree.XMLSyntaxError`) are propagated to the caller.
+
+    :param filepath: Path to the XML file to parse.
+    :returns: Parsed :class:`lxml.etree._ElementTree`.
+    """
+    return await asyncio.to_thread(etree.parse, str(filepath), parser=None)
+
+
+async def run_checks_and_display(
+    tree: etree._ElementTree, shortname: str, context: DocBuildContext, max_len: int
+) -> bool:
+    """Execute registered Python checks and print formatted results.
+
+    :param tree: Parsed XML tree to check.
+    :param shortname: Short name used for display output.
+    :param context: :class:`DocBuildContext` used to read verbosity.
+    :param max_len: Maximum length used for aligned output.
+    :returns: True when all checks succeeded (or when no checks are registered).
+    """
+    check_results = await run_python_checks(tree)
+    if check_results:
+        display_results(shortname, check_results, context.verbose, max_len)
+    return all(result.success for _, result in check_results)
+
+
 async def process_file(
     filepath: Path | str,
     context: DocBuildContext,
@@ -228,63 +308,23 @@ async def process_file(
     :param rng_schema_path: Optional path to an RNG schema for validation.
     :return: An exit code (0 for success, non-zero for failure).
     """
-    # Shorten the filename to last two parts for display
-    path_obj = Path(filepath)
-    shortname = (
-        "/".join(path_obj.parts[-2:]) if len(path_obj.parts) >= 2 else str(filepath)
-    )
-
-    # IDEA: Should we replace jing and validate with etree.RelaxNG?
-    #
-    # 1. RNG Validation
-    validation_method = context.validation_method
+    shortname = build_shortname(filepath)
 
-    if validation_method == "lxml":
-        # Use lxml-based validator (requires .rng schema)
-        rng_success, rng_output = await asyncio.to_thread(
-            validate_rng_lxml,
-            path_obj,
-            XMLDATADIR / "product-config-schema.rng",
-        )
-    elif validation_method == "jing":
-        # Use existing jing-based validator (.rnc or .rng)
-        jing_result = await validate_rng(path_obj, idcheck=True)
-    else:
+    # 1. RNG Validation (normalized)
+    validation = await run_validation(filepath, context.validation_method)
+    if not validation.success:
         console_err.print(
             f"{shortname:<{max_len}}: RNG validation => [red]failed[/red]"
         )
-        console_err.print(
-            f"  [bold red]Error:[/] Unknown validation method: {validation_method}"
-        )
-        return 11  # Custom error code for unknown validation method
+        if validation.message:
+            console_err.print(f"  [bold red]Error:[/] {validation.message}")
+        return validation.exit_code
 
-    # Handle validation result for jing
-    if validation_method == "jing":
-        if jing_result.returncode != 0:
-            console_err.print(
-                f"{shortname:<{max_len}}: RNG validation => [red]failed[/red]"
-            )
-            output = (jing_result.stdout or "") + (jing_result.stderr or "")
-            if output:
-                console_err.print(f"  [bold red]Error:[/] {output.strip()}")
-            return 10  # Specific error code for RNG failure
-
-    # Handle validation result for lxml
-    if validation_method == "lxml":
-        if not rng_success:
-            console_err.print(
-                f"{shortname:<{max_len}}: RNG validation => [red]failed[/red]"
-            )
-            if rng_output:
-                console_err.print(f"  [bold red]Error:[/] {rng_output}")
-            return 10
-
-    # 2. Python-based checks
+    # 2. Parse XML and run Python checks
     try:
-        tree = await asyncio.to_thread(etree.parse, str(filepath), parser=None)
+        tree = await parse_tree(filepath)
 
     except etree.XMLSyntaxError as err:
-        # This can happen if xmllint passes but lxml's parser is stricter.
         console_err.print(
             f"{shortname:<{max_len}}: XML Syntax Error => [red]failed[/red]"
         )
@@ -295,14 +335,8 @@ async def process_file(
         console_err.print(f"  [bold red]Error:[/] {err}")
         return 200
 
-    # Run all checks for this file
-    check_results = await run_python_checks(tree)
-
-    if check_results:
-        # Display results based on verbosity level
-        display_results(shortname, check_results, context.verbose, max_len)
-
-    return 0 if all(result.success for _, result in check_results) else 1
+    success = await run_checks_and_display(tree, shortname, context, max_len)
+    return 0 if success else 1
 
 
 async def process(

tests/cli/cmd_repo/test_cmd_clone.py

@@ -1,6 +1,7 @@
 """Tests for the 'docbuild repo clone' command."""
 
 import logging
+import re
 from unittest.mock import AsyncMock
 
 import pytest
@@ -112,15 +113,16 @@ async def test_process_stitchnode_none(monkeypatch, tmp_path):
     (tmp_path / "config").mkdir()
     (tmp_path / "repos").mkdir()
 
-    with pytest.raises(ValueError, match="Stitch node could not be created."):
+    with pytest.raises(ValueError,
+                       match=re.escape("Stitch node could not be created.")):
         await mod_process.process(context, repos=())
 
 
 async def test_process_configdir_none():
     context = DocBuildContext(envconfig={"paths": {}})
     with pytest.raises(
         ValueError,
-        match="Could not get a value from envconfig.paths.config_dir",
+        match=re.escape("Could not get a value from envconfig.paths.config_dir"),
     ):
         await mod_process.process(context, repos=())
 
@@ -129,6 +131,6 @@ async def test_process_repodir_none():
     context = DocBuildContext(envconfig={"paths": {"config_dir": "/dummy/config"}})
     with pytest.raises(
         ValueError,
-        match="Could not get a value from envconfig.paths.repo_dir",
+        match=re.escape("Could not get a value from envconfig.paths.repo_dir"),
     ):
         await mod_process.process(context, repos=())

tests/cli/cmd_validate/test_process.py

@@ -1,6 +1,7 @@
 """Tests for the XML validation process module."""
 
 from pathlib import Path
+import re
 from subprocess import CompletedProcess
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -58,7 +59,8 @@ async def test_process_file_with_generic_parsing_error(
 
 async def test_process_no_envconfig(mock_context):
     mock_context.envconfig = None
-    with pytest.raises(ValueError, match="No envconfig found in context."):
+    with pytest.raises(ValueError,
+                       match=re.escape("No envconfig found in context.")):
         await process(mock_context, xmlfiles=(Path("dummy.xml"),))
 
 

tests/cli/cmd_validate/validate/test_cmd_validate.py

@@ -3,6 +3,7 @@
 from collections.abc import Iterator
 from os import PathLike
 from pathlib import Path
+import re
 from subprocess import CompletedProcess
 import tempfile
 from unittest.mock import AsyncMock, Mock, patch
@@ -123,7 +124,8 @@ async def test_process_invalid_paths_config(self):
         context = Mock(spec=DocBuildContext)
         context.envconfig = {"paths": "not_a_dict"}
 
-        with pytest.raises(ValueError, match="'paths.config' must be a dictionary"):
+        with pytest.raises(ValueError,
+                           match=re.escape("'paths.config' must be a dictionary")):
             await process_mod.process(context, [])
 
     async def test_process_with_no_xml_files(self, mock_context, caplog):

tests/cli/cmd_validate/validate/test_process_validation.py

@@ -223,3 +223,79 @@ async def test_process_file_with_xmlsyntax_error(capsys, tmp_path):
         result = await process_mod.process_file(xmlfile, mock_context, 40)
 
     assert result == 20
+
+
+def test_validate_rng_lxml_success(monkeypatch):
+    """When RelaxNG validates the XML, function returns (True, '')."""
+    fake_xml_doc = object()
+
+    # parse called twice: first for schema, second for xml file
+    calls = {"n": 0}
+
+    def fake_parse(path):
+        calls["n"] += 1
+        return object() if calls["n"] == 1 else fake_xml_doc
+
+    monkeypatch.setattr(process_mod.etree, "parse", fake_parse)
+
+    fake_relaxng = Mock()
+    fake_relaxng.validate.return_value = True
+    monkeypatch.setattr(process_mod.etree, "RelaxNG", lambda doc: fake_relaxng)
+
+    ok, msg = process_mod.validate_rng_lxml(Path("/tmp/f.xml"), Path("/tmp/schema.rng"))
+    assert ok is True
+    assert msg == ""
+
+
+def test_validate_rng_lxml_validation_failure(monkeypatch):
+    """When RelaxNG.validate() returns False, function returns (False, error_log)."""
+    monkeypatch.setattr(process_mod.etree, "parse", lambda p: object())
+    fake_relaxng = Mock()
+    fake_relaxng.validate.return_value = False
+    fake_relaxng.error_log = "some relaxng errors"
+    monkeypatch.setattr(process_mod.etree, "RelaxNG", lambda doc: fake_relaxng)
+
+    ok, msg = process_mod.validate_rng_lxml(Path("/tmp/f.xml"), Path("/tmp/schema.rng"))
+    assert ok is False
+    assert "some relaxng errors" in msg
+
+
+def test_validate_rng_lxml_xml_syntax_error(monkeypatch):
+    """If lxml raises XMLSyntaxError, function returns a descriptive message."""
+
+    def raise_syntax(path):
+        raise process_mod.etree.XMLSyntaxError("bad xml", None, 0, 0, "file")
+
+    monkeypatch.setattr(process_mod.etree, "parse", raise_syntax)
+
+    ok, msg = process_mod.validate_rng_lxml(Path("/tmp/f.xml"), Path("/tmp/schema.rng"))
+    assert ok is False
+    assert "XML or RNG syntax error" in msg
+
+
+def test_validate_rng_lxml_relaxng_parse_error(monkeypatch):
+    """If RelaxNG constructor raises RelaxNGParseError, return an error message."""
+    monkeypatch.setattr(process_mod.etree, "parse", lambda p: object())
+
+    def raise_relaxng(doc):
+        raise process_mod.etree.RelaxNGParseError("schema parse failure")
+
+    monkeypatch.setattr(process_mod.etree, "RelaxNG", raise_relaxng)
+
+    ok, msg = process_mod.validate_rng_lxml(Path("/tmp/f.xml"), Path("/tmp/schema.rng"))
+    assert ok is False
+    assert "RELAX NG schema parsing error" in msg
+
+
+def test_validate_rng_lxml_generic_exception(monkeypatch):
+    """Any other exception is caught and reported as an unexpected error."""
+    monkeypatch.setattr(process_mod.etree, "parse", lambda p: object())
+
+    def raise_generic(doc):
+        raise Exception("boom")
+
+    monkeypatch.setattr(process_mod.etree, "RelaxNG", raise_generic)
+
+    ok, msg = process_mod.validate_rng_lxml(Path("/tmp/f.xml"), Path("/tmp/schema.rng"))
+    assert ok is False
+    assert "An unexpected error occurred during validation" in msg

tests/models/test_language.py

@@ -73,7 +73,8 @@ def test_compare_with_one_language_and_with_different_object():
 def test_language_code_is_frozen():
     lang = LanguageCode(language="en-us")
 
-    with pytest.raises(ValidationError, match=".*frozen_instance.*"):
+    with pytest.raises(ValidationError,
+                       match=r".*frozen_instance.*"):
         lang.language = "de-de"
 
 

tests/utils/test_pidlock.py

@@ -6,6 +6,7 @@
 from multiprocessing import Event
 from pathlib import Path
 import platform
+import re
 import time
 from unittest.mock import Mock, patch
 
@@ -143,7 +144,7 @@ def test_lock_is_reentrant_per_process(lock_setup):
         # (internal API misuse)
         with pytest.raises(
             RuntimeError,
-            match="Lock already acquired by this PidFileLock instance."
+            match=re.escape("Lock already acquired by this PidFileLock instance.")
         ):
             with lock2:
                 pass