diff --git a/examples/tests/v3/basic_flask_server.py b/examples/tests/v3/basic_flask_server.py new file mode 100644 index 0000000000..3722825cff --- /dev/null +++ b/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[^ ]+)") + 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/") + 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() diff --git a/examples/tests/v3/test_matchers.py b/examples/tests/v3/test_matchers.py new file mode 100644 index 0000000000..78affa78ec --- /dev/null +++ b/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() diff --git a/src/pact/v3/matchers/__init__.py b/src/pact/v3/matchers/__init__.py new file mode 100644 index 0000000000..e8b23a4b6d --- /dev/null +++ b/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", +] diff --git a/src/pact/v3/matchers/matchers.py b/src/pact/v3/matchers/matchers.py new file mode 100644 index 0000000000..4659472349 --- /dev/null +++ b/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()