diff --git a/CHANGELOG.md b/CHANGELOG.md index 5bfe8ad..efce9bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,10 +5,24 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [0.3.1] ### Added - Retry logic for fetching `/openapi.json`: the plugin now retries up to 3 times (4 total attempts) before failing, with a warning printed on each failed attempt. +- **Format-based negative test generation** for string fields annotated with + `format: email`, `uri`, `ipv4`, `ipv6`, `hostname`, `idn-hostname`, or + `uuid`. When the OpenAPI spec documents a `400` response for an endpoint, + the plugin automatically generates one request body per required + format-constrained field that contains a deliberately invalid value. The + server is expected to return 400 or 422; the plugin additionally requires + the response body to contain at least one non-whitespace string (a + human-readable error message) — an empty `{}` body causes the test to fail. +- Eight new mock servers and corresponding integration tests covering all + supported formats plus URL port-range edge cases (port 0 and ports > 65535 + are invalid; ports 1–65535 are valid). +- OpenAPI compatibility table added to `README.md` showing 3.0.x vs 3.1.x + feature support (as of v0.3.0). +- Format validation documented in `docs/how-it-works/test-cases.md`. ## [0.3.0] - 2026-03-26 diff --git a/README.md b/README.md index 60f07cd..f8119a1 100644 --- a/README.md +++ b/README.md @@ -136,6 +136,7 @@ Each OpenAPI test appears as an individual pytest test item. ✔️ Compares live responses against examples ✔️ Produces a readable test report ✔️ Supports OpenAPI 3.0 and **3.1.x** (nullable types, `const`, `$ref` siblings, `allOf`) +✔️ Generates **format-based negative tests** for `email`, `uri`, `ipv4`, `ipv6`, `hostname`, `uuid` fields — expects 400/422 with a descriptive error message # ▶️ Detailed Example @@ -218,12 +219,52 @@ See here an example server - `email-server`: [tests/test_servers/email_server/se [tests/test_servers/email_server/email_test_output.txt](tests/test_servers/email_server/email_test_output.txt) +## Format-Based Negative Test Generation + +When a string property in an OpenAPI request body declares a `format`, pytest-openapi automatically generates an invalid value and sends it to the server. The server **must** return 400 or 422 **and** include a human-readable error message in the body (an empty `{}` body causes the test to fail). + +| Format | Tested? | Example invalid value | +|---|---|---| +| `email` | ✔️ | `not-an-email` | +| `uri` / `url` | ✔️ | `not-a-uri` | +| `ipv4` / `ip` | ✔️ | `999.999.999.999` | +| `ipv6` | ✔️ | `gggg::1` | +| `hostname` / `idn-hostname` | ✔️ | `-invalid.start.com` | +| `uuid` | ✔️ | `not-a-uuid` | +| `date` / `date-time` / `time` | ❌ (out of scope) | — | + +## OpenAPI Version Compatibility + +As of **v0.3.0**, pytest-openapi supports both OpenAPI 3.0.x and OpenAPI 3.1.x. + +|pytest-openapi|OpenAPI| +|--------------|-------| +|0.3.x |3.0.x | +|0.3.x |3.1.x | + +The plan is to keep this backwards compatible for now. + + +Here is a detailed breakdown. + +| Feature | OpenAPI 3.0.x | OpenAPI 3.1.x | +|---|---|---| +| Basic types (`string`, `integer`, `boolean`, `array`, `object`) | ✔️ | ✔️ | +| `nullable: true` | ✔️ | ✔️ (via `type: ["string", "null"]`) | +| `type: ["string", "null"]` array syntax | ❌ | ✔️ | +| `$ref` resolution (`components/schemas`) | ✔️ | ✔️ | +| `$ref` siblings (keywords next to `$ref`) | ❌ | ✔️ | +| `allOf` composition | ✔️ | ✔️ | +| `oneOf` / `anyOf` | ✔️ | ✔️ | +| `const` keyword | ✔️ | ✔️ | +| `enum` validation & negative tests | ✔️ | ✔️ | +| Format-based negative tests (`email`, `uri`, …) | ✔️ | ✔️ | + # Future Plans / TODO This is a work in progress. - [ ] A check that the example matches the schema - [ ] Ask that 400 responses be in the documentation. -- [ ] A check for regexp and email formats. ## Issues? Feedback? diff --git a/docs/how-it-works/test-cases.md b/docs/how-it-works/test-cases.md index 1143b61..7822854 100644 --- a/docs/how-it-works/test-cases.md +++ b/docs/how-it-works/test-cases.md @@ -21,6 +21,27 @@ Before generating test cases, schemas are fully resolved: For object schemas, pytest-openapi generates test cases by taking the Cartesian product of valid values for each property. Only valid enum values are included in these combinations to avoid generating request bodies that would be rejected by the server. +In addition to the valid combinations, one extra object is generated for every string property annotated with a recognised format (see below). That extra object replaces only the format-constrained field with a deliberately invalid value while keeping all other fields valid. This produces a negative test case that must trigger a 400 or 422 from the server. + +### Format-Based Negative Tests + +When an OpenAPI schema marks a string property with one of the following formats, pytest-openapi generates an invalid value and expects the server to return **400 Bad Request** or **422 Unprocessable Entity** with a **non-empty, human-readable error message** in the response body. + +| Format | Example valid value | Example invalid value used | +|---|---|---| +| `email` | `test@example.com` | `not-an-email` | +| `uri` / `url` | `https://example.com/path` | `not-a-uri` | +| `ipv4` / `ip` | `192.168.1.1` | `999.999.999.999` | +| `ipv6` | `2001:db8::1` | `gggg::1` | +| `hostname` / `idn-hostname` | `api.example.com` | `-invalid.start.com` | +| `uuid` | `550e8400-e29b-41d4-a716-446655440000` | `not-a-uuid` | + +Formats intentionally excluded: `date`, `date-time`, `time` (server-side validation for these is inconsistent and out of scope). + +#### Error message requirement + +When the server returns 400 or 422 for an invalid-format value, the response body must contain at least one non-whitespace string. An empty body (`{}`) causes the test to **fail**. The plugin does not check for any specific word — it only verifies that a human-readable explanation is present. + ## Validation Strategy pytest-openapi uses different validation approaches depending on the test case origin: diff --git a/src/pytest_openapi/__init__.py b/src/pytest_openapi/__init__.py index 493f741..260c070 100644 --- a/src/pytest_openapi/__init__.py +++ b/src/pytest_openapi/__init__.py @@ -1 +1 @@ -__version__ = "0.3.0" +__version__ = "0.3.1" diff --git a/src/pytest_openapi/case_generator.py b/src/pytest_openapi/case_generator.py index 7295e84..20b1410 100644 --- a/src/pytest_openapi/case_generator.py +++ b/src/pytest_openapi/case_generator.py @@ -5,6 +5,22 @@ from .schema import primary_type, resolve_schema +_INVALID_FORMAT_VALUES = { + "email": [ + "user name@domain.com", + "user@@double.com", + "user;name@domain.com", + ], + "uri": ["not-a-uri", "://missing-host"], + "url": ["not-a-url", "://missing-host"], + "ipv4": ["999.999.999.999", "1.2.3"], + "ip": ["999.999.999.999", "1.2.3"], + "ipv6": ["gggg::1", "not-ipv6"], + "hostname": ["-invalid.start.com", "has space.com"], + "idn-hostname": ["-invalid.start.com", "has space.com"], + "uuid": ["not-a-uuid", "xxxxxxxx-not-a-valid-uuid"], +} + def generate_invalid_enum_value(schema): """Generate an invalid value for an enum field. @@ -178,7 +194,16 @@ def generate_string_test_cases(schema, valid_only=False): target_length = min_length + 5 filtered.append("a" * target_length) - return filtered if filtered else ["test-string"] + result = filtered if filtered else ["test-string"] + + # Append invalid-format values for negative tests (not subject to length + # constraints — they must be recognisably broken to trigger server-side 400) + if not valid_only and "format" in schema: + fmt = schema["format"] + if fmt in _INVALID_FORMAT_VALUES: + result = result + _INVALID_FORMAT_VALUES[fmt] + + return result def generate_integer_test_cases(schema, field_name="field", valid_only=False): @@ -460,6 +485,7 @@ def generate_object_test_cases(schema, field_name="field", spec=None): tuple: (list of test objects, list of warnings) """ properties = schema.get("properties", {}) + required_props = set(schema.get("required", [])) warnings = [] @@ -495,6 +521,21 @@ def generate_object_test_cases(schema, field_name="field", spec=None): if not combos: combos = [{}] + # Append one format-invalid object per REQUIRED format-constrained property. + # Required fields must always be present and are the most likely to be + # server-validated; non-required format fields are skipped to keep the + # test-case count predictable for servers that don't validate format. + base = combos[0] + for prop_name, prop_schema in properties.items(): + if prop_name not in required_props: + continue + resolved = resolve_schema(spec, prop_schema) + fmt = resolved.get("format") + if fmt and fmt in _INVALID_FORMAT_VALUES: + invalid_obj = dict(base) + invalid_obj[prop_name] = _INVALID_FORMAT_VALUES[fmt][0] + combos.append(invalid_obj) + return combos, warnings diff --git a/src/pytest_openapi/contract.py b/src/pytest_openapi/contract.py index 62a4a19..7599f7a 100644 --- a/src/pytest_openapi/contract.py +++ b/src/pytest_openapi/contract.py @@ -5,7 +5,10 @@ import requests -from .case_generator import generate_test_cases_for_schema +from .case_generator import ( + _INVALID_FORMAT_VALUES, + generate_test_cases_for_schema, +) from .schema import primary_type, resolve_schema # Global list to store test reports @@ -386,6 +389,68 @@ def contains_invalid_enum_value(schema, data, path="", spec=None): return False +def _has_nonempty_string_value(data): + """Return True if data contains at least one non-whitespace + string.""" + if isinstance(data, str): + return bool(data.strip()) + if isinstance(data, dict): + return any(_has_nonempty_string_value(v) for v in data.values()) + if isinstance(data, list): + return any(_has_nonempty_string_value(item) for item in data) + return False + + +def contains_invalid_format_value(schema, data, path="", spec=None): + """Check if data contains any invalid format values according to + schema. + + Walks the schema/data tree and returns True when a string field annotated + with a known format (email, uri, ipv4, ipv6, hostname, uuid, …) holds one + of the deliberately broken values from _INVALID_FORMAT_VALUES. + + Args: + schema: OpenAPI schema object (may contain $ref or allOf) + data: Request/response data to check + path: Dot-separated path for tracking nested location + spec: Full OpenAPI spec dict for $ref resolution + + Returns: + bool: True if data contains an invalid format value, False otherwise + """ + if not isinstance(schema, dict): + return False + + schema = resolve_schema(spec, schema) + schema_type = primary_type(schema.get("type")) + + if schema_type == "string" and "format" in schema: + fmt = schema["format"] + if fmt in _INVALID_FORMAT_VALUES and isinstance(data, str): + if data in _INVALID_FORMAT_VALUES[fmt]: + return True + + if schema_type == "object" and isinstance(data, dict): + properties = schema.get("properties", {}) + for prop, prop_schema in properties.items(): + if prop in data: + child_path = f"{path}.{prop}" if path else prop + if contains_invalid_format_value( + prop_schema, data[prop], child_path, spec + ): + return True + + elif schema_type == "array" and isinstance(data, list): + items_schema = schema.get("items", {}) + for i, item in enumerate(data): + if contains_invalid_format_value( + items_schema, item, f"{path}[{i}]", spec + ): + return True + + return False + + def validate_against_schema(schema, actual, path="", spec=None): """Validate actual response against JSON schema. @@ -1087,6 +1152,10 @@ def test_post_endpoint( is_negative_test = contains_invalid_enum_value( request_schema, request_test_case ) + if not is_negative_test and 400 in documented_statuses: + is_negative_test = contains_invalid_format_value( + request_schema, request_test_case + ) # Make the POST request try: @@ -1111,7 +1180,7 @@ def test_post_endpoint( errors.append(error_msg) continue - # If this is a negative test (invalid enum), expect 400 + # If this is a negative test (invalid value), expect 400 if is_negative_test: # Parse response try: @@ -1121,13 +1190,34 @@ def test_post_endpoint( # Should get 400 Bad Request or 422 Unprocessable Entity if response.status_code in [400, 422]: - # Success - invalid enum was properly rejected + if not _has_nonempty_string_value(actual_response): + error_msg = ( + f"Got {response.status_code} for invalid input but " + "the response body contains no descriptive error " + "message. The server must return a human-readable " + "explanation." + ) + log_test_result( + "POST", + path, + request_test_case, + 400, + "400/422 with descriptive message", + response.status_code, + actual_response, + False, + error_msg, + test_origin, + documented_statuses=documented_statuses, + ) + errors.append(error_msg) + continue log_test_result( "POST", path, request_test_case, 400, - "400/422 (invalid enum value)", + "400/422 (invalid value)", response.status_code, actual_response, True, @@ -1139,9 +1229,9 @@ def test_post_endpoint( elif response.status_code >= 500: # Server error - this is bad, should have returned 400/422 error_msg = ( - f"Expected 400/422 for invalid enum value, got " + f"Expected 400/422 for invalid value, got " f"{response.status_code} (server error). Server " - "should validate enum values and return 400 or " + "should validate input and return 400 or " "422, not 5xx." ) log_test_result( @@ -1149,7 +1239,7 @@ def test_post_endpoint( path, request_test_case, 400, - "400/422 (invalid enum value)", + "400/422 (invalid value)", response.status_code, actual_response, False, @@ -1160,12 +1250,12 @@ def test_post_endpoint( errors.append(error_msg) continue else: - # Got 200/201 or other status - # Invalid enum should have been rejected + # Got 200/201 or other status — invalid value should have been + # rejected error_msg = ( - f"Expected 400 or 422 for invalid enum value, got " + f"Expected 400 or 422 for invalid value, got " f"{response.status_code}. Server should validate " - "enum values and return 400 Bad Request or " + "input and return 400 Bad Request or " "422 Unprocessable Entity." ) log_test_result( @@ -1173,7 +1263,7 @@ def test_post_endpoint( path, request_test_case, 400, - "400 Bad Request (invalid enum value)", + "400 Bad Request (invalid value)", response.status_code, actual_response, False, @@ -1608,6 +1698,10 @@ def test_put_endpoint( is_negative_test = contains_invalid_enum_value( request_schema, request_test_case ) + if not is_negative_test and 400 in documented_statuses: + is_negative_test = contains_invalid_format_value( + request_schema, request_test_case + ) # Make the PUT request try: @@ -1632,7 +1726,7 @@ def test_put_endpoint( errors.append(error_msg) continue - # If this is a negative test (invalid enum), expect 400 + # If this is a negative test (invalid value), expect 400 if is_negative_test: # Parse response try: @@ -1642,13 +1736,34 @@ def test_put_endpoint( # Should get 400 Bad Request or 422 Unprocessable Entity if response.status_code in [400, 422]: - # Success - invalid enum was properly rejected + if not _has_nonempty_string_value(actual_response): + error_msg = ( + f"Got {response.status_code} for invalid input but " + "the response body contains no descriptive error " + "message. The server must return a human-readable " + "explanation." + ) + log_test_result( + "PUT", + resolved_path, + request_test_case, + 400, + "400/422 with descriptive message", + response.status_code, + actual_response, + False, + error_msg, + test_origin, + documented_statuses=documented_statuses, + ) + errors.append(error_msg) + continue log_test_result( "PUT", resolved_path, request_test_case, 400, - "400/422 (invalid enum value)", + "400/422 (invalid value)", response.status_code, actual_response, True, @@ -1658,11 +1773,10 @@ def test_put_endpoint( ) continue elif response.status_code >= 500: - # Server error - this is bad, should have returned 400/422 error_msg = ( - f"Expected 400/422 for invalid enum value, got " + f"Expected 400/422 for invalid value, got " f"{response.status_code} (server error). Server " - "should validate enum values and return 400 or " + "should validate input and return 400 or " "422, not 5xx." ) log_test_result( @@ -1670,7 +1784,7 @@ def test_put_endpoint( resolved_path, request_test_case, 400, - "400/422 (invalid enum value)", + "400/422 (invalid value)", response.status_code, actual_response, False, @@ -1681,12 +1795,10 @@ def test_put_endpoint( errors.append(error_msg) continue else: - # Got 200 or other status - invalid enum should have - # been rejected error_msg = ( - f"Expected 400 or 422 for invalid enum value, got " + f"Expected 400 or 422 for invalid value, got " f"{response.status_code}. Server should validate " - "enum values and return 400 Bad Request or " + "input and return 400 Bad Request or " "422 Unprocessable Entity." ) log_test_result( @@ -1694,7 +1806,7 @@ def test_put_endpoint( resolved_path, request_test_case, 400, - "400 Bad Request (invalid enum value)", + "400 Bad Request (invalid value)", response.status_code, actual_response, False, @@ -2233,6 +2345,10 @@ def test_post_endpoint_single( is_negative_test = contains_invalid_enum_value( request_schema, request_body, spec=spec ) + if not is_negative_test and 400 in documented_statuses: + is_negative_test = contains_invalid_format_value( + request_schema, request_body, spec=spec + ) # Make the POST request try: @@ -2265,12 +2381,32 @@ def test_post_endpoint_single( actual_response = response.text if response.status_code in [400, 422]: + if not _has_nonempty_string_value(actual_response): + error_msg = ( + f"Got {response.status_code} for invalid input but the " + "response body contains no descriptive error message. " + "The server must return a human-readable explanation." + ) + log_test_result( + "POST", + path, + request_body, + 400, + "400/422 with descriptive message", + response.status_code, + actual_response, + False, + error_msg, + test_origin, + documented_statuses=documented_statuses, + ) + return False, error_msg log_test_result( "POST", path, request_body, 400, - "400/422 (invalid enum value)", + "400/422 (invalid value)", response.status_code, actual_response, True, @@ -2281,9 +2417,9 @@ def test_post_endpoint_single( return True, None elif response.status_code >= 500: error_msg = ( - f"Expected 400/422 for invalid enum value, got " + f"Expected 400/422 for invalid value, got " f"{response.status_code} (server error). Server " - "should validate enum values and return 400 or " + "should validate input and return 400 or " "422, not 5xx." ) log_test_result( @@ -2291,7 +2427,7 @@ def test_post_endpoint_single( path, request_body, 400, - "400/422 (invalid enum value)", + "400/422 (invalid value)", response.status_code, actual_response, False, @@ -2302,9 +2438,9 @@ def test_post_endpoint_single( return False, error_msg else: error_msg = ( - f"Expected 400 or 422 for invalid enum value, got " + f"Expected 400 or 422 for invalid value, got " f"{response.status_code}. Server should validate " - "enum values and return 400 Bad Request or " + "input and return 400 Bad Request or " "422 Unprocessable Entity." ) log_test_result( @@ -2312,7 +2448,7 @@ def test_post_endpoint_single( path, request_body, 400, - "400 Bad Request (invalid enum value)", + "400 Bad Request (invalid value)", response.status_code, actual_response, False, @@ -2333,55 +2469,6 @@ def test_post_endpoint_single( ] ) - if is_streaming: - # Streaming responses are valid - we can't validate their - # content in the same way. Just verify we got a 200/201/202. - if response.status_code in [200, 201, 202]: - log_test_result( - "POST", - path, - request_body, - expected_status, - expected_response, - response.status_code, - collect_streaming_response(response), - True, - None, - test_origin, - documented_statuses=documented_statuses, - ) - return True, None - else: - error_msg = ( - f"Expected {expected_status}, got " - f"{response.status_code} for streaming response" - ) - log_test_result( - "POST", - path, - request_body, - expected_status, - expected_response, - response.status_code, - collect_streaming_response(response), - False, - error_msg, - test_origin, - documented_statuses=documented_statuses, - ) - return False, error_msg - - # Check if response is a streaming response - content_type = response.headers.get("Content-Type", "") - is_streaming = any( - stream_type in content_type.lower() - for stream_type in [ - "text/event-stream", - "application/x-ndjson", - "application/stream+json", - ] - ) - if is_streaming: # Streaming responses are valid - we can't validate their # content in the same way. Just verify we got a 200/201/202. diff --git a/tests/docker-compose.yaml b/tests/docker-compose.yaml index 628d510..65d4731 100644 --- a/tests/docker-compose.yaml +++ b/tests/docker-compose.yaml @@ -363,6 +363,110 @@ services: cpus: '0.1' memory: 128M + mock-server-email-format-validation: + build: + context: test_servers/email_format_validation + dockerfile: Dockerfile + image: mock-server-email-format-validation:latest + ports: + - "8029:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + + mock-server-email-format-no-message: + build: + context: test_servers/email_format_no_message + dockerfile: Dockerfile + image: mock-server-email-format-no-message:latest + ports: + - "8030:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + + mock-server-uri-format-validation: + build: + context: test_servers/uri_format_validation + dockerfile: Dockerfile + image: mock-server-uri-format-validation:latest + ports: + - "8031:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + + mock-server-uuid-format-validation: + build: + context: test_servers/uuid_format_validation + dockerfile: Dockerfile + image: mock-server-uuid-format-validation:latest + ports: + - "8032:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + + mock-server-ipv4-format-validation: + build: + context: test_servers/ipv4_format_validation + dockerfile: Dockerfile + image: mock-server-ipv4-format-validation:latest + ports: + - "8033:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + + mock-server-ipv6-format-validation: + build: + context: test_servers/ipv6_format_validation + dockerfile: Dockerfile + image: mock-server-ipv6-format-validation:latest + ports: + - "8034:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + + mock-server-hostname-format-validation: + build: + context: test_servers/hostname_format_validation + dockerfile: Dockerfile + image: mock-server-hostname-format-validation:latest + ports: + - "8035:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + + mock-server-url-port-validation: + build: + context: test_servers/url_port_validation + dockerfile: Dockerfile + image: mock-server-url-port-validation:latest + ports: + - "8036:8000" + deploy: + resources: + limits: + cpus: '0.1' + memory: 128M + test: build: context: .. diff --git a/tests/test_integration.py b/tests/test_integration.py index cba6981..2dffd73 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1694,3 +1694,548 @@ def test_post_202_spec_but_server_returns_200_fails(): assert ( "202" in output or "200" in output ), f"Expected error mentioning the status code mismatch, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_email_format_valid_address_passes(): + """Test that a POST endpoint with format: email accepts a valid email address. + + The contact server documents 'email' field with format: email and validates + it server-side. A valid address like alice@example.com should produce 200. + """ + print( + "\n🔍 Testing valid email address passes format validation...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-email-format-validation:8000", + "-k", + "contact", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + or "test_openapi[POST /contact" in output + ), f"Expected contact endpoint test to appear, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_email_format_invalid_address_gets_400(): + """Test that the plugin generates invalid email test cases and the server + returns 400 for them, which the plugin must recognise as a passing test. + + Invalid addresses like 'sinan,ozel @somehere' (space, comma, no valid domain) + must be generated by the plugin and the server must respond with 400. + The plugin should treat a 400 response to an invalid-format input as PASS. + """ + print( + "\n🔍 Testing invalid email format is generated and server returns 400...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-email-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + + # The plugin should generate invalid email test cases from the schema + assert ( + "test_openapi[POST /contact" in output + ), f"Expected POST /contact test items to be generated, got: {output}" + + # All tests should pass: valid emails get 200, invalid emails get 400 + # (400 for invalid-format input is treated as a passing contract test) + assert ( + result.returncode == 0 + ), f"Expected all email format tests to pass (valid→200, invalid→400), got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_email_format_spec_validated_successfully(): + """Test that an OpenAPI spec with format: email fields passes structural validation.""" + print( + "\n🔍 Testing OpenAPI spec with email format fields validates...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-email-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + ), f"Expected OpenAPI spec with email format to validate successfully, got: {output}" + + +# --------------------------------------------------------------------------- +# 400 with no error message — plugin must fail +# --------------------------------------------------------------------------- + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_email_format_400_without_message_fails(): + """Test that the plugin FAILS when a server returns 400 with no error description. + + A server that rejects invalid email values with an empty 400 body (no 'error' + or 'detail' key) violates the contract: the API consumer has no information + about what went wrong. The plugin must treat this as a failing test. + """ + print( + "\n🔍 Testing plugin fails when 400 has no descriptive error message...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-email-format-no-message:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + + # Plugin must fail: 400 with no descriptive body is not acceptable + assert ( + result.returncode != 0 + ), f"Expected plugin to fail when 400 response has no error message, got: {output}" + + # The failure output should mention the lack of error detail + assert ( + "400" in output + ), f"Expected output to reference the 400 status code, got: {output}" + + +# --------------------------------------------------------------------------- +# format: uri / url +# --------------------------------------------------------------------------- + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_uri_format_valid_url_passes(): + """Test that a server with format: uri accepts valid URIs (e.g. https://example.com/hook).""" + print("\n🔍 Testing valid URI format passes...", flush=True) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-uri-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + ), f"Expected OpenAPI spec with URI format to validate, got: {output}" + assert ( + "test_openapi[POST /webhook" in output + ), f"Expected POST /webhook test items, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_uri_format_invalid_url_gets_400_with_message(): + """Test that the plugin generates invalid URI values, the server returns a + descriptive 400, and the plugin treats this as a passing test. + + Invalid URIs include strings with no scheme ('not-a-uri'), a bare path + ('://broken'), or a plain hostname with no scheme ('example.com'). + """ + print( + "\n🔍 Testing invalid URI gets descriptive 400 and plugin passes...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-uri-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + # All tests should pass: valid URIs → 200, invalid URIs → 400 with message + assert ( + result.returncode == 0 + ), f"Expected all URI format tests to pass (valid→200, invalid→400+message), got: {output}" + + +# --------------------------------------------------------------------------- +# format: uuid +# --------------------------------------------------------------------------- + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_uuid_format_valid_id_passes(): + """Test that a server with format: uuid accepts a well-formed UUID.""" + print("\n🔍 Testing valid UUID format passes...", flush=True) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-uuid-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + ), f"Expected OpenAPI spec with UUID format to validate, got: {output}" + assert ( + "test_openapi[POST /resource" in output + ), f"Expected POST /resource test items, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_uuid_format_invalid_value_gets_400_with_message(): + """Test that invalid UUID values (e.g. 'not-a-uuid', truncated strings, or + strings with invalid hex characters) cause the server to return a descriptive + 400 and the plugin treats this as a passing contract test. + """ + print( + "\n🔍 Testing invalid UUID gets descriptive 400 and plugin passes...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-uuid-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + result.returncode == 0 + ), f"Expected all UUID format tests to pass (valid→200, invalid→400+message), got: {output}" + + +# --------------------------------------------------------------------------- +# format: ipv4 +# --------------------------------------------------------------------------- + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_ipv4_format_valid_address_passes(): + """Test that a server with format: ipv4 accepts a well-formed IPv4 address.""" + print("\n🔍 Testing valid IPv4 format passes...", flush=True) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-ipv4-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + ), f"Expected OpenAPI spec with IPv4 format to validate, got: {output}" + assert ( + "test_openapi[POST /allow" in output + ), f"Expected POST /allow test items, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_ipv4_format_invalid_address_gets_400_with_message(): + """Test that invalid IPv4 values (e.g. '999.999.999.999', 'not-an-ip', + or a string with only three octets) cause the server to return a + descriptive 400 and the plugin treats this as a passing contract test. + """ + print( + "\n🔍 Testing invalid IPv4 gets descriptive 400 and plugin passes...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-ipv4-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + result.returncode == 0 + ), f"Expected all IPv4 format tests to pass (valid→200, invalid→400+message), got: {output}" + + +# --------------------------------------------------------------------------- +# format: ipv6 +# --------------------------------------------------------------------------- + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_ipv6_format_valid_address_passes(): + """Test that a server with format: ipv6 accepts a well-formed IPv6 address.""" + print("\n🔍 Testing valid IPv6 format passes...", flush=True) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-ipv6-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + ), f"Expected OpenAPI spec with IPv6 format to validate, got: {output}" + assert ( + "test_openapi[POST /allow6" in output + ), f"Expected POST /allow6 test items, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_ipv6_format_invalid_address_gets_400_with_message(): + """Test that invalid IPv6 values (e.g. 'not-ipv6', 'gggg::1' with non-hex + characters, or a plain IPv4 address) cause the server to return a descriptive + 400 and the plugin treats this as a passing contract test. + """ + print( + "\n🔍 Testing invalid IPv6 gets descriptive 400 and plugin passes...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-ipv6-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + result.returncode == 0 + ), f"Expected all IPv6 format tests to pass (valid→200, invalid→400+message), got: {output}" + + +# --------------------------------------------------------------------------- +# format: hostname +# --------------------------------------------------------------------------- + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_hostname_format_valid_hostname_passes(): + """Test that a server with format: hostname accepts well-formed hostnames.""" + print("\n🔍 Testing valid hostname format passes...", flush=True) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-hostname-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + ), f"Expected OpenAPI spec with hostname format to validate, got: {output}" + assert ( + "test_openapi[POST /config" in output + ), f"Expected POST /config test items, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_hostname_format_invalid_hostname_gets_400_with_message(): + """Test that invalid hostnames (e.g. 'has space.com', '-starts-with-dash', + 'example..com' with consecutive dots) cause the server to return a + descriptive 400 and the plugin treats this as a passing contract test. + """ + print( + "\n🔍 Testing invalid hostname gets descriptive 400 and plugin passes...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-hostname-format-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + result.returncode == 0 + ), f"Expected all hostname format tests to pass (valid→200, invalid→400+message), got: {output}" + + +# --------------------------------------------------------------------------- +# URL port-number edge cases +# --------------------------------------------------------------------------- + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_url_port_valid_ports_pass(): + """Test that URLs with valid port numbers (e.g. :80, :8080, :65535) pass. + + Port numbers 1-65535 are valid for HTTP/HTTPS URLs. + The spec example uses :8080 which must produce a 200 response. + """ + print("\n🔍 Testing URL with valid port numbers passes...", flush=True) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-url-port-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + "✅ OpenAPI spec validated successfully" in output + ), f"Expected OpenAPI spec with URL port format to validate, got: {output}" + assert ( + "test_openapi[POST /webhook" in output + ), f"Expected POST /webhook test items for URL port server, got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_url_port_zero_gets_400_with_message(): + """Test that a URL containing port 0 (https://example.com:0/path) causes the + server to return a descriptive 400. + + Port 0 is reserved and not valid for HTTP URLs even though it is technically + within the 0-65535 unsigned-16-bit range. The plugin must generate this edge + case and the server must reject it with a message that mentions port 0. + """ + print( + "\n🔍 Testing URL with port 0 gets descriptive 400...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-url-port-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + # All tests should pass: valid URLs → 200, invalid-port URLs → 400 with message + assert ( + result.returncode == 0 + ), f"Expected URL port tests to pass (valid ports→200, port 0→400+message), got: {output}" + + +@pytest.mark.depends(on=["test_openapi_flag_is_recognized"]) +def test_url_port_above_65535_gets_400_with_message(): + """Test that a URL containing a port above 65535 (e.g. :65536 or :99999) + causes the server to return a descriptive 400. + + Port numbers above 65535 are outside the 16-bit unsigned range and are + never valid. The plugin must generate this edge case and the server must + reject it with a message that references the maximum port number. + """ + print( + "\n🔍 Testing URL with port > 65535 gets descriptive 400...", + flush=True, + ) + time.sleep(0.5) + + result = subprocess.run( + [ + "pytest", + "--openapi=http://mock-server-url-port-validation:8000", + "-v", + ], + capture_output=True, + text=True, + cwd="/app", + ) + + output = result.stdout + result.stderr + assert ( + result.returncode == 0 + ), f"Expected URL port tests to pass (valid ports→200, port>65535→400+message), got: {output}" diff --git a/tests/test_servers/email_format_no_message/Dockerfile b/tests/test_servers/email_format_no_message/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/email_format_no_message/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/email_format_no_message/server.py b/tests/test_servers/email_format_no_message/server.py new file mode 100644 index 0000000..4771d6e --- /dev/null +++ b/tests/test_servers/email_format_no_message/server.py @@ -0,0 +1,120 @@ +"""Mock server - validates email but returns 400 with NO descriptive body. + +This is a deliberately bad server: it rejects invalid emails with an empty +400 body instead of explaining what went wrong. The plugin should FAIL +tests against this server because a useful error message is required. +""" + +import re + +from flask import Flask, jsonify, request + +app = Flask(__name__) + +_EMAIL_RE = re.compile(r"^[^@\s,]+@[^@\s,]+\.[^@\s,]+$") + + +@app.route("/openapi.json") +def openapi(): + return jsonify( + { + "openapi": "3.0.0", + "info": { + "title": "Contact API (no error messages)", + "version": "1.0.0", + }, + "paths": { + "/contact": { + "post": { + "summary": "Submit contact form", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["name", "email"], + "properties": { + "name": { + "type": "string", + "description": "Full name", + }, + "email": { + "type": "string", + "format": "email", + "description": "Contact email address", + }, + }, + }, + "example": { + "name": "Alice", + "email": "alice@example.com", + }, + } + }, + }, + "responses": { + "200": { + "description": "Contact form received", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Submission ID", + }, + "status": { + "type": "string", + "description": "Status", + }, + }, + }, + "example": { + "id": 1, + "status": "received", + }, + } + }, + }, + "400": { + "description": "Bad request", + "content": { + "application/json": {"example": {}} + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/contact", methods=["POST"]) +def submit_contact(): + """Reject invalid emails with 400 but give NO error description.""" + global _next_id + data = request.get_json(silent=True) or {} + email = data.get("email", "") + if not isinstance(email, str) or not _EMAIL_RE.match(email): + # Deliberately empty body — no error message + return jsonify({}), 400 + result = {"id": _next_id, "status": "received"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_servers/email_format_validation/Dockerfile b/tests/test_servers/email_format_validation/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/email_format_validation/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/email_format_validation/server.py b/tests/test_servers/email_format_validation/server.py new file mode 100644 index 0000000..6cb2249 --- /dev/null +++ b/tests/test_servers/email_format_validation/server.py @@ -0,0 +1,136 @@ +"""Mock server - validates email format in request body, returns 400 for invalid emails.""" + +import re + +from flask import Flask, jsonify, request + +app = Flask(__name__) + +_EMAIL_RE = re.compile(r"^[^@\s,]+@[^@\s,]+\.[^@\s,]+$") + + +def _is_valid_email(value): + return isinstance(value, str) and bool(_EMAIL_RE.match(value)) + + +@app.route("/openapi.json") +def openapi(): + """OpenAPI spec with format: email on the contact endpoint.""" + return jsonify( + { + "openapi": "3.0.0", + "info": {"title": "Contact API", "version": "1.0.0"}, + "paths": { + "/contact": { + "post": { + "summary": "Submit a contact form", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "name", + "email", + "message", + ], + "properties": { + "name": { + "type": "string", + "description": "Full name", + }, + "email": { + "type": "string", + "format": "email", + "description": "Contact email address", + }, + "message": { + "type": "string", + "description": "Message body", + }, + }, + }, + "example": { + "name": "Alice", + "email": "alice@example.com", + "message": "Hello there!", + }, + } + }, + }, + "responses": { + "200": { + "description": "Contact form received", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Submission ID", + }, + "status": { + "type": "string", + "description": "Submission status", + }, + }, + }, + "example": { + "id": 1, + "status": "received", + }, + } + }, + }, + "400": { + "description": "Invalid email format", + "content": { + "application/json": { + "example": { + "error": "Invalid email format", + "field": "email", + } + } + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/contact", methods=["POST"]) +def submit_contact(): + """Accept contact form; validate email format strictly.""" + global _next_id + data = request.get_json(silent=True) or {} + + email = data.get("email", "") + if not _is_valid_email(email): + return ( + jsonify({"error": "Invalid email format", "field": "email"}), + 400, + ) + + result = {"id": _next_id, "status": "received"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + """Reset server state.""" + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_servers/hostname_format_validation/Dockerfile b/tests/test_servers/hostname_format_validation/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/hostname_format_validation/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/hostname_format_validation/server.py b/tests/test_servers/hostname_format_validation/server.py new file mode 100644 index 0000000..0616b2a --- /dev/null +++ b/tests/test_servers/hostname_format_validation/server.py @@ -0,0 +1,156 @@ +"""Mock server - validates hostname format in request body, returns descriptive 400.""" + +import re + +from flask import Flask, jsonify, request + +app = Flask(__name__) + +# Each label: 1-63 chars, alphanumeric or hyphen, must not start/end with hyphen. +# Full hostname: labels joined by '.', total ≤ 253 chars. +_LABEL_RE = re.compile(r"^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?$") + + +def _is_valid_hostname(value): + if not isinstance(value, str) or not value.strip(): + return False, "Hostname must be a non-empty string" + if len(value) > 253: + return ( + False, + f"Hostname exceeds maximum length of 253 characters (got {len(value)})", + ) + labels = value.rstrip(".").split(".") + if not labels: + return False, "Hostname must contain at least one label" + for label in labels: + if not label: + return ( + False, + "Hostname must not contain consecutive dots or a leading dot", + ) + if not _LABEL_RE.match(label): + return ( + False, + f"Label '{label}' is invalid: each label must be 1-63 alphanumeric characters " + "or hyphens and must not start or end with a hyphen", + ) + return True, None + + +@app.route("/openapi.json") +def openapi(): + return jsonify( + { + "openapi": "3.0.0", + "info": {"title": "Server Config API", "version": "1.0.0"}, + "paths": { + "/config": { + "post": { + "summary": "Configure a remote server hostname", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["server_hostname", "port"], + "properties": { + "server_hostname": { + "type": "string", + "format": "hostname", + "description": "Fully-qualified hostname of the server", + }, + "port": { + "type": "integer", + "description": "Port number", + "minimum": 1, + "maximum": 65535, + }, + }, + }, + "example": { + "server_hostname": "api.example.com", + "port": 443, + }, + } + }, + }, + "responses": { + "200": { + "description": "Server configured", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Config ID", + }, + "status": { + "type": "string", + "description": "Config status", + }, + }, + }, + "example": { + "id": 1, + "status": "configured", + }, + } + }, + }, + "400": { + "description": "Invalid hostname", + "content": { + "application/json": { + "example": { + "error": "Invalid hostname", + "field": "server_hostname", + "detail": "Each label must be 1-63 alphanumeric characters or hyphens", + } + } + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/config", methods=["POST"]) +def configure_server(): + global _next_id + data = request.get_json(silent=True) or {} + hostname = data.get("server_hostname", "") + valid, reason = _is_valid_hostname(hostname) + if not valid: + return ( + jsonify( + { + "error": "Invalid hostname", + "field": "server_hostname", + "detail": reason, + } + ), + 400, + ) + result = {"id": _next_id, "status": "configured"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_servers/ipv4_format_validation/Dockerfile b/tests/test_servers/ipv4_format_validation/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/ipv4_format_validation/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/ipv4_format_validation/server.py b/tests/test_servers/ipv4_format_validation/server.py new file mode 100644 index 0000000..f23b214 --- /dev/null +++ b/tests/test_servers/ipv4_format_validation/server.py @@ -0,0 +1,142 @@ +"""Mock server - validates IPv4 format in request body, returns descriptive 400.""" + +import re + +from flask import Flask, jsonify, request + +app = Flask(__name__) + +_IPV4_RE = re.compile(r"^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$") + + +def _is_valid_ipv4(value): + if not isinstance(value, str): + return False, "IPv4 address must be a string" + m = _IPV4_RE.match(value) + if not m: + return ( + False, + "IPv4 address must be four decimal octets separated by dots (e.g. 192.168.1.1)", + ) + octets = [int(g) for g in m.groups()] + if any(o > 255 for o in octets): + bad = [str(o) for o in octets if o > 255] + return False, f"Each octet must be 0-255; invalid: {', '.join(bad)}" + return True, None + + +@app.route("/openapi.json") +def openapi(): + return jsonify( + { + "openapi": "3.0.0", + "info": {"title": "Network API", "version": "1.0.0"}, + "paths": { + "/allow": { + "post": { + "summary": "Add an IP address to the allowlist", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["ip_address", "label"], + "properties": { + "ip_address": { + "type": "string", + "format": "ipv4", + "description": "IPv4 address to allowlist", + }, + "label": { + "type": "string", + "description": "Description of this IP", + }, + }, + }, + "example": { + "ip_address": "192.168.1.1", + "label": "Office network", + }, + } + }, + }, + "responses": { + "200": { + "description": "IP added to allowlist", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Rule ID", + }, + "status": { + "type": "string", + "description": "Rule status", + }, + }, + }, + "example": { + "id": 1, + "status": "allowed", + }, + } + }, + }, + "400": { + "description": "Invalid IPv4 address", + "content": { + "application/json": { + "example": { + "error": "Invalid IPv4 address", + "field": "ip_address", + "detail": "Each octet must be 0-255", + } + } + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/allow", methods=["POST"]) +def add_allowlist(): + global _next_id + data = request.get_json(silent=True) or {} + ip_address = data.get("ip_address", "") + valid, reason = _is_valid_ipv4(ip_address) + if not valid: + return ( + jsonify( + { + "error": "Invalid IPv4 address", + "field": "ip_address", + "detail": reason, + } + ), + 400, + ) + result = {"id": _next_id, "status": "allowed"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_servers/ipv6_format_validation/Dockerfile b/tests/test_servers/ipv6_format_validation/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/ipv6_format_validation/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/ipv6_format_validation/server.py b/tests/test_servers/ipv6_format_validation/server.py new file mode 100644 index 0000000..d7b45d2 --- /dev/null +++ b/tests/test_servers/ipv6_format_validation/server.py @@ -0,0 +1,138 @@ +"""Mock server - validates IPv6 format in request body, returns descriptive 400.""" + +import socket + +from flask import Flask, jsonify, request + +app = Flask(__name__) + + +def _is_valid_ipv6(value): + if not isinstance(value, str) or not value.strip(): + return False, "IPv6 address must be a non-empty string" + try: + socket.inet_pton(socket.AF_INET6, value) + return True, None + except (socket.error, OSError): + return ( + False, + f"'{value}' is not a valid IPv6 address; " + "expected format like '2001:db8::1' or '::1'", + ) + + +@app.route("/openapi.json") +def openapi(): + return jsonify( + { + "openapi": "3.0.0", + "info": {"title": "IPv6 Network API", "version": "1.0.0"}, + "paths": { + "/allow6": { + "post": { + "summary": "Add an IPv6 address to the allowlist", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["ipv6_address", "label"], + "properties": { + "ipv6_address": { + "type": "string", + "format": "ipv6", + "description": "IPv6 address to allowlist", + }, + "label": { + "type": "string", + "description": "Description of this address", + }, + }, + }, + "example": { + "ipv6_address": "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + "label": "Remote office", + }, + } + }, + }, + "responses": { + "200": { + "description": "IPv6 address added to allowlist", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Rule ID", + }, + "status": { + "type": "string", + "description": "Rule status", + }, + }, + }, + "example": { + "id": 1, + "status": "allowed", + }, + } + }, + }, + "400": { + "description": "Invalid IPv6 address", + "content": { + "application/json": { + "example": { + "error": "Invalid IPv6 address", + "field": "ipv6_address", + "detail": "expected format like '2001:db8::1' or '::1'", + } + } + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/allow6", methods=["POST"]) +def add_allowlist_ipv6(): + global _next_id + data = request.get_json(silent=True) or {} + ipv6_address = data.get("ipv6_address", "") + valid, reason = _is_valid_ipv6(ipv6_address) + if not valid: + return ( + jsonify( + { + "error": "Invalid IPv6 address", + "field": "ipv6_address", + "detail": reason, + } + ), + 400, + ) + result = {"id": _next_id, "status": "allowed"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_servers/uri_format_validation/Dockerfile b/tests/test_servers/uri_format_validation/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/uri_format_validation/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/uri_format_validation/server.py b/tests/test_servers/uri_format_validation/server.py new file mode 100644 index 0000000..1dc760a --- /dev/null +++ b/tests/test_servers/uri_format_validation/server.py @@ -0,0 +1,145 @@ +"""Mock server - validates URI format in request body, returns descriptive 400.""" + +from urllib.parse import urlparse + +from flask import Flask, jsonify, request + +app = Flask(__name__) + +_VALID_SCHEMES = {"http", "https", "ftp", "ftps"} + + +def _is_valid_uri(value): + if not isinstance(value, str) or not value.strip(): + return False, "URI must be a non-empty string" + try: + parsed = urlparse(value) + except Exception: + return False, "URI could not be parsed" + if not parsed.scheme: + return False, "URI must include a scheme (e.g. https://)" + if parsed.scheme not in _VALID_SCHEMES: + return ( + False, + f"URI scheme '{parsed.scheme}' is not supported; use one of {sorted(_VALID_SCHEMES)}", + ) + if not parsed.netloc: + return False, "URI must include a host" + return True, None + + +@app.route("/openapi.json") +def openapi(): + return jsonify( + { + "openapi": "3.0.0", + "info": {"title": "Webhook API", "version": "1.0.0"}, + "paths": { + "/webhook": { + "post": { + "summary": "Register a webhook callback", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["callback_url", "event"], + "properties": { + "callback_url": { + "type": "string", + "format": "uri", + "description": "Callback URL for webhook delivery", + }, + "event": { + "type": "string", + "description": "Event type to subscribe to", + }, + }, + }, + "example": { + "callback_url": "https://example.com/hook", + "event": "order.created", + }, + } + }, + }, + "responses": { + "200": { + "description": "Webhook registered", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Webhook ID", + }, + "status": { + "type": "string", + "description": "Registration status", + }, + }, + }, + "example": { + "id": 1, + "status": "registered", + }, + } + }, + }, + "400": { + "description": "Invalid URI format", + "content": { + "application/json": { + "example": { + "error": "Invalid URI format", + "field": "callback_url", + "detail": "URI must include a scheme (e.g. https://)", + } + } + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/webhook", methods=["POST"]) +def register_webhook(): + global _next_id + data = request.get_json(silent=True) or {} + callback_url = data.get("callback_url", "") + valid, reason = _is_valid_uri(callback_url) + if not valid: + return ( + jsonify( + { + "error": "Invalid URI format", + "field": "callback_url", + "detail": reason, + } + ), + 400, + ) + result = {"id": _next_id, "status": "registered"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_servers/url_port_validation/Dockerfile b/tests/test_servers/url_port_validation/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/url_port_validation/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/url_port_validation/server.py b/tests/test_servers/url_port_validation/server.py new file mode 100644 index 0000000..c3011aa --- /dev/null +++ b/tests/test_servers/url_port_validation/server.py @@ -0,0 +1,162 @@ +"""Mock server - validates URI format including port-number range. + +Port rules: + - No port in URL → valid (uses the scheme default) + - Port 1–65535 → valid + - Port 0 → invalid (reserved, not usable for HTTP) + - Port > 65535 → invalid (exceeds 16-bit range) +""" + +from urllib.parse import urlparse + +from flask import Flask, jsonify, request + +app = Flask(__name__) + +_VALID_SCHEMES = {"http", "https"} + + +def _validate_url_with_port(value): + """Return (is_valid, error_detail).""" + if not isinstance(value, str) or not value.strip(): + return False, "URL must be a non-empty string" + try: + parsed = urlparse(value) + except Exception: + return False, "URL could not be parsed" + if not parsed.scheme or parsed.scheme not in _VALID_SCHEMES: + return ( + False, + f"URL scheme must be http or https, got '{parsed.scheme or '(none)'}'", + ) + if not parsed.hostname: + return False, "URL must include a hostname" + if parsed.port is not None: + if parsed.port == 0: + return False, "Port 0 is reserved and not valid for HTTP URLs" + if parsed.port > 65535: + return ( + False, + f"Port {parsed.port} exceeds the maximum valid port number (65535)", + ) + return True, None + + +@app.route("/openapi.json") +def openapi(): + return jsonify( + { + "openapi": "3.0.0", + "info": {"title": "Webhook Port API", "version": "1.0.0"}, + "paths": { + "/webhook": { + "post": { + "summary": "Register a webhook; URL port must be in range 1-65535", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["callback_url", "event"], + "properties": { + "callback_url": { + "type": "string", + "format": "uri", + "description": ( + "Callback URL; port (if given) must be 1-65535. " + "Port 0 and ports above 65535 are rejected." + ), + }, + "event": { + "type": "string", + "description": "Event type", + }, + }, + }, + "example": { + "callback_url": "https://example.com:8080/hook", + "event": "order.created", + }, + } + }, + }, + "responses": { + "200": { + "description": "Webhook registered", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Webhook ID", + }, + "status": { + "type": "string", + "description": "Registration status", + }, + }, + }, + "example": { + "id": 1, + "status": "registered", + }, + } + }, + }, + "400": { + "description": "Invalid URL or port number", + "content": { + "application/json": { + "example": { + "error": "Invalid URL or port number", + "field": "callback_url", + "detail": "Port 0 is reserved and not valid for HTTP URLs", + } + } + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/webhook", methods=["POST"]) +def register_webhook(): + global _next_id + data = request.get_json(silent=True) or {} + callback_url = data.get("callback_url", "") + valid, reason = _validate_url_with_port(callback_url) + if not valid: + return ( + jsonify( + { + "error": "Invalid URL or port number", + "field": "callback_url", + "detail": reason, + } + ), + 400, + ) + result = {"id": _next_id, "status": "registered"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_servers/uuid_format_validation/Dockerfile b/tests/test_servers/uuid_format_validation/Dockerfile new file mode 100644 index 0000000..fd06797 --- /dev/null +++ b/tests/test_servers/uuid_format_validation/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install flask + +COPY server.py . + +CMD ["python", "server.py"] diff --git a/tests/test_servers/uuid_format_validation/server.py b/tests/test_servers/uuid_format_validation/server.py new file mode 100644 index 0000000..1c364ff --- /dev/null +++ b/tests/test_servers/uuid_format_validation/server.py @@ -0,0 +1,140 @@ +"""Mock server - validates UUID format in request body, returns descriptive 400.""" + +import re + +from flask import Flask, jsonify, request + +app = Flask(__name__) + +_UUID_RE = re.compile( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" +) + + +def _is_valid_uuid(value): + if not isinstance(value, str): + return False, "UUID must be a string" + if not _UUID_RE.match(value): + return ( + False, + "UUID must follow the pattern xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx " + "(8-4-4-4-12 hexadecimal characters separated by hyphens)", + ) + return True, None + + +@app.route("/openapi.json") +def openapi(): + return jsonify( + { + "openapi": "3.0.0", + "info": {"title": "Resource API", "version": "1.0.0"}, + "paths": { + "/resource": { + "post": { + "summary": "Link to an existing resource by UUID", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["resource_id", "label"], + "properties": { + "resource_id": { + "type": "string", + "format": "uuid", + "description": "UUID of the target resource", + }, + "label": { + "type": "string", + "description": "Human-readable label", + }, + }, + }, + "example": { + "resource_id": "550e8400-e29b-41d4-a716-446655440000", + "label": "My resource", + }, + } + }, + }, + "responses": { + "200": { + "description": "Resource linked", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "Link ID", + }, + "status": { + "type": "string", + "description": "Link status", + }, + }, + }, + "example": { + "id": 1, + "status": "linked", + }, + } + }, + }, + "400": { + "description": "Invalid UUID format", + "content": { + "application/json": { + "example": { + "error": "Invalid UUID format", + "field": "resource_id", + "detail": "UUID must follow the pattern xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + } + } + }, + }, + }, + } + } + }, + } + ) + + +_next_id = 1 + + +@app.route("/resource", methods=["POST"]) +def link_resource(): + global _next_id + data = request.get_json(silent=True) or {} + resource_id = data.get("resource_id", "") + valid, reason = _is_valid_uuid(resource_id) + if not valid: + return ( + jsonify( + { + "error": "Invalid UUID format", + "field": "resource_id", + "detail": reason, + } + ), + 400, + ) + result = {"id": _next_id, "status": "linked"} + _next_id += 1 + return jsonify(result), 200 + + +@app.route("/reset", methods=["POST"]) +def reset(): + global _next_id + _next_id = 1 + return jsonify({"status": "reset"}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/tests/test_unit.py b/tests/test_unit.py index ab6eb95..c645652 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -583,3 +583,124 @@ def test_post_endpoint_single_no_202_example_fails(): assert not success assert "200/201/202" in error + + +# --------------------------------------------------------------------------- +# email format validation tests +# --------------------------------------------------------------------------- + +KNOWN_VALID_EMAILS = [ + "alice@example.com", + "sinan+tester@ozel.com", + "user+tag@subdomain.example.co.uk", + "test.user@example.com", +] + +KNOWN_INVALID_EMAILS = [ + "sinan,ozel @somehere", # comma, space — the user's own example + "not-an-email", # no @ sign + "@nodomain", # missing local part + "noatsign.com", # no @ at all + "double@@at.com", # two @ signs + "spaces in@email.com", # space in local part + "missing.tld@", # no domain after @ +] + + +def test_email_format_generates_valid_addresses(): + """generate_string_test_cases for format: email produces at least one valid address.""" + from pytest_openapi.case_generator import generate_string_test_cases + + schema = {"type": "string", "format": "email"} + test_cases = generate_string_test_cases(schema) + + assert len(test_cases) > 0, "Expected at least one email test case" + for case in test_cases: + assert ( + "@" in case + ), f"Expected all generated values to look like emails, got: {case!r}" + + +def test_email_format_generates_invalid_addresses(): + """generate_string_test_cases for format: email produces at least one invalid address. + + The plugin must probe servers that declare format: email by also sending + syntactically broken addresses. Servers that validate properly should + return 400; the plugin treats that 400 as a passing contract test. + """ + from pytest_openapi.case_generator import generate_string_test_cases + + schema = {"type": "string", "format": "email"} + test_cases = generate_string_test_cases(schema) + + # There must be at least one invalid email among the generated cases + # (i.e. something that does NOT conform to a basic user@domain.tld shape) + import re + + basic_email_re = re.compile(r"^[^@\s,]+@[^@\s,]+\.[^@\s,]+$") + invalid_cases = [tc for tc in test_cases if not basic_email_re.match(tc)] + assert len(invalid_cases) >= 1, ( + f"Expected at least one invalid email test case to be generated, " + f"but all {len(test_cases)} cases look valid: {test_cases}" + ) + + +def test_email_format_invalid_cases_are_recognisably_broken(): + """The generated invalid email values must be clearly malformed. + + We verify that at least one of our documented broken patterns is + produced (no @ sign, spaces, commas, etc.). + """ + from pytest_openapi.case_generator import generate_string_test_cases + + schema = {"type": "string", "format": "email"} + test_cases = generate_string_test_cases(schema) + + broken_indicators = [" ", ",", "@@", "@."] + found_broken = any( + "@" not in tc or any(ind in tc for ind in broken_indicators) + for tc in test_cases + ) + assert found_broken, ( + f"Expected at least one clearly malformed email in generated cases, " + f"got: {test_cases}" + ) + + +def test_full_object_schema_with_email_format_generates_invalid_email(): + """When an object schema contains a format: email field, generated test cases + include at least one object where the email field is invalid. + + This mirrors what the plugin does for request bodies: it calls + generate_test_cases_for_schema on the full request schema and forwards + each generated object to the server. + """ + from pytest_openapi.case_generator import generate_test_cases_for_schema + + schema = { + "type": "object", + "required": ["name", "email", "message"], + "properties": { + "name": {"type": "string"}, + "email": {"type": "string", "format": "email"}, + "message": {"type": "string"}, + }, + } + test_cases, _ = generate_test_cases_for_schema(schema) + + assert len(test_cases) > 0, "Expected at least one object test case" + + import re + + basic_email_re = re.compile(r"^[^@\s,]+@[^@\s,]+\.[^@\s,]+$") + invalid_email_objects = [ + tc + for tc in test_cases + if isinstance(tc, dict) + and "email" in tc + and not basic_email_re.match(str(tc["email"])) + ] + assert len(invalid_email_objects) >= 1, ( + f"Expected at least one generated object with an invalid email, " + f"but all email values look valid: {[tc.get('email') for tc in test_cases if isinstance(tc, dict)]}" + )