adding matcher POC

examples/tests/v3/basic_flask_server.py

@@ -0,0 +1,109 @@
+import logging
+import re
+import signal
+import subprocess
+import sys
+import time
+from contextlib import contextmanager
+from pathlib import Path
+from threading import Thread
+from typing import Generator, NoReturn
+
+import requests
+from flask import Flask, Response, make_response
+from yarl import URL
+
+logger = logging.getLogger(__name__)
+
+@contextmanager
+def start_provider() -> Generator[URL, None, None]:  # noqa: C901
+    """
+    Start the provider app.
+    """
+    process = subprocess.Popen(  # noqa: S603
+        [
+            sys.executable,
+            Path(__file__),
+        ],
+        cwd=Path.cwd(),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        encoding="utf-8",
+    )
+
+    pattern = re.compile(r" \* Running on (?P<url>[^ ]+)")
+    while True:
+        if process.poll() is not None:
+            logger.error("Provider process exited with code %d", process.returncode)
+            logger.error(
+                "Provider stdout: %s", process.stdout.read() if process.stdout else ""
+            )
+            logger.error(
+                "Provider stderr: %s", process.stderr.read() if process.stderr else ""
+            )
+            msg = f"Provider process exited with code {process.returncode}"
+            raise RuntimeError(msg)
+        if (
+            process.stderr
+            and (line := process.stderr.readline())
+            and (match := pattern.match(line))
+        ):
+            break
+        time.sleep(0.1)
+
+    url = URL(match.group("url"))
+    logger.debug("Provider started on %s", url)
+    for _ in range(50):
+        try:
+            response = requests.get(str(url / "_test" / "ping"), timeout=1)
+            assert response.text == "pong"
+            break
+        except (requests.RequestException, AssertionError):
+            time.sleep(0.1)
+            continue
+    else:
+        msg = "Failed to ping provider"
+        raise RuntimeError(msg)
+
+    def redirect() -> NoReturn:
+        while True:
+            if process.stdout:
+                while line := process.stdout.readline():
+                    logger.debug("Provider stdout: %s", line.strip())
+            if process.stderr:
+                while line := process.stderr.readline():
+                    logger.debug("Provider stderr: %s", line.strip())
+
+    thread = Thread(target=redirect, daemon=True)
+    thread.start()
+
+    try:
+        yield url
+    finally:
+        process.send_signal(signal.SIGINT)
+
+if __name__ == "__main__":
+    app = Flask(__name__)
+
+    @app.route("/path/to/<test_id>")
+    def hello_world(test_id: int) -> Response:
+        response = make_response(
+            {
+                "response": {
+                    "id": test_id,
+                    "regex": "must end with 'hello world'",
+                    "integer": 42,
+                    "include": "hello world",
+                    "minMaxArray": [1.0, 1.1, 1.2],
+                }
+            }
+        )
+        response.headers["SpecialHeader"] = "Special: Hi"
+        return response
+
+    @app.get("/_test/ping")
+    def ping() -> str:
+        """Simple ping endpoint for testing."""
+        return "pong"
+
+    app.run()

examples/tests/v3/test_matchers.py

@@ -0,0 +1,51 @@
+import json
+from pathlib import Path
+
+from examples.tests.v3.basic_flask_server import start_provider
+from pact.v3 import Pact, Verifier, matchers
+
+
+def test_matchers() -> None:
+    pact_dir = Path(Path(__file__).parent.parent.parent / "pacts")
+    pact = Pact("consumer", "provider").with_specification("V4")
+    (
+        pact.upon_receiving("a request")
+        .given("a state")
+        .with_request(
+            "GET",
+            json.dumps(
+                matchers.regex("/path/to/100", r"/path/to/\d+", generator="Regex")
+            ),
+        )
+        .will_respond_with(200)
+        .with_body(
+            json.dumps({
+                "response": matchers.like(
+                    {
+                        "regex": matchers.regex(
+                            "must end with 'hello world'", r".*hello world'$"
+                        ),
+                        "integer": matchers.integer(42),
+                        "include": matchers.include("hello world", "world"),
+                        "minMaxArray": matchers.each_like(
+                            matchers.decimal(1.0),
+                            min_count=3,
+                            max_count=5,
+                        ),
+                    },
+                    min_count=1,
+                )
+            })
+        )
+        .with_header(
+            "SpecialHeader", json.dumps(matchers.regex("Special: Foo", r"Special: \w+"))
+        )
+    )
+    pact.write_file(pact_dir, overwrite=True)
+    with start_provider() as url:
+        verifier = (
+            Verifier()
+            .set_info("My Provider", url=url)
+            .add_source(pact_dir / "consumer-provider.json")
+        )
+        verifier.verify()

src/pact/v3/matchers/__init__.py

@@ -0,0 +1,18 @@
+from .matchers import (  # noqa: TID252
+    decimal,
+    each_like,
+    include,
+    integer,
+    like,
+    regex,
+)
+
+__all__ = [
+    "decimal",
+    "each_like",
+    "integer",
+    "include",
+    "like",
+    "type",
+    "regex",
+]

src/pact/v3/matchers/matchers.py

@@ -0,0 +1,73 @@
+from collections.abc import Iterable
+from typing import Any, Dict, Optional
+
+
+class Matcher:
+    def __init__(
+        self,
+        matcher_type: str,
+        value: Optional[Any] = None, # noqa: ANN401
+        generator: Optional[str] = None,
+        **kwargs: Dict[str, Any],
+    ) -> None:
+        self.type = matcher_type
+        self.value = value
+        self.generator = generator
+        self.extra_attrs = {k: v for k, v in kwargs.items() if v is not None}
+
+    def to_json(self) -> Dict[str, Any]:
+        json_data = {
+            "pact:matcher:type": self.type,
+        }
+        if self.value is not None:
+            if isinstance(self.value, Matcher):
+                json_data["value"] = self.value.to_json()
+            elif isinstance(self.value, Iterable) and not isinstance(self.value, str):
+                if isinstance(self.value, dict):
+                    json_data["value"] = {
+                        k: v.to_json() if isinstance(v, Matcher) else v
+                        for k, v in self.value.items()
+                    }
+                else:
+                    json_data["value"] = [
+                        v.to_json() if isinstance(v, Matcher) else v
+                        for v in self.value
+                    ]
+            else:
+                json_data["value"] = self.value
+        if self.generator is not None:
+            json_data["pact:generator:type"] = self.generator
+        [json_data.update({k: v}) for k, v in self.extra_attrs.items() if v is not None]
+        return json_data
+
+
+def decimal(value: float) -> Dict[str, Any]:
+    return Matcher("decimal", value).to_json()
+
+
+def each_like(
+    value: Any, # noqa: ANN401
+    min_count: Optional[int] = None,
+    max_count: Optional[int] = None,
+) -> Dict[str, Any]:
+    return Matcher("type", [value], min=min_count, max=max_count).to_json()
+
+
+def include(value: str, include: str) -> Dict[str, Any]:
+    return Matcher("include", value, include=include).to_json()
+
+
+def integer(value: int) -> Dict[str, Any]:
+    return Matcher("integer", value).to_json()
+
+
+def like(
+    value: Any, # noqa: ANN401
+    min_count: Optional[int] = None,
+    max_count: Optional[int] = None,
+) -> Dict[str, Any]:
+    return Matcher("type", value, min=min_count, max=max_count).to_json()
+
+
+def regex(value: str, regex: str, generator: Optional[str] = None) -> Dict[str, Any]:
+    return Matcher("regex", value, regex=regex, generator=generator).to_json()