|
2 | 2 | from pact import Verifier |
3 | 3 | import pact_ffi |
4 | 4 |
|
| 5 | +from typing import Literal, Any |
| 6 | + |
| 7 | +def provider_state_handler( |
| 8 | + state: str, |
| 9 | + action: Literal["setup", "teardown"], |
| 10 | + parameters: dict[str, Any] | None, |
| 11 | +) -> None: |
| 12 | + """Handle all provider state changes.""" |
| 13 | + parameters = parameters or {} |
| 14 | + if state == "a product with ID 10 exists": |
| 15 | + if action == "setup": |
| 16 | + return |
| 17 | + if action == "teardown": |
| 18 | + return |
| 19 | + msg = f"Unknown state/action: {state}/{action}" |
| 20 | + raise ValueError(msg) |
| 21 | + |
| 22 | + |
5 | 23 | def test_provider(): |
6 | 24 | """Test the provider against the consumer contract.""" |
7 | 25 | pact_ffi.log_to_stderr("DEBUG") |
8 | 26 | verifier = ( |
9 | 27 | Verifier("pactflow-example-provider-python") # Provider name |
10 | 28 | .add_transport(url="http://localhost:8000") |
| 29 | + .state_handler(provider_state_handler, teardown=False) |
11 | 30 | .broker_source( |
12 | | - os.environ.get("PACT_BROKER_BASE_URL", "https://broker.pactflow.io/"), |
13 | | - token=os.environ.get("PACT_BROKER_TOKEN", "broker-token"), # or use token="bearer-token" |
| 31 | + url=os.environ.get("PACT_BROKER_BASE_URL", "https://broker.pactflow.io/"), |
| 32 | + token=os.environ.get("PACT_BROKER_TOKEN", "broker-token"), |
| 33 | + selector=True |
14 | 34 | ) |
| 35 | + .consumer_versions('{"mainBranch": true}', '{"deployedOrReleased": true}') |
| 36 | + .include_pending() |
| 37 | + .include_wip_since("2025-11-10") |
| 38 | + .build() |
15 | 39 | ) |
16 | | - |
| 40 | + # verifier.set_publish_options( |
| 41 | + # version="123", |
| 42 | + # branch="foo", |
| 43 | + # ) |
17 | 44 | verifier.verify() |
18 | 45 |
|
19 | 46 | if __name__ == "__main__": |
|
0 commit comments