Python DSC Adapter and Test Resource implementation

.gitignore

@@ -4,6 +4,7 @@ target/
 bin/
 .DS_Store
 *.msix
+.vs/
 
 # Generated files for tree-sitter
 grammars/**/bindings/
@@ -16,3 +17,9 @@ tree-sitter-dscexpression/
 lcov.info
 *.profraw
 *.profdata
+
+/adapters/__pycache__
+/adapters/python/__pycache__
+/adapters/python/pyDscAdapter/__pycache__
+/adapters/python/tests/__pycache__
+/adapters/python/tests/src/__pycache__

adapters/powershell/Tests/powershellgroup.resource.tests.ps1

@@ -279,8 +279,8 @@ Describe 'PowerShell adapter resource tests' {
             $adapterPath = Join-Path $PSScriptRoot 'TestAdapter'
             $env:PATH += [System.IO.Path]::PathSeparator + $adapterPath
 
-            $r = '{"TestCaseId": 1}' | dsc resource test -r 'Test/TestCase' -f -
-            $LASTEXITCODE | Should -Be 0
+            $r = '{"TestCaseId": 1}' | dsc resource test -r 'Test/TestCase' -f - 2> $TestDrive/tracing.txt
+            $LASTEXITCODE | Should -Be 0 -Because (Get-Content -Path $TestDrive/tracing.txt | Out-String)
             $resources = $r | ConvertFrom-Json
             $resources.actualState.TestCaseId | Should -Be 1
         }

adapters/python/.project.data.json

@@ -0,0 +1,10 @@
+{
+    "Name": "python-adapter",
+    "Kind": "Adapter",
+    "CopyFiles": {
+        "All": [
+            "pyDscAdapter/*",
+            "pythonadapter.dsc.resource.json"
+        ]
+    }
+}

adapters/python/pyDscAdapter/__main__.py

@@ -0,0 +1,6 @@
+import sys
+from cli import main  # TODO: Currently using absolute imports. Switch to relative imports if this is later used as a module in the adapter manifest; otherwise, keep as-is for direct script execution.
+
+if __name__ == "__main__":
+    sys.exit(main())
+

adapters/python/pyDscAdapter/cli.py

@@ -0,0 +1,65 @@
+import sys
+import json
+import argparse
+from typing import Optional
+# TODO: Currently using absolute imports. Switch to relative imports if this is later used as a module in the adapter manifest; otherwise, keep as-is for direct script execution.
+from adapter import ResourceAdapter
+
+# --------------------
+# CLI / entrypoint API
+# --------------------
+def _build_parser() -> argparse.ArgumentParser:
+    """Construct the argument parser for the DSC adapter CLI."""
+    parser = argparse.ArgumentParser(
+        prog="pyDscAdapter",
+        description="DSC v3 Python adapter."
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    adapter = sub.add_parser("adapter", help="Adapter operations")
+    adapter.add_argument("--operation", required=True, choices=["list", "get", "set", "test", "export", "validate"],
+                         help="Adapter operation to execute.")
+    adapter.add_argument("--input", default="{}", help="JSON string with resource configuration (single input).")
+    adapter.add_argument("--resource", dest="ResourceType", default="", help="Resource type selector (e.g., Microsoft.Linux.Apt/Package).")
+    adapter.add_argument("--resource-path", dest="ResourcePath", default="", help="Optional resource module file path.")
+    return parser
+
+
+def main(argv: Optional[list] = None) -> int:
+    """Main entry point for the DSC adapter CLI."""
+    parser = _build_parser()
+    args = parser.parse_args(argv)
+
+    if args.command != "adapter":
+        print(json.dumps({"error": "Unsupported command"}))
+        return 2
+
+    adapter = ResourceAdapter()
+
+
+    # 1. Start with --input as the authoritative source
+    input_str = args.input
+
+    # 2. If stdin has data, it overrides --input (DSC convention) for operations that accept input
+    if args.operation in ("get", "set", "test", "export", "validate"):
+        stdin_data = sys.stdin.read().strip() if not sys.stdin.isatty() else ""
+        if stdin_data:
+            input_str = stdin_data
+
+    # 3. Call operation handler
+    exit_code, result = adapter.run_operation(
+        args.operation,
+        input_str,
+        args.ResourceType,
+        getattr(args, "ResourcePath", "")
+    )
+
+    # If set branch (or similar) already wrote to stdout, skip emitting a wrapper
+    if isinstance(result, dict) and result.get("_stdout_emitted"):
+        return exit_code
+
+    # 4. Capture EXACT output passed to DSC
+    out_json = json.dumps(result, ensure_ascii=False)
+
+    print(out_json)
+    return exit_code

adapters/python/pyDscAdapter/discovery.py

@@ -0,0 +1,200 @@
+import importlib.util
+from pathlib import Path
+from typing import Any, Dict
+
+try:
+    import tomllib as toml_parser
+except ModuleNotFoundError:
+    toml_parser = None
+
+
+def _load_pyproject_data(pyproject_path: Path) -> Dict[str, Any]:
+    """Load pyproject.toml into a dictionary when possible."""
+    pyproject_path = Path(pyproject_path)
+    if not pyproject_path.exists() or toml_parser is None:
+        return {}
+
+    try:
+        with pyproject_path.open("rb") as f:
+            data = toml_parser.load(f)
+        return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def get_project_metadata_from_pyproject(pyproject_path: Path) -> Dict[str, str]:
+    """Parse [project] metadata from pyproject.toml."""
+    pyproject_path = Path(pyproject_path)
+
+    default_metadata = {
+        "version": "",
+        "description": "",
+        "author": "",
+    }
+
+    if not pyproject_path.exists():
+        return default_metadata
+
+    data = _load_pyproject_data(pyproject_path)
+    if data:
+        project = data.get("project", {})
+        if not isinstance(project, dict):
+            return default_metadata
+
+        authors = project.get("authors", [])
+        author = ""
+        if isinstance(authors, list):
+            for item in authors:
+                if isinstance(item, dict) and item.get("name"):
+                    author = str(item["name"])
+                    break
+
+        return {
+            "version": str(project.get("version", "") or ""),
+            "description": str(project.get("description", "") or ""),
+            "author": author,
+        }
+
+    try:
+        content = pyproject_path.read_text(encoding="utf-8")
+    except Exception:
+        return default_metadata
+
+    in_project = False
+    in_authors = False
+    metadata = dict(default_metadata)
+
+    for raw_line in content.splitlines():
+        stripped = raw_line.strip()
+        if stripped == "[project]":
+            in_project = True
+            in_authors = False
+            continue
+        if in_project and stripped.startswith("[") and stripped != "[[project.authors]]":
+            if not in_authors:
+                break
+        if not in_project:
+            continue
+
+        if stripped == "[[project.authors]]":
+            in_authors = True
+            continue
+
+        if in_authors:
+            if stripped.startswith("name") and "=" in stripped and not metadata["author"]:
+                _, value = stripped.split("=", 1)
+                metadata["author"] = value.strip().strip('"\'')
+            elif stripped.startswith("[[") or (stripped.startswith("[") and stripped != "[[project.authors]]"):
+                in_authors = False
+            continue
+
+        if stripped.startswith("version") and "=" in stripped:
+            _, value = stripped.split("=", 1)
+            metadata["version"] = value.strip().strip('"\'')
+        elif stripped.startswith("description") and "=" in stripped:
+            _, value = stripped.split("=", 1)
+            metadata["description"] = value.strip().strip('"\'')
+
+    return metadata
+
+def get_class_map_from_pyproject(pyproject_path: Path) -> Dict[str, str]:
+    """
+    Parse [tool.dsc.resources] section from pyproject.toml.
+    Returns: {"ResourceType": "ClassName", ...}
+    No external dependencies required.
+    """
+    pyproject_path = Path(pyproject_path)
+
+    if not pyproject_path.exists():
+        return {}
+
+    data = _load_pyproject_data(pyproject_path)
+    if data:
+        resources = (
+            data.get("tool", {})
+                .get("dsc", {})
+                .get("resources", {})
+        )
+        if isinstance(resources, dict):
+            class_map: Dict[str, str] = {}
+            for resource_type, resource_value in resources.items():
+                if isinstance(resource_value, dict):
+                    class_name = resource_value.get("class", "")
+                    if class_name:
+                        class_map[str(resource_type)] = str(class_name)
+                elif resource_value is not None:
+                    class_map[str(resource_type)] = str(resource_value)
+            if class_map:
+                return class_map
+
+    try:
+        content = pyproject_path.read_text(encoding="utf-8")
+    except Exception:
+        return {}
+
+    class_map = {}
+    in_section = False
+
+    for line in content.splitlines():
+        stripped = line.strip()
+        if stripped == "[tool.dsc.resources]":
+            in_section = True
+            continue
+        if in_section:
+            if stripped.startswith("["):
+                break
+            if "=" in stripped and not stripped.startswith("#"):
+                key, val = stripped.split("=", 1)
+                key = key.strip().strip('"\'')
+                val = val.strip().strip('"\'')
+                class_map[key] = val
+
+    return class_map
+
+
+def get_resource_metadata_from_pyproject(pyproject_path: Path) -> Dict[str, Dict[str, str]]:
+    """Parse per-resource metadata from pyproject.toml."""
+    pyproject_path = Path(pyproject_path)
+
+    if not pyproject_path.exists():
+        return {}
+
+    data = _load_pyproject_data(pyproject_path)
+    if not data:
+        return {}
+
+    resources = (
+        data.get("tool", {})
+            .get("dsc", {})
+            .get("resources", {})
+    )
+    if not isinstance(resources, dict):
+        return {}
+
+    metadata: Dict[str, Dict[str, str]] = {}
+    for resource_type, resource_value in resources.items():
+        if not isinstance(resource_value, dict):
+            continue
+
+        metadata[str(resource_type)] = {
+            "version": str(resource_value.get("version", "") or ""),
+            "description": str(resource_value.get("description", "") or ""),
+            "author": str(resource_value.get("author", "") or ""),
+        }
+
+    return metadata
+
+def import_class_from_file(resource_path: Path, resource_type: str, class_name: str) -> type:
+    """Dynamically import a class from a given file path."""
+    module_name = f"dsc_{resource_type.replace('/', '_').replace('.', '_').lower()}" #if resource_type else f"dsc_{resource_path.stem.lower()}"
+    spec = importlib.util.spec_from_file_location(module_name, str(resource_path))
+    if not spec or not spec.loader:
+        raise ImportError(f"Unable to load module '{resource_path}'")
+
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    try:
+        return getattr(mod, class_name)
+    except AttributeError as e:
+        raise ImportError(f"Class '{class_name}' not found in '{resource_path}': {e}")
+