en-milie
commented
5 months ago Hi @AbdulinRuslan. Currently, responses are not fully matched. It's only checking if any field in the response exists in the schema, but not the other way around. This is because responses can have a lot of variations, especially when using anyOf/oneOf combinations. This might be a feature that will be added in the future, but currently, indeed, is not supported.
Describe the bug No errors or warnings appear in case of difference between actual response body and response body described in OpenAPI contract.
To Reproduce Steps to reproduce the behaviour:
cats --contract=fastapi.json --server=http://127.0.0.1:8000 --httpMethods=GET --fuzzers=HappyPath --urlParams="user_id:2945"from fastapi import FastAPI, HTTPException, Request, status from pydantic import BaseModel, Field import json import psycopg2 from psycopg2.extras import RealDictCursor from psycopg2 import connect
app = FastAPI()
conn = connect(dbname='mydatabase', user='myuser', password='mypassword', host='localhost')
class User(BaseModel): name: str = Field(min_length=1, max_length=50) last_name: Optional[str] = Field(min_length=1, max_length=100)
@app.get("/users", response_model=list[User]) def get_users(): with conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute("SELECT * FROM Users;") users = cursor.fetchall() return users
@app.post("/users", response_model=User, status_code=status.HTTP_201_CREATED) def create_user(user: User): with open(f'data/create.txt', mode='a') as file: data = user.dict() file.write(f"Request POST /users with body: {json.dumps(data)}\n") with conn.cursor(cursor_factory=RealDictCursor) as cursor: try: cursor.execute( "INSERT INTO Users (name, last_name, age) VALUES (%s, %s, %s) RETURNING *;", (user.name, user.last_name, user.age) ) created_user = cursor.fetchone() conn.commit() return created_user except Exception as e: conn.rollback() raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
@app.get("/users/{user_id}", response_model=User) def get_user(user_id: int): with conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute("SELECT * FROM Users WHERE id = %s;", (user_id,)) user = cursor.fetchone() if user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") return user
@app.put("/users/{user_id}", response_model=User) def update_user(user_id: int, user: User): with open(f'data/put.txt', mode='a') as file: data = user.dict() file.write(f"Request PUT /users/{user_id} with body: {json.dumps(data)}\n") with conn.cursor(cursor_factory=RealDictCursor) as cursor: try: cursor.execute( """ UPDATE Users SET name = COALESCE(%s, name), last_name = COALESCE(%s, last_name), age = COALESCE(%s, age) WHERE id = %s RETURNING *; """, (user.name, user.last_name, user.age, user_id) ) updated_user = cursor.fetchone() conn.commit() if updated_user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") return updated_user except Exception as e: conn.rollback() raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_user(user_id: int): with conn.cursor(cursor_factory=RealDictCursor) as cursor: try: cursor.execute("DELETE FROM Users WHERE id = %s RETURNING id;", (user_id,)) deleted_user_id = cursor.fetchone() conn.commit() if deleted_user_id is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") except Exception as e: conn.rollback() raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))