From ed8a9d311b2d0a3c8f969d4bc0cf6543303f78f8 Mon Sep 17 00:00:00 2001 From: Alphabin Technology Consulting <83493276+alphabin-01@users.noreply.github.com> Date: Mon, 2 Jun 2025 20:03:38 +0530 Subject: [PATCH] Improve tests and add schema validation --- README.md | 35 +++++++ config/config_reader_api.py | 7 +- requirements.txt | 5 +- tests/test_jsonplaceholder.py | 168 ++++++++++++++++---------------- tests/test_products.py | 178 ++++++++++++++++++---------------- tests/test_reqres.py | 10 +- utils/assertions.py | 32 +++++- utils/request_handler.py | 35 +++++-- utils/schema_validator.py | 16 +++ utils/schemas.py | 52 ++++++++++ 10 files changed, 352 insertions(+), 186 deletions(-) create mode 100644 utils/schema_validator.py create mode 100644 utils/schemas.py diff --git a/README.md b/README.md index fb5c83b..4365259 100644 --- a/README.md +++ b/README.md @@ -1 +1,36 @@ # api-tests-python + +This repository contains a small collection of pytest tests used to +exercise various public APIs such as **reqres.in**, +**jsonplaceholder.typicode.com** and **fakestoreapi.com**. The tests +are intentionally simple and can be used as a starting point for API +automation examples. + +## Setup + +Create a virtual environment and install the dependencies: + +```bash +python -m venv venv +source venv/bin/activate +pip install -r requirements.txt +``` + +## Running the tests + +Execute all tests using `pytest` from the repository root: + +```bash +pytest -v +``` + +The tests make real HTTP requests to public APIs so internet access is +required. + +## Configuration + +The base URLs for the APIs reside in `config/config.json`. `RequestHandler` +reads these values to build the full request URLs. + +Sample payloads can be found under `data/` and are loaded within the +test modules when required. diff --git a/config/config_reader_api.py b/config/config_reader_api.py index e1f3551..e2eb89f 100644 --- a/config/config_reader_api.py +++ b/config/config_reader_api.py @@ -1,7 +1,12 @@ +"""Helper for reading API configuration files.""" + import json + def read_config(): + """Load configuration values from ``config.json``.""" with open("config/config.json", "r") as f: return json.load(f) -CONFIG = read_config() \ No newline at end of file + +CONFIG = read_config() diff --git a/requirements.txt b/requirements.txt index 5e40a10..85b7fe9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ -pytest -requests \ No newline at end of file +pytest +requests +jsonschema diff --git a/tests/test_jsonplaceholder.py b/tests/test_jsonplaceholder.py index c4701d8..80d0fa8 100644 --- a/tests/test_jsonplaceholder.py +++ b/tests/test_jsonplaceholder.py @@ -1,81 +1,87 @@ -import pytest -from utils.request_handler import RequestHandler -from utils.assertions import assert_status_code, assert_json_key -from utils import logger - -handler = RequestHandler(base_url_key="jsonplaceholder") -log = logger.log - - - -def test_get_todo_by_id(): - response = handler.get("/todos/1") - - log.info(f"GET /todos/1 → {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) - - for key in ["userId", "id", "title", "completed"]: - assert_json_key(response, key) - - -def test_get_all_posts(): - response = handler.get("/posts") - - log.info(f"GET /posts → {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) - - data = response.json() - assert isinstance(data, list), "Response is not a list" - if data: - for key in ["userId", "id", "title", "body"]: - assert key in data[0], f"Key '{key}' not found in first post" - - -def test_create_post(): - payload = { - "title": "foo", - "body": "bar", - "userId": 1 - } - response = handler.post("/posts", json_data=payload) - - log.info(f"POST /posts → {response.status_code}") - log.info(f"Payload: {payload}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 201) - # Check for expected keys in response - for key in ["id", "title", "body", "userId"]: - assert_json_key(response, key) - - -def test_update_post(): - payload = { - "id": 1, - "title": "foo", - "body": "bar", - "userId": 1 - } - response = handler.put("/posts/1", json_data=payload) - - log.info(f"PUT /posts/1 → {response.status_code}") - log.info(f"Payload: {payload}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) - # Check for expected keys in response - for key in ["id", "title", "body", "userId"]: - assert_json_key(response, key) - - -def test_delete_post(): - response = handler.delete("/posts/1") - - log.info(f"DELETE /posts/1 → {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) # Or 204 if that’s the expected status \ No newline at end of file +import pytest +from utils.request_handler import RequestHandler +from utils.assertions import assert_status_code, assert_json_key +from utils.schema_validator import validate_json_schema +from utils.schemas import TODO_SCHEMA, POST_SCHEMA +from utils import logger + +handler = RequestHandler(base_url_key="jsonplaceholder") +log = logger.log + + +def test_get_todo_by_id(): + """Validate that a single TODO item matches the expected schema.""" + response = handler.get("/todos/1") + + log.info(f"GET /todos/1 → {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) + validate_json_schema(response.json(), TODO_SCHEMA) + for key in ["userId", "id", "title", "completed"]: + assert_json_key(response, key) + + +def test_get_all_posts(): + """Ensure the posts endpoint returns a list of posts.""" + response = handler.get("/posts") + + log.info(f"GET /posts → {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) + + data = response.json() + assert isinstance(data, list), "Response is not a list" + if data: + for key in ["userId", "id", "title", "body"]: + assert key in data[0], f"Key '{key}' not found in first post" + + +def test_create_post(): + """Create a new post and validate the response schema.""" + payload = { + "title": "foo", + "body": "bar", + "userId": 1, + } + response = handler.post("/posts", json_data=payload) + + log.info(f"POST /posts → {response.status_code}") + log.info(f"Payload: {payload}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 201) + validate_json_schema(response.json(), POST_SCHEMA) + for key in ["id", "title", "body", "userId"]: + assert_json_key(response, key) + + +def test_update_post(): + """Update an existing post and verify the returned payload.""" + payload = { + "id": 1, + "title": "foo", + "body": "bar", + "userId": 1, + } + response = handler.put("/posts/1", json_data=payload) + + log.info(f"PUT /posts/1 → {response.status_code}") + log.info(f"Payload: {payload}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) + validate_json_schema(response.json(), POST_SCHEMA) + for key in ["id", "title", "body", "userId"]: + assert_json_key(response, key) + + +def test_delete_post(): + """Delete an existing post.""" + response = handler.delete("/posts/1") + + log.info(f"DELETE /posts/1 → {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) diff --git a/tests/test_products.py b/tests/test_products.py index f840f46..214be07 100644 --- a/tests/test_products.py +++ b/tests/test_products.py @@ -1,84 +1,94 @@ - -# get all products for fakestore - -from utils.request_handler import RequestHandler -from utils.assertions import assert_status_code, assert_json_key -from utils import logger -import pytest - -log = logger.log -handler = RequestHandler(base_url_key="fakestore_base_url") - - - -def test_create_product_fakestore(): - payload = { - "title": "New Product", - "price": 29.99 - } - - response = handler.post("/products", json_data=payload) - - log.info(f"Status Code: {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) - assert_json_key(response, "id") # fakestoreapi returns an 'id' in the response - - -def test_get_all_products_fakestore(): - response = handler.get("/products") - - log.info(f"Status Code: {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) - assert isinstance(response.json(), list), "Expected response to be a list of products" - assert "title" in response.json()[0], "Missing key 'title' in first product" - - - - -# update call for fakestore - -@pytest.mark.parametrize("product_id", [1, 2, 3]) -def test_get_single_product_by_id(product_id): - response = handler.get(f"/products/{product_id}") - - log.info(f"Status Code: {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) - assert_json_key(response, "title") - assert_json_key(response, "price") - - - -@pytest.mark.parametrize("product_id", [1]) -def test_update_product(product_id): - payload = { - "title": "Updated Product", - "price": 39.99 - } - - response = handler.put(f"/products/{product_id}", json_data=payload) - - log.info(f"Status Code: {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) - assert_json_key(response, "title") - assert_json_key(response, "price") - - - -#DELETE Call for fakestore - -@pytest.mark.parametrize("product_id", [1]) -def test_delete_product(product_id): - response = handler.delete(f"/products/{product_id}") - - log.info(f"Status Code: {response.status_code}") - log.info(f"Response Body: {response.text}") - - assert_status_code(response, 200) + +# get all products for fakestore + +import json +from utils.request_handler import RequestHandler +from utils.assertions import assert_status_code, assert_json_key +from utils.schema_validator import validate_json_schema +from utils.schemas import PRODUCT_SCHEMA +from utils import logger +import pytest + +log = logger.log +handler = RequestHandler(base_url_key="fakestore_base_url") +with open("data/test_data.json") as f: + SAMPLE_PRODUCT = json.load(f) + + + +def test_create_product_fakestore(): + """Create a product using the sample payload.""" + payload = SAMPLE_PRODUCT + + response = handler.post("/products", json_data=payload) + + log.info(f"Status Code: {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) + validate_json_schema(response.json(), PRODUCT_SCHEMA) + assert_json_key(response, "id") # fakestoreapi returns an 'id' + + +def test_get_all_products_fakestore(): + """Verify all products are returned.""" + response = handler.get("/products") + + log.info(f"Status Code: {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) + assert isinstance(response.json(), list), "Expected response to be a list of products" + assert "title" in response.json()[0], "Missing key 'title' in first product" + + + + +# update call for fakestore + +@pytest.mark.parametrize("product_id", [1, 2, 3]) +def test_get_single_product_by_id(product_id): + """Retrieve a specific product.""" + response = handler.get(f"/products/{product_id}") + + log.info(f"Status Code: {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) + validate_json_schema(response.json(), PRODUCT_SCHEMA) + assert_json_key(response, "title") + assert_json_key(response, "price") + + + +@pytest.mark.parametrize("product_id", [1]) +def test_update_product(product_id): + """Update an existing product.""" + payload = { + "title": "Updated Product", + "price": 39.99, + } + + response = handler.put(f"/products/{product_id}", json_data=payload) + + log.info(f"Status Code: {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) + validate_json_schema(response.json(), PRODUCT_SCHEMA) + assert_json_key(response, "title") + assert_json_key(response, "price") + + + +#DELETE Call for fakestore + +@pytest.mark.parametrize("product_id", [1]) +def test_delete_product(product_id): + """Delete a product from the store.""" + response = handler.delete(f"/products/{product_id}") + + log.info(f"Status Code: {response.status_code}") + log.info(f"Response Body: {response.text}") + + assert_status_code(response, 200) diff --git a/tests/test_reqres.py b/tests/test_reqres.py index 34347ca..2d7a87f 100644 --- a/tests/test_reqres.py +++ b/tests/test_reqres.py @@ -1,7 +1,8 @@ import pytest -import requests from utils.request_handler import RequestHandler from utils.assertions import assert_status_code, assert_json_key +from utils.schema_validator import validate_json_schema +from utils.schemas import USER_LIST_SCHEMA, UPDATE_USER_SCHEMA from utils import logger handler = RequestHandler() @@ -10,20 +11,23 @@ def test_get_users_page_2(): + """Retrieve the second page of users and validate schema.""" log.info("Test Started") response = handler.get("/users", params={"page": 2}) log.info(f"Response: {response.status_code} | Body: {response.text}") assert_status_code(response, 200) assert_json_key(response, "data") + validate_json_schema(response.json(), USER_LIST_SCHEMA) @pytest.mark.parametrize("user_id", [2, 5]) def test_put_user(user_id): + """Update a user and validate the response schema.""" payload = { "name": "Updated User", - "job": "QA Engineer" + "job": "QA Engineer", } response = handler.put(f"/users/{user_id}", data=payload) - print(response) assert_status_code(response, 200) assert_json_key(response, "job") + validate_json_schema(response.json(), UPDATE_USER_SCHEMA) diff --git a/utils/assertions.py b/utils/assertions.py index 2569608..8feae16 100644 --- a/utils/assertions.py +++ b/utils/assertions.py @@ -1,14 +1,36 @@ -# utils/assertions.py +"""Assertion helpers for API tests.""" def assert_status_code(response, expected_code): + """Assert that a response has the expected HTTP status code. - assert response.status_code == expected_code - print(f"Expected {expected_code}, got {response.status_code}") + Parameters + ---------- + response : :class:`requests.Response` + Response object returned by the request. + expected_code : int + Expected HTTP status code. + """ + + actual = response.status_code + assert actual == expected_code, ( + f"Expected status code {expected_code}, got {actual}" + ) -# def assert_json_key(response, key): -# assert key in response.json(), f"Key '{key}' not found in response" def assert_json_key(response, key, index=None): + """Assert that a key exists in the JSON body of the response. + + Parameters + ---------- + response : :class:`requests.Response` + Response object from the API request. + key : str + JSON key expected to be present. + index : int, optional + When the response body is a list, index specifies which item to + inspect. + """ + json_data = response.json() if index is not None: assert key in json_data[index], f"Key '{key}' not in item {index}" diff --git a/utils/request_handler.py b/utils/request_handler.py index 8a17c5d..d1bf591 100644 --- a/utils/request_handler.py +++ b/utils/request_handler.py @@ -3,39 +3,54 @@ class RequestHandler: - def __init__(self, base_url_key="reqres_base_url"): + """Simple wrapper around :mod:`requests` for making API calls.""" + + def __init__(self, base_url_key: str = "reqres_base_url"): + """Create a new request handler. + + Parameters + ---------- + base_url_key : str, optional + Key inside ``config.json`` that contains the base URL of the API. + """ + self.base_url = CONFIG[base_url_key] self.session = requests.Session() self.session.headers.update({"accept": "application/json"}) - def _full_url(self, endpoint): + def _full_url(self, endpoint: str) -> str: + """Return the absolute URL for a given endpoint.""" return f"{self.base_url}{endpoint}" - def get(self, endpoint, params=None, headers=None): + def get(self, endpoint: str, params=None, headers=None): + """Send a ``GET`` request.""" return self.session.get( self._full_url(endpoint), params=params, - headers=headers or self.session.headers + headers=headers or self.session.headers, ) - def post(self, endpoint, data=None, json_data=None, headers=None): + def post(self, endpoint: str, data=None, json_data=None, headers=None): + """Send a ``POST`` request.""" return self.session.post( self._full_url(endpoint), data=data, json=json_data, - headers=headers or self.session.headers + headers=headers or self.session.headers, ) - def put(self, endpoint, data=None, json_data=None, headers=None): + def put(self, endpoint: str, data=None, json_data=None, headers=None): + """Send a ``PUT`` request.""" return self.session.put( self._full_url(endpoint), data=data, json=json_data, - headers=headers or self.session.headers + headers=headers or self.session.headers, ) - def delete(self, endpoint, headers=None): + def delete(self, endpoint: str, headers=None): + """Send a ``DELETE`` request.""" return self.session.delete( self._full_url(endpoint), - headers=headers or self.session.headers + headers=headers or self.session.headers, ) \ No newline at end of file diff --git a/utils/schema_validator.py b/utils/schema_validator.py new file mode 100644 index 0000000..5a8e997 --- /dev/null +++ b/utils/schema_validator.py @@ -0,0 +1,16 @@ +"""Utilities for validating JSON responses against predefined schemas.""" + +from jsonschema import validate + + +def validate_json_schema(instance, schema): + """Validate a JSON object against a schema. + + Parameters + ---------- + instance : dict + JSON data returned from the API. + schema : dict + JSON schema to validate against. + """ + validate(instance=instance, schema=schema) diff --git a/utils/schemas.py b/utils/schemas.py new file mode 100644 index 0000000..44b92d2 --- /dev/null +++ b/utils/schemas.py @@ -0,0 +1,52 @@ +"""JSON Schemas used for validating API responses.""" + +TODO_SCHEMA = { + "type": "object", + "properties": { + "userId": {"type": "integer"}, + "id": {"type": "integer"}, + "title": {"type": "string"}, + "completed": {"type": "boolean"} + }, + "required": ["userId", "id", "title", "completed"] +} + +POST_SCHEMA = { + "type": "object", + "properties": { + "userId": {"type": "integer"}, + "id": {"type": "integer"}, + "title": {"type": "string"}, + "body": {"type": "string"} + }, + "required": ["userId", "id", "title", "body"] +} + +# Schemas for reqres.in API +USER_LIST_SCHEMA = { + "type": "object", + "properties": { + "page": {"type": "integer"}, + "data": {"type": "array"}, + }, + "required": ["page", "data"], +} + +UPDATE_USER_SCHEMA = { + "type": "object", + "properties": { + "name": {"type": "string"}, + "job": {"type": "string"}, + }, + "required": ["name", "job"], +} + +PRODUCT_SCHEMA = { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "title": {"type": "string"}, + "price": {"type": ["number", "string"]}, + }, + "required": ["id", "title", "price"], +}