Handle optional and unsupported challenge tests

tools/dojo/parse-dojo-yml

@@ -91,6 +91,7 @@ class ChallengeResource(BaseModel):
     progression_locked: Optional[bool] = None
     interfaces: Optional[list] = None
     visibility: Optional[Visibility] = None
+    auxiliary: Optional[dict] = None
 
 
 class HeaderResource(BaseModel):

tools/pwnshop/src/pwnshop/commands/test.py

@@ -9,6 +9,7 @@
 
 import click
 import requests
+import yaml
 from rich.progress import (
     BarColumn,
     MofNCompleteColumn,
@@ -23,6 +24,35 @@
 
 logger = logging.getLogger(__name__)
 
+UNSUPPORTED_TEST_EXIT_CODE = 77
+
+
+def _read_metadata(path: pathlib.Path) -> dict:
+    if not path.is_file():
+        return {}
+    metadata = yaml.safe_load(path.read_text()) or {}
+    return metadata if isinstance(metadata, dict) else {}
+
+
+def _read_pwnshop_metadata(path: pathlib.Path) -> dict:
+    auxiliary = _read_metadata(path).get("auxiliary")
+    if not isinstance(auxiliary, dict):
+        return {}
+    metadata = auxiliary.get("pwnshop")
+    return metadata if isinstance(metadata, dict) else {}
+
+
+def _allows_unsupported_tests(challenge_path: pathlib.Path) -> bool:
+    return _read_pwnshop_metadata(challenge_path / "challenge.yml").get("allow_unsupported_tests") is True
+
+
+def _challenge_requires_solve(challenge_path: pathlib.Path) -> bool:
+    module = _read_metadata(challenge_path.parent / "module.yml")
+    for resource in module.get("resources", []):
+        if resource.get("type") == "challenge" and resource.get("id") == challenge_path.name:
+            return resource.get("required", True) is not False
+    return True
+
 
 def run_workspace_command(
     workspace_url: str,
@@ -73,7 +103,7 @@ def run_workspace_command(
     default=None,
     help="Timeout in seconds for each individual test.",
 )
-@click.option("--require-solved", is_flag=True, help="Fail if any challenge is unsolved.")
+@click.option("--require-solved", is_flag=True, help="Fail if any required challenge is unsolved.")
 @click.option(
     "--log-failures",
     metavar="DIR",
@@ -110,22 +140,25 @@ def test_challenge(challenge_path):
         challenge_path = pathlib.Path(challenge_path)
         rendered = None
         try:
+            allow_unsupported_tests = _allows_unsupported_tests(challenge_path)
+            requires_solve = _challenge_requires_solve(challenge_path)
             rendered = lib.render_challenge(challenge_path)
             image_id = lib.build_challenge(challenge_path)
             tests = sorted(rendered.rglob("test*/test_*"))
             if not tests:
                 logger.warning("no tests found for %s", challenge_path)
-                return {"path": challenge_path, "tests": [], "solved": False}
+                return {"path": challenge_path, "tests": [], "solved": not requires_solve}
             logger.info("running %d test(s) for %s", len(tests), challenge_path)
             results = []
-            solved = False
+            solved = not requires_solve
             for test in tests:
                 test_name = test.relative_to(rendered)
                 logger.debug("running test %s in %s", test_name, challenge_path)
                 passed = False
                 last_output = ""
                 failed_attempt_outputs = []
                 for attempt in range(1, attempts + 1):
+                    test_unsupported = False
                     logger.debug("running test %s in %s (attempt %d/%d)", test_name, challenge_path, attempt, attempts)
                     with lib.run_challenge(challenge_path, image_id, volumes=[test]) as (
                         _container,
@@ -148,17 +181,18 @@ def test_challenge(challenge_path):
                                 last_output += e.stderr
                             passed = False
                         else:
-                            passed = run.returncode == 0
+                            test_unsupported = allow_unsupported_tests and run.returncode == UNSUPPORTED_TEST_EXIT_CODE
                             last_output = (run.stdout or "") + (run.stderr or "")
+                            passed = run.returncode == 0 or test_unsupported
                             logger.debug(
                                 "test %s %s (rc=%d, attempt %d/%d)",
                                 test_name,
-                                "PASSED" if passed else "FAILED",
+                                "UNSUPPORTED" if test_unsupported else "PASSED" if passed else "FAILED",
                                 run.returncode,
                                 attempt,
                                 attempts,
                             )
-                    solved = solved or flag in last_output
+                    solved = solved or test_unsupported or flag in last_output
                     if passed:
                         if attempt > 1:
                             logger.info(
@@ -217,7 +251,8 @@ def test_challenge(challenge_path):
                 else:
                     console.print(f"[red]FAIL[/] {challenge}: {error}")
             elif not tests:
-                unsolved.add(challenge)
+                if not solved:
+                    unsolved.add(challenge)
                 passed_count += 1
             else:
                 for test_path, passed, output in tests:

tools/pwnshop/tests/test_unsupported.py

@@ -0,0 +1,179 @@
+import contextlib
+import json
+import pathlib
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+from click.testing import CliRunner
+
+
+PWNSHOP_ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(PWNSHOP_ROOT / "src"))
+
+from pwnshop.commands import test as test_command  # noqa: E402
+
+
+class TestCommandMetadataTests(unittest.TestCase):
+    @staticmethod
+    def challenge_context(flag):
+        @contextlib.contextmanager
+        def manager():
+            yield "container-id", "http://workspace/", flag
+
+        return manager()
+
+    def invoke_test_command(
+        self,
+        returncode,
+        *,
+        allow_unsupported_tests=False,
+        attempts=1,
+        required=None,
+    ):
+        with tempfile.TemporaryDirectory() as directory:
+            module = pathlib.Path(directory)
+            challenge = module / "challenge"
+            challenge.mkdir()
+            if allow_unsupported_tests:
+                (challenge / "challenge.yml").write_text("auxiliary:\n  pwnshop:\n    allow_unsupported_tests: true\n")
+            if required is not None:
+                (module / "module.yml").write_text(
+                    f"resources:\n  - type: challenge\n    id: challenge\n    required: {str(required).lower()}\n"
+                )
+
+            rendered = module / "rendered"
+            test = rendered / "tests_private" / "test_solve.py"
+            test.parent.mkdir(parents=True)
+            test.write_text("#!/usr/bin/python3\n")
+
+            arguments = [
+                "--jobs",
+                "1",
+                "--attempts",
+                str(attempts),
+                "--require-solved",
+                str(challenge),
+            ]
+            outcome = subprocess.CompletedProcess(
+                [str(test)],
+                returncode,
+                stdout="test output without the generated flag\n",
+                stderr="",
+            )
+
+            with (
+                mock.patch.object(test_command.lib, "resolve_targets", return_value=[challenge]),
+                mock.patch.object(test_command.lib, "render_challenge", return_value=rendered),
+                mock.patch.object(test_command.lib, "build_challenge", return_value="image-id"),
+                mock.patch.object(
+                    test_command.lib,
+                    "run_challenge",
+                    side_effect=lambda *_args, **_kwargs: self.challenge_context("generated-flag"),
+                ) as run_challenge,
+                mock.patch.object(test_command, "run_workspace_command", return_value=outcome) as run_test,
+            ):
+                result = CliRunner().invoke(test_command.test_command, arguments)
+
+        return result, run_challenge.call_count, run_test.call_count
+
+    def test_optional_challenge_does_not_need_to_emit_the_flag(self):
+        result, container_calls, test_calls = self.invoke_test_command(0, required=False)
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        self.assertEqual((container_calls, test_calls), (1, 1))
+        self.assertNotIn("Unsolved challenges", result.output)
+
+    def test_optional_challenge_test_failure_still_fails(self):
+        result, container_calls, test_calls = self.invoke_test_command(1, required=False)
+
+        self.assertEqual(result.exit_code, 1)
+        self.assertEqual((container_calls, test_calls), (1, 1))
+        self.assertIn("Some tests failed", result.output)
+        self.assertNotIn("Unsolved challenges", result.output)
+
+    def test_unsupported_test_is_exempt_and_not_retried(self):
+        result, container_calls, test_calls = self.invoke_test_command(
+            test_command.UNSUPPORTED_TEST_EXIT_CODE,
+            allow_unsupported_tests=True,
+            attempts=4,
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        self.assertEqual((container_calls, test_calls), (1, 1))
+        self.assertNotIn("Unsolved challenges", result.output)
+
+    def test_unsupported_exit_requires_opt_in(self):
+        result, container_calls, test_calls = self.invoke_test_command(test_command.UNSUPPORTED_TEST_EXIT_CODE)
+
+        self.assertEqual(result.exit_code, 1)
+        self.assertEqual((container_calls, test_calls), (1, 1))
+        self.assertIn("Some tests failed", result.output)
+
+    def test_unsupported_opt_in_requires_auxiliary_pwnshop_boolean(self):
+        invalid_metadata = [
+            "allow_unsupported_tests: true\n",
+            "auxiliary:\n",
+            "auxiliary: invalid\n",
+            "auxiliary:\n  pwnshop:\n",
+            "auxiliary:\n  pwnshop: invalid\n",
+            'auxiliary:\n  pwnshop:\n    allow_unsupported_tests: "true"\n',
+        ]
+        with tempfile.TemporaryDirectory() as directory:
+            challenge = pathlib.Path(directory)
+            challenge_yml = challenge / "challenge.yml"
+            for metadata in invalid_metadata:
+                with self.subTest(metadata=metadata):
+                    challenge_yml.write_text(metadata)
+                    self.assertFalse(test_command._allows_unsupported_tests(challenge))
+
+    def test_dojo_parser_preserves_auxiliary_metadata(self):
+        with tempfile.TemporaryDirectory() as directory:
+            dojo = pathlib.Path(directory)
+            module = dojo / "module"
+            challenge = module / "challenge"
+            challenge.mkdir(parents=True)
+            (dojo / "dojo.yml").write_text("id: test-dojo\nname: Test Dojo\nmodules:\n- id: module\n")
+            (module / "module.yml").write_text(
+                "name: Test Module\nresources:\n- type: challenge\n  id: challenge\n  name: Test Challenge\n"
+            )
+            (challenge / "challenge.yml").write_text(
+                "auxiliary:\n  pwnshop:\n    allow_unsupported_tests: true\n  unrelated:\n    preserved: true\n"
+            )
+
+            parse_dojo_yml = PWNSHOP_ROOT.parents[1] / "tools" / "dojo" / "parse-dojo-yml"
+            result = subprocess.run(
+                [str(parse_dojo_yml), str(dojo / "dojo.yml"), "--json"],
+                text=True,
+                capture_output=True,
+            )
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+        auxiliary = json.loads(result.stdout)["modules"][0]["resources"][0]["auxiliary"]
+        self.assertEqual(
+            auxiliary,
+            {
+                "pwnshop": {"allow_unsupported_tests": True},
+                "unrelated": {"preserved": True},
+            },
+        )
+
+    def test_other_nonzero_exit_is_not_unsupported(self):
+        result, container_calls, test_calls = self.invoke_test_command(1, allow_unsupported_tests=True)
+
+        self.assertEqual(result.exit_code, 1)
+        self.assertEqual((container_calls, test_calls), (1, 1))
+        self.assertIn("Some tests failed", result.output)
+
+    def test_required_challenge_without_flag_is_still_unsolved(self):
+        result, container_calls, test_calls = self.invoke_test_command(0)
+
+        self.assertEqual(result.exit_code, 1)
+        self.assertEqual((container_calls, test_calls), (1, 1))
+        self.assertIn("Unsolved challenges", result.output)
+
+
+if __name__ == "__main__":
+    unittest.main()