test: add readme test for compiler

examples/pddl/domain.pddl

@@ -3,7 +3,7 @@
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 (define (domain blocks)
-  (:requirements :strips :typing)
+  (:requirements :strips :typing :disjunctive-preconditions)
   (:types block)
   (:predicates (on ?x - block ?y - block)
 	       (ontable ?x - block)

tests/conftest.py

@@ -21,3 +21,26 @@
 #
 
 """This module contains fixtures for the tests."""
+import pytest
+from docker import DockerClient
+
+from tests.helpers.planutils.base import PlanutilsDockerImage
+from tests.helpers.planutils.val import VALWrapper
+
+
+@pytest.fixture(scope="session")
+def docker_client() -> DockerClient:
+    """Return the docker client."""
+    return DockerClient()
+
+
+@pytest.fixture(scope="session")
+def planutils_docker_image(docker_client) -> PlanutilsDockerImage:  # pylint: disable=redefined-outer-name
+    """Return the planutils docker image."""
+    return PlanutilsDockerImage(docker_client)
+
+
+@pytest.fixture(scope="session")
+def val(planutils_docker_image) -> VALWrapper:  # pylint: disable=redefined-outer-name
+    """Return the val wrapper."""
+    return VALWrapper(planutils_docker_image)

tests/helpers/docker.py

@@ -150,7 +150,11 @@ def stop_if_already_running(self):
                 container.wait()
                 logger.info("Image %s stopped", self.tag)
 
-    def run(self, cmd: Sequence[str]) -> str:
+    def run(self, cmd: Sequence[str], **kwargs) -> str:
         """Create the container."""
-        stdout = self._client.containers.run(self.tag, command=cmd, remove=True)
+        assert "command" not in kwargs
+        assert "remove" not in kwargs
+        stdout = self._client.containers.run(
+            self.tag, command=cmd, remove=True, **kwargs
+        )
         return stdout.decode("utf-8")

tests/helpers/planutils/val.py

@@ -22,13 +22,18 @@
 
 
 """This module contains the VAL wrapper for the planutils docker image."""
+import logging
 import re
+import shutil
 from dataclasses import dataclass
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import Optional, Sequence
 
 from tests.helpers.docker import DockerImage
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass(frozen=True)
 class VALResult:
@@ -76,20 +81,45 @@ def _validate(self, domain: Path, problem: Optional[Path]) -> VALResult:
             assert problem.exists(), f"problem {problem} does not exist"
             assert problem.is_file(), f"problem {problem} is not a file"
 
-        cmd = ["planutils", "run", "val", "Parser", str(domain)]
-        if problem is not None:
-            cmd.append(str(problem))
-
-        stdout = self._planutils_docker.run(cmd)
-        result = self._process_output(stdout)
-        return result
+        # move files to a temporary directory
+        with TemporaryDirectory() as tmp_dir:
+            tmp_dir_path = Path(tmp_dir)
+
+            domain_tmp = tmp_dir_path / domain.name
+            shutil.copy(domain, domain_tmp)
+
+            if problem is not None:
+                problem_tmp = tmp_dir_path / problem.name
+                shutil.copy(problem, problem_tmp)
+
+            bind_path = Path("/root/temp")
+            volumes = {
+                str(tmp_dir_path): {
+                    "bind": str(bind_path),
+                    "mode": "rw",
+                },
+            }
+
+            cmd = [
+                "planutils",
+                "run",
+                "val",
+                "Parser",
+                str(bind_path / domain_tmp.name),
+            ]
+            if problem is not None:
+                cmd.append(str(bind_path / problem_tmp.name))
+
+            stdout = self._planutils_docker.run(cmd, volumes=volumes)
+            result = self._process_output(stdout)
+            return filter_val_result(result)
 
     def _process_output(self, output: str) -> VALResult:
         """Process the output of the validator."""
         match = re.search("Errors: ([0-9]+), warnings: ([0-9]+)\n", output)
         assert match is not None, f"output {output} does not match the expected format"
 
-        start = match.endpos
+        start = match.end()
         errors_and_warnings = output[start:]
         errors_and_warnings_lines = errors_and_warnings.splitlines(keepends=False)
 
@@ -104,3 +134,35 @@ def _process_output(self, output: str) -> VALResult:
                 raise ValueError(f"line {line} does not match the expected format")
 
         return VALResult(errors=errors, warnings=warnings)
+
+
+def filter_val_result(val_result: VALResult) -> VALResult:
+    """
+    Check the validity of the result.
+
+    This method filters out some warnings and errors that at the moment are not considered relevant for our library.
+
+    For example, since the compilation makes the domain goal-dependent, it requires to define certain objects that are
+     then redefined in the problem. This causes the error 'Re-declaration of symbol in same scope'.
+
+    Similarly, the presence of constants related to the goal in the domain file causes the warning message 'Undeclared
+     symbol'.
+    """
+    ERRORS_TO_IGNORE = ["Re-declaration of symbol in same scope"]
+    WARNINGS_TO_IGNORE = ["Undeclared symbol"]
+
+    filtered_errors = []
+    for error in val_result.errors:
+        if not any(err in error for err in ERRORS_TO_IGNORE):
+            filtered_errors.append(error)
+        else:
+            logger.debug("ignoring VAL error: %s", error)
+
+    filtered_warnings = []
+    for warning in val_result.warnings:
+        if not any(warn in warning for warn in WARNINGS_TO_IGNORE):
+            filtered_warnings.append(warning)
+        else:
+            logger.debug("ignoring VAL warning: %s", warning)
+
+    return VALResult(errors=filtered_errors, warnings=filtered_warnings)

tests/test_compiler.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2021 -- 2023 WhiteMech
+#
+# ------------------------------
+#
+# This file is part of Plan4Past.
+#
+# Plan4Past is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Plan4Past is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Plan4Past.  If not, see <https://www.gnu.org/licenses/>.
+#
+
+"""This module contain tests for the compiler module."""
+
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from pddl.formatter import domain_to_string, problem_to_string
+from pddl.parser.domain import DomainParser
+from pddl.parser.problem import ProblemParser
+from pylogics.parsers import parse_pltl
+
+from plan4past.compiler import Compiler
+from tests.helpers.constants import EXAMPLES_DIR
+
+
+def test_readme_example(val) -> None:
+    """Test the example from the README."""
+    formula = "on_b_a & O(ontable_c)"
+    domain_parser = DomainParser()
+    problem_parser = ProblemParser()
+
+    pddl_example_domain_path = EXAMPLES_DIR / Path("pddl/domain.pddl")
+    pddl_example_problem_path = EXAMPLES_DIR / Path("pddl/p-0.pddl")
+
+    domain = domain_parser(pddl_example_domain_path.read_text(encoding="utf-8"))
+    problem = problem_parser(pddl_example_problem_path.read_text(encoding="utf-8"))
+    goal = parse_pltl(formula)
+
+    compiler = Compiler(domain, problem, goal)
+    compiler.compile()
+    compiled_domain, compiled_problem = compiler.result
+
+    with TemporaryDirectory() as tmpdir:
+        tmpdir_path = Path(tmpdir)
+        new_domain_path = tmpdir_path / "new-domain.pddl"
+        new_problem_path = tmpdir_path / "new-problem.pddl"
+        with open(new_domain_path, "w+", encoding="utf-8") as d:
+            d.write(domain_to_string(compiled_domain))
+        with open(new_problem_path, "w+", encoding="utf-8") as p:
+            p.write(problem_to_string(compiled_problem))
+
+        result = val.validate_problem(new_domain_path, new_problem_path)
+        assert result.is_valid(strict=True)